communicator.java

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

package com.ict.netcom2.trash;

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

public class Communicator {
	
	public String ip;
	Socket skt161;
	Socket skt9090;
	DataInputStream dis161;
	DataInputStream dis9090;
	PrintStream ps161;
	PrintStream ps9090;
	MessageDecoder dec161;
	MessageDecoder dec9090;
	MessageEncoder enc = new MessageEncoder();
	
	public Communicator(String ip) {
		this.ip = ip;
		try {
			skt161 = new Socket(InetAddress.getByName(ip), 161);
			dis161 = new DataInputStream(new BufferedInputStream(
					skt161.getInputStream()));
			dec161 = new MessageDecoder(dis161);
			ps161 = new PrintStream(skt161.getOutputStream());
						
			skt9090 = new Socket(InetAddress.getByName(ip), 9090);
			dis9090 = new DataInputStream(new BufferedInputStream(
					skt9090.getInputStream()));
			dec9090 = new MessageDecoder(dis9090);
			ps9090 = new PrintStream(skt9090.getOutputStream());
			
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public void close() {
		ps161.close();
		ps9090.close();
		try {
			dis161.close();
			dis9090.close();
			skt161.close();
			skt9090.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}
	
	public int connect(int version, String key) {
		try {
			ps9090.write(enc.encodeConnect(version, key));
			ps9090.flush();			
		} catch (IOException e) {
			e.printStackTrace();
		}		
		return dec9090.decodeConnectAck();
	}
	
	public void disconnect() {
		try {
			ps9090.write(enc.encodeDisconnect());
			ps9090.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		dec9090.decodeDisconnectAck();
	}
	
	/**
	 * Register the netcom specified by "netcomIp".
	 * When netpro starts, it trys to connect the registered netcom.
	 * NetPro会给所有注册的NetCom发送状态变迁消息和事件告警通知.
	 * @param netproId
	 * @param netcomIp
	 * @return
	 */
	public int register(int netproId, String netcomIp) {
		try {
			ps9090.write(enc.encodeRegister(netproId, netcomIp));
			ps9090.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		return dec9090.decodeRegisterAck();
	}
	
	public void unRegister(String netcomIp) {
		try {
			ps9090.write(enc.encodeUnRegister(netcomIp));
			ps9090.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		dec9090.docodeUnRegisterAck();
	}
	
	public boolean checkLiving() {
		try {
			ps9090.write(enc.encodeLivingCheck());
			ps9090.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		return dec9090.decodeLivingCheckAck();		
	}
	
	public int setComm(String old, String newComm) {
		try {
			ps9090.write(enc.encodeSetComm(old, newComm));
			ps9090.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		return dec9090.decodeSetKey();
		
	}
	
	public SysInfo querySysInfo() {
		try {
			ps9090.write(enc.encodeQuerySysInfo());
			ps9090.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		SysInfo info = dec9090.decodeQuerySysInfoRep();
		System.out.println(info);
		return info;
	}
	
	/**
	 * Query task list of NetPro
	 * @return
	 * QueryTaskRep : 查询到的任务信息 
	 * @see QueryTaskRep
	 */
	public QueryTaskRep[] queryTask() {
		QueryTaskRep[] rep = new QueryTaskRep[2];
		try {
			// for READY list...can't use 0
			ps9090.write(enc.encodeQueryTask(1));
			ps9090.flush();
			rep[0] = dec9090.decodeQueryTaskRep();

			// for RUNNING list...
			ps9090.write(enc.encodeQueryTask(2));
			ps9090.flush();
			rep[1] = dec9090.decodeQueryTaskRep();
		} catch(IOException e) {
			e.printStackTrace();
		}
		return rep;
	}
	
	/**
	 * 
	 * @param type
	 * Only 1 and 2 is available.<br>
	 * 1 for READY list, 2 for RUNNING list.
	 * @return
	 */
	public QueryTaskRep queryTask(int type) {
		QueryTaskRep rep = null;
		try {
			ps9090.write(enc.encodeQueryTask(type));
			ps9090.flush();
			rep = dec9090.decodeQueryTaskRep();
		} catch(IOException e) {
			e.printStackTrace();
		}
		return rep;
	}
	
	public QueryThreadRep queryThreads() {
		try {
			ps9090.write(enc.encodeQueryThreads());
			ps9090.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		return dec9090.decodeQueryThreadRep();		
		
	}
	
	public int[] addTask(Task task) {
		try {
			ps9090.write(enc.encodeAddTask(task));
			ps9090.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		return dec9090.decodeAddTaskAck();		
	}
	
	public void stopTask(int taskId) {
		try {
			ps9090.write(enc.encodeTaskStop(taskId));
			ps9090.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		dec9090.decodeStopTaskAck();
		
	}
	
	public GetResultAck getResult(Task task) {
		try {
			ps161.write(enc.encodeGetResult(task));
			ps161.flush();
		} catch(IOException e) {
			e.printStackTrace();
		}
		GetResultAck ack = dec161.decodeGetResultAck(task);
		ack.result.setProbeIp(ip);
		ack.result.setTaskId(task.getId());
		return ack;
	}
	

	public void notifyTaskStatus(int netproId, int taskId,
			int statusInfo) {
		
	}
	
	public void notifyEvent(int netproId, int priority,
			int eventCode, String value){
		
	}
	
	
	public static void main(String[] str) {
		TaskManager tm = new TaskManager();
		Communicator ck = new Communicator("10.21.2.30");
		//ck.connect(4, "justatest");
		ck.querySysInfo();
		ck.register(1, "10.21.2.48");

		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.close();
		
		//********************** 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();		
		/*
		try {
			Socket skt = new Socket(InetAddress.getByName("10.21.2.30"), 161);
			DataInputStream dis = new DataInputStream(skt.getInputStream());
			DataOutputStream dos = new DataOutputStream(
					new BufferedOutputStream(skt.getOutputStream()));
			int netproId = 1;
			int version = Integer.parseInt("02000000", 16);
			InetAddress ia = InetAddress.getByName("::");
			byte[] b = ia.getAddress();
			System.out.println(b.length);
			dos.writeInt(24);
			dos.writeInt(15);
			dos.writeInt(10);
			dos.writeInt(netproId);
			dos.writeInt(version);
			dos.write(b);
			dos.flush();
			dos.close();
			skt.close();
			
		} catch (IOException e) {
			e.printStackTrace();
		}
		*/
		
	}
}