communicator2.java

利用Java Socket写的一段通讯协议

package com.ict.netcom2.net;

import java.io.*;
import java.net.*;
import com.ict.netcom2.task.*;
import com.ict.netcom2.hardware.*;
import com.ict.netcom2.message.*;

public class Communicator2 {
	
	public String ip;
	
	private int version;
	private String comm;	
	private Socket skt;
	private DataInputStream in;
	private PrintStream out;
	private MessageEncoder enc = new MessageEncoder();
	
	/**
	 * Initialize a communicator using the
	 * specified ip,version and community.
	 * @param ip
	 * The probe's ip
	 * @param version
	 * 双方协商的版本号
	 * @param comm
	 * 由NetCom指定的团体字
	 */
	public Communicator2(String ip, int version, String comm) {
		this.ip = ip;
		this.version = version;
		this.comm = comm;
	}
	
	/**
	 * Initialize a socket connect using the specified port.<br>
	 * Then send the connect message to probe and get the ack.
	 * @param port 9090 or 161
	 * @return int : Connect ack<br>
	 * 0：请求连接成功<br>
	 * 1：消息结构错<br>
	 * 2：版本不匹配<br>
	 * 3：团体字不匹配<br>
	 * 4：其他错误
	 */
	private int connect(int port) {
		int ret = -1;
		try {			
			skt = new Socket(InetAddress.getByName(ip), port);
			in = new DataInputStream(new BufferedInputStream(
					skt.getInputStream()));
			out = new PrintStream(skt.getOutputStream());

			out.write(enc.encodeConnect(version, comm));
			out.flush();			
			ret = MessageDecoder2.decodeConnectAck(read());			
			if (ret != 0) {
				//TODO status handle
				System.err.println("Connect error.");
				disconnect();
			}			
		} catch (IOException e) {
			e.printStackTrace();
		}
		return ret;
	}
	
