packetlogger.java

基于Jpcap开发的流量监控软件

package joel.ethernettools;
/**
 * THIS CLASS IS USED TO FETCH OriginalPacket/ProcessedPacket FROM PacketPool,
 * AND SAVE TO DATABASE.
 * @author panzer
 *
 */
public class PacketLogger implements Runnable {
	private PacketPool pool;
	private MySQLOperator operator;
	private boolean enableOriginalPacketLog, enableProcessedPacketLog;
	private OriginalPacket[] original;
	private ProcessedPacket[] processed;
	private int Capcity = 10;
	private MsgProvider provider;
	private int totalReceivedOriPacket = 0, totalReceivedProPacket = 0,
			totalSavedOriPacket = 0, totalSavedPropacket = 0;
	private int currentOriCapcity = 0, currentProCapcity = 0;
	private boolean saveFlag=false;
	
	protected int getTotalReceivedOriPacket() {
		return totalReceivedOriPacket;
	}

	protected int getTotalReceivedProPacket() {
		return totalReceivedProPacket;
	}

	protected int getCurrentOriCapcity() {
		return currentOriCapcity;
	}

	protected int getCurrentProCapcity() {
		return currentProCapcity;
	}

	protected void setEnableOriginalPacketLog(boolean enableOriginalPacketLog) {
		this.enableOriginalPacketLog = enableOriginalPacketLog;
	}

	protected void setEnableProcessedPacketLog(boolean enableProcessedPacketLog) {
		this.enableProcessedPacketLog = enableProcessedPacketLog;
	}

	protected void setMsgReceiver(MsgReceiver m) {
		provider.addMsgReceiver(m);
	}


	public PacketLogger() {
		provider = new MsgProvider();
		operator = new MySQLOperator();
		original = new OriginalPacket[Capcity];
		processed = new ProcessedPacket[Capcity];
	}
//	protected void setMySQLOperator(MySQLOperator o){
//		this.operator=o;
//	}
	public PacketLogger(boolean enableOriginalPacketLog,
			boolean enableProcessedPacketLog) {
		this();
		this.enableOriginalPacketLog = enableOriginalPacketLog;
		this.enableProcessedPacketLog = enableProcessedPacketLog;
	}

	protected void setPacketPool(PacketPool pool) {
		this.pool = pool;
	}

	protected void logOriginalPacket(OriginalPacket[] p) {
		if (enableOriginalPacketLog == false)
			return;
		saveFlag=operator.add(p);
//		if(saveFlag==true){
		provider.addNewMsg("saved OriginalPacket.");
		totalSavedOriPacket += p.length;
		p=null;
//		}
	}

	protected void logProcessedPacket(ProcessedPacket[] p) {
		if (enableProcessedPacketLog == false)
			return;
		saveFlag=operator.add(p);
//		if(saveFlag==true){
		provider.addNewMsg("saved ProcessedPacket.");
		totalSavedPropacket += p.length;
		p=null;
//		}
	}

	protected void setSaveEveryManyPackets(int i){
		this.Capcity=i;
		original = new OriginalPacket[Capcity];
		processed = new ProcessedPacket[Capcity];
	}
	/**
	 * @param args
	 */
	public static void main(String[] args) {

	}

	@Override
	public void run() {
		boolean flag=false;
		provider.addNewMsg("log thread: " + Thread.currentThread().getName()
				+ " start.");
		while (true) {
			if ((original[currentOriCapcity] = pool.getNewOriginalPacket()) != null) {
				totalReceivedOriPacket++;
				if (currentOriCapcity == Capcity - 1) {
					this.logOriginalPacket(original);
					currentOriCapcity = 0;
				} else {
					currentOriCapcity++;					
				}
			} else if ((processed[currentProCapcity] = pool
					.getNewProcessedPacket()) != null) {
				totalReceivedProPacket++;
				if (currentProCapcity == Capcity - 1) {
					this.logProcessedPacket(processed);
					currentProCapcity = 0;
				} else {
					currentProCapcity++;					
				}
			} else
				try {
					Thread.currentThread().sleep(
							Math.round(Math.random() * 1000 * 5));
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
		}
	}

	protected int getTotalSavedOriPacket() {
		return totalSavedOriPacket;
	}

	protected int getTotalSavedPropacket() {
		return totalSavedPropacket;
	}
}