	/**
	 * Send disconnect message to porbe,
	 * then close the socket.
	 */
	private void disconnect() {
		try {
			out.write(enc.encodeDisconnect());
			out.flush();
			read();
			
			out.close();
			in.close();
			skt.close();
		} catch(IOException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * Read a message from the opening inputstream.
	 * @return byte[] : The message body.
	 */
	private byte[] read() {
		byte[] b = null;
		try {
			int len = in.readInt();
			in.readInt();
			in.readInt();
			
			if (len > 12) {
				b = new byte[len-12];
				in.read(b);
			}
			//TODO long length buffer MyByteBuffer
			
		} catch (IOException e) {
			e.printStackTrace();
		}
		return b;
	}
	
	/**
	 * Register the netcom specified by "netcomIp".
	 * When netpro starts, it trys to connect the registered netcom.<br>
	 * NetPro会给所有注册的NetCom发送状态变迁消息和事件告警通知.
	 * @param netproId
	 * @param netcomIp
	 * @return int : Register ack.<br>
	 * 0：注册成功<br>1：错误
	 */
	public int register(int netproId, String netcomIp) {
		if (connect(9090) != 0) {
			return -1;
		}		
		try {
			out.write(enc.encodeRegister(netproId, netcomIp));
			out.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		byte[] b = read();
		disconnect();
		return MessageDecoder2.decodeRegisterAck(b);
	}
	
	public void unRegister(String netcomIp) {
		if (connect(9090) != 0) {
			return;
		}
		try {
			out.write(enc.encodeUnRegister(netcomIp));
			out.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		read();
		disconnect();
	}
	
	public boolean checkLiving() {
		if (connect(9090) != 0) {
			return false;
		}
		try {
			out.write(enc.encodeLivingCheck());
			out.flush();
		} catch(IOException e) {
			e.printStackTrace();
			return false;
		}
		read();
		disconnect();
		return true;		
	}
	
	/**
	 * Set community.
	 * @param old
	 * Old community.
	 * @param newComm
	 * New community.
	 * @return int : Status<br>
	 * 0：设置团体字成功<br>
	 * 1：旧团体字不匹配，不成功<br>
	 * 2 ：其他错误
	 */
	public int setComm(String old, String newComm) {
		if (connect(9090) != 0) {
			return -1;
		}
		try {
			out.write(enc.encodeSetComm(old, newComm));
			out.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		byte[] b = read();
		disconnect();
		return MessageDecoder2.decodeSetKey(b);
	}
	
	public SysInfo querySysInfo() {
		if (connect(9090) != 0) {
			return null;
		}
		try {
			out.write(enc.encodeQuerySysInfo());
			out.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		byte[] b = read();
		disconnect();
		return MessageDecoder2.decodeQuerySysInfoRep(b);
	}
	
	/**
	 * Query task lists of NetPro.
	 * @return
	 * QueryTaskRep : 查询到的任务信息 
	 * @see QueryTaskRep
	 */
	public QueryTaskRep[] queryTask() {
		if (connect(9090) != 0) {
			return null;
		}
		QueryTaskRep[] rep = new QueryTaskRep[2];
		try {
			// for READY list...can't use 0
			out.write(enc.encodeQueryTask(1));
			out.flush();
			rep[0] = MessageDecoder2.decodeQueryTaskRep(read());

			// for RUNNING list...
			out.write(enc.encodeQueryTask(2));
			out.flush();
			rep[1] = MessageDecoder2.decodeQueryTaskRep(read());
		} catch(IOException e) {
			e.printStackTrace();
		}
		disconnect();
		return rep;
	}
	
	/**
	 * Query task lists of NetPro.
	 * @param type
	 * Only 1 and 2 is available.<br>
	 * 1 for READY list, 2 for RUNNING list.
	 * @return
	 */
	public QueryTaskRep queryTask(int type) {
		if (connect(9090) != 0) {
			return null;
		}
		QueryTaskRep rep = null;
		try {
			out.write(enc.encodeQueryTask(type));
			out.flush();
			rep = MessageDecoder2.decodeQueryTaskRep(read());
		} catch(IOException e) {
			e.printStackTrace();
		}
		disconnect();
		return rep;
	}
	
	public QueryThreadRep queryThreads() {
		if (connect(9090) != 0) {
			return null;
		}
		try {
			out.write(enc.encodeQueryThreads());
			out.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		byte[] b = read();
		disconnect();
		return MessageDecoder2.decodeQueryThreadRep(b);
	}
	
	public int[] addTask(Task task) {
		if (connect(9090) != 0) {
			return null;
		}
		try {
			out.write(enc.encodeAddTask(task));
			out.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		byte[] b = read();
		disconnect();
		return MessageDecoder2.decodeAddTaskAck(b);		
	}
	
	public int[] stopTask(int taskId) {
		if (connect(9090) != 0) {
			return null;
		}
		try {
			out.write(enc.encodeTaskStop(taskId));
			out.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		byte[] b = read();
		disconnect();
		return MessageDecoder2.decodeStopTaskAck(b);
		
	}
	
	public GetResultAck getResult(Task task) {
		if (connect(161) != 0) {
			return null;
		}
		try {
			out.write(enc.encodeGetResult(task));
			out.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		byte[] b = read();
		disconnect();
		GetResultAck ack = MessageDecoder2.decodeGetResultAck(b, task);
		if (ack.status == 0) {
			ack.result.setProbeIp(ip);
			ack.result.setTaskId(task.getId());
		}
		return ack;
	}
	
	public static void main(String[] str) {
		TaskManager tm = new TaskManager();
		Communicator2 ck = new Communicator2("10.21.2.29", 4, "justatest");
		//ck.connect(4, "justatest");
		ck.register(1, "10.21.2.48");
		ck.querySysInfo();

		HttpPerfTaskParam hptParam = new HttpPerfTaskParam(
				2, 80, 10, 2, "www.baidu.com");
		Task httpTask = tm.createTask(10, false, hptParam);
		
		PasvEthTaskParam petParam = new PasvEthTaskParam(
				new Card(1,0), new Card(1,0), 
				2, 20, "0:1:2:3:4:5");
		Task ethTask = tm.createTask(0, false, petParam);
		
		//ck.addTask(ethTask);
		//ck.addTask(httpTask);
		//ck.stopTask(48000);
		//ck.stopTask(48001);
		ck.queryTask();
		ck.getResult(httpTask);
		
		//********************** these message maybe done
		//ck.connect(4, "justatest");
		//ck.disconnect();
		//ck.setKey("new", "justatest");
		//ck.register(1, "10.21.2.48");
		//ck.checkLiving();
		//ck.querySysInfo();
		//ck.queryTask();  //taskId == -1 cause exception
		//ck.queryThreads();
		//ck.stopTask(1);
		//byte[] task = TaskEncoder.encodeHttpRefTask(30, 80, 60, 2, "www.google.com");
		//ck.addTask(13,10, false, TaskType.HTTP_PERF, task);
		//ck.getResult(false, 13);
		//****************************************
		
		
		//************** with fucking problems
		//ck.unRegister("10.21.2.49"); //it will unregister all ...
		
		//byte[] task = TaskEncoder.encodePasvEth();
		//ck.addTask(0, false, TaskType.PASSIVE_ETH, task);
		//****************************************
		
		//ck.addTask(httpTask);

		//byte[] task = TaskEncoder.encodePasvEth();
		//ck.addTask(4, 0, false, TaskType.PASSIVE_ETH, task);
		//ck.queryThreads();
		//ck.stopTask(1345);

	
		//ck.stopTask(48000);
		
		//ck.getResult(httpTask);
		//ck.getResult(ethTask);
		//ck.disconnect();		

		
	}
}