自定义的投票协议的通讯(tcp/ip 与 udp示例)

VoteMsg.java

package com.tcpip;

/**
* 投票信息
* @author Administrator
*
*/
public class VoteMsg {
//请求
private boolean isInquiry ;
//回应
private boolean isResponse;
//投票ID
private int candidateID;
//投票数
private long voteCount;

public static final int MAX_CANDIDATE_ID=1000;

public VoteMsg(boolean isResponse,boolean isInquiry,int candidateID,long voteCount) throws IllegalArgumentException{
if(voteCount!=0 && !isResponse){
throw new IllegalArgumentException("Requset vote count must be zero");
}
if(candidateID<0 || candidateID>MAX_CANDIDATE_ID){
throw new IllegalArgumentException("Bad Candidate ID:"+candidateID);
}
if(voteCount<0){
throw new IllegalArgumentException("Total must be >= zero");
}
this.candidateID = candidateID;
this.isResponse = isResponse;
this.isInquiry = isInquiry;
this.voteCount = voteCount;
}

public boolean isInquiry() {
return isInquiry;
}

public void setInquiry(boolean isInquiry) {
this.isInquiry = isInquiry;
}

public boolean isResponse() {
return isResponse;
}

public void setResponse(boolean isResponse) {
this.isResponse = isResponse;
}

public int getCandidateID() {
return candidateID;
}

public void setCandidateID(int candidateID) throws IllegalArgumentException{
if(candidateID<0 || candidateID>MAX_CANDIDATE_ID){
throw new IllegalArgumentException("Bad Candidate ID:"+candidateID);
}
this.candidateID = candidateID;
}

public long getVoteCount() {
return voteCount;
}

public void setVoteCount(long voteCount) {
if((voteCount!=0&&!isResponse) || voteCount<0){
throw new IllegalArgumentException("Bad vote count");
}
this.voteCount = voteCount;
}

public String toString(){
String res = (isInquiry ? "inquiry" : "vote")+" for candidate "+candidateID;
if(isResponse){
res = "response to" + res +" who now has " + voteCount+ " vote(s)";
}
return res;
}

}

接口类：VoteMsgCoder.java

package com.tcpip;

import java.io.IOException;

/**
* 投票消息处理(编码与解码)
* @author Administrator
*
*/
public interface VoteMsgCoder {
/**
* 将投票消息转换为一个字节序列(根据特定的协议)
* @param msg
* @return
* @throws IOException
*/
byte[] toWire(VoteMsg msg) throws IOException;

/**
* 对给定的字节序列进行解析(根据特定的协议)
* @param input
* @return
* @throws IOException
*/
VoteMsg fromWire(byte[] input) throws IOException;
}

实现类：VoteMsgTextCoder.java:

package com.tcpip;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.Scanner;

/**
* 用文本方式对消息进行编码
* @author Administrator
*
*/
public class VoteMsgTextCoder implements VoteMsgCoder {

public static final String MAGIC = "Voting";
public static final String VOTESTR = "v";
public static final String INQSTR = "i";
public static final String RESPONSESTR = "R";
public static final String CHARSETNAME = "US-ASCII";
public static final String DELIMSTR = " ";
public static final int MAX_WIRE_LENGTH = 2000;

/*
* 将投票消息转换为一个字节序列(根据特定的协议)
* @see com.tcpip.VoteMsgCoder#toWire(com.tcpip.VoteMsg)
*/
@Override
public byte[] toWire(VoteMsg msg) throws IOException {
String msgString = MAGIC + DELIMSTR + (msg.isInquiry() ? INQSTR : VOTESTR)
+ DELIMSTR + (msg.isResponse() ? RESPONSESTR + DELIMSTR : "")
+ Integer.toString(msg.getCandidateID()) + DELIMSTR
+ Long.toString(msg.getVoteCount());

byte[]  data = msgString.getBytes(CHARSETNAME);
return data;
}

/*
* 对给定的字节序列进行解析(根据特定的协议)
* @see com.tcpip.VoteMsgCoder#fromWire(byte[])
*/
@Override
public VoteMsg fromWire(byte[] message) throws IOException {
ByteArrayInputStream msgStream = new ByteArrayInputStream(message);
Scanner s = new Scanner(new InputStreamReader(msgStream,CHARSETNAME));

boolean isInquiry;
boolean isResponse;
int candidateID;
long voteCount;
String token;

try{
token = s.next();
if(!token.equals(MAGIC)){
throw new IOException("Bad magic string:"+token);
}
token = s.next();
if(token.equals(VOTESTR)){
isInquiry = false;
}else if(!token.equals(INQSTR)){
throw new IOException("Bad vote/inq indicator: "+token);
}else {
isInquiry = true;
}
token = s.next();
if(token.equals(RESPONSESTR)){
isResponse = true;
token = s.next();
}else{
isResponse = false;
}

candidateID = Integer.parseInt(token);
if(isResponse){
token  = s.next();
voteCount = Long.parseLong(token);
}else{
voteCount = 0;
}
}catch(IOException ioe){
throw new IOException("Parse error...");
}
return new VoteMsg(isResponse,isInquiry,candidateID,voteCount);
}

}

实现类：VoteMsgBinCoder.java

package com.tcpip;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;

/**
* 用二进制表示方法对消息进行编码
* @author Administrator
*
*/
public class VoteMsgBinCoder implements VoteMsgCoder {

public static final int MIN_WIRE_LENGTH = 4;
public static final int MAX_WIRE_LENGTH = 16;
public static final int MAGIC = 0x5400;
public static final int MAGIC_MASK = 0xfc00;
public static final int MAGIC_SHIFT = 8;
public static final int RESPONSE_FLAG = 0x0200;
public static final int INQUIRY_FLAG = 0x0100;

/*
* 将投票消息转换为一个字节序列(根据特定的协议)
* @see com.tcpip.VoteMsgCoder#toWire(com.tcpip.VoteMsg)
*/
@Override
public byte[] toWire(VoteMsg msg) throws IOException {
ByteArrayOutputStream byteStream = new ByteArrayOutputStream();
DataOutputStream out = new DataOutputStream(byteStream);

short magicAndFlags = MAGIC;
if(msg.isInquiry()){
magicAndFlags |= INQUIRY_FLAG;
}
if(msg.isResponse()){
magicAndFlags |= RESPONSE_FLAG;
}
out.writeShort(magicAndFlags);
out.writeShort((short)msg.getCandidateID());
if(msg.isResponse()){
out.writeLong(msg.getVoteCount());
}
out.flush();
byte[] data = byteStream.toByteArray();
return data;
}

/*
* 对给定的字节序列进行解析(根据特定的协议)
* @see com.tcpip.VoteMsgCoder#fromWire(byte[])
*/
@Override
public VoteMsg fromWire(byte[] input) throws IOException {
if(input.length<MIN_WIRE_LENGTH){
throw new IOException("Runt message");
}
ByteArrayInputStream bs = new ByteArrayInputStream(input);
DataInputStream in = new DataInputStream(bs);

int magic = in.readShort();
if((magic&MAGIC_MASK) != MAGIC){
throw new IOException("Bad Magic #:"+((magic&MAGIC_MASK)>>MAGIC_SHIFT));
}
boolean resp = ((magic&RESPONSE_FLAG)!=0);
boolean inq = ((magic&INQUIRY_FLAG)!=0);
int candidateID = in.readShort();
if(candidateID<0 || candidateID>1000){
throw new IOException("Bad candidate ID:"+candidateID);
}
long count = 0;
if(resp){
count = in.readLong();
if(count<0){
throw new IOException("Bad vote count: "+count);
}
}

return new VoteMsg(resp,inq,candidateID,count);
}

}

VoteService.java

package com.tcpip;

import java.util.HashMap;
import java.util.Map;

/**
* 投票服务
* @author Administrator
*
*/
public class VoteService {
private Map<Integer, Long> results = new HashMap<Integer, Long>();

/**
* 处理投票请求，返回投票的信息
* @param msg
* @return
*/
public VoteMsg handleRequest(VoteMsg msg){
if(msg.isResponse()){
return msg;
}
msg.setResponse(true);
int candidateID = msg.getCandidateID();
Long count = results.get(candidateID);
if(count==null){
count = 0L;
}
if(!msg.isInquiry()){
results.put(candidateID, ++count);
}
msg.setVoteCount(count);
return msg;
}

}

tcp客户端：

VoteClientTCP.java:

package com.tcpip;

import java.io.OutputStream;
import java.net.Socket;

public class VoteClientTCP {

public static final int CANDIDATEID = 888;
public static void main(String args[]) throws Exception{

args = new String[]{"127.0.0.1","9999"};

String destAddr= args[0];
int destPort = Integer.parseInt(args[1]);
Socket sock = new Socket(destAddr,destPort);
OutputStream out = sock.getOutputStream();

VoteMsgCoder coder = new VoteMsgBinCoder();

Framer framer = new LengthFramer(sock.getInputStream());
VoteMsg msg = new VoteMsg(false, true, CANDIDATEID, 0);
byte[] encodedMsg = coder.toWire(msg);

System.out.println("Sending Inquiry ("+encodedMsg.length+" bytes):");
System.out.println(msg);
framer.frameMsg(encodedMsg, out);

msg.setInquiry(false);
encodedMsg = coder.toWire(msg);
System.out.println("Sending Vote ("+encodedMsg.length+" bytes):");
framer.frameMsg(encodedMsg, out);

encodedMsg = framer.nextMsg();
msg = coder.fromWire(encodedMsg);
System.out.println("Received Response ("+encodedMsg.length+" bytes):");
System.out.println(msg);

msg = coder.fromWire(framer.nextMsg());
System.out.println("Received Response ("+encodedMsg.length+" bytes):");
System.out.println(msg);
sock.close();
}

}

tcp服务器端：VoteServerTCP.java:

package com.tcpip;

import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;

public class VoteServerTCP {

public static void main(String[] args) throws IOException {

args = new String[]{"9999"};

int port = Integer.parseInt(args[0]);
ServerSocket servSock = new ServerSocket(port);

VoteMsgCoder coder = new VoteMsgBinCoder();
VoteService service = new VoteService();

while(true){
Socket clntSock = servSock.accept();
System.out.println("Handling client at "+clntSock.getRemoteSocketAddress());
Framer framer = new LengthFramer(clntSock.getInputStream());
try{
byte[] req;
while((req = framer.nextMsg())!=null){
System.out.println("Received Message ("+req.length+" bytes)");
VoteMsg responseMsg = service.handleRequest(coder.fromWire(req));
framer.frameMsg(coder.toWire(responseMsg), clntSock.getOutputStream());
}
}catch(IOException ioe){
System.out.println("Error handling client: "+ioe.getMessage());
}finally{
System.out.println("Closing connection");
clntSock.close();
}
}
}

}

udp客户端：VoteClientUDP.java:

package com.tcpip;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.Arrays;

public class VoteClientUDP {

public static void main(String[] args) throws IOException{

args = new String[]{"127.0.0.1","9999","111"};

InetAddress destAddr = InetAddress.getByName(args[0]);
int destPort = Integer.parseInt(args[1]);
int candidate = Integer.parseInt(args[2]);

DatagramSocket sock = new DatagramSocket();
sock.connect(destAddr,destPort);

VoteMsg vote = new VoteMsg(false,false,candidate,0);

VoteMsgCoder coder = new VoteMsgTextCoder();

byte[] encodedVote = coder.toWire(vote);
System.out.println("Sending Text-Encoded Request ("+encodedVote.length+" bytes):");
System.out.println(vote);

DatagramPacket message = new DatagramPacket(encodedVote, encodedVote.length);

sock.send(message);

message = new DatagramPacket(new byte[VoteMsgTextCoder.MAX_WIRE_LENGTH],VoteMsgTextCoder.MAX_WIRE_LENGTH);
sock.receive(message);
encodedVote = Arrays.copyOfRange(message.getData(),0,message.getLength());

System.out.println("Received Text-Encoded Response ("+encodedVote.length + " bytes):");
vote = coder.fromWire(encodedVote);
System.out.println(vote);
}
}

udp服务端：VoteClientUDP.java:

package com.tcpip;

import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.InetAddress;
import java.util.Arrays;

public class VoteClientUDP {

public static void main(String[] args) throws IOException{

args = new String[]{"127.0.0.1","9999","111"};

InetAddress destAddr = InetAddress.getByName(args[0]);
int destPort = Integer.parseInt(args[1]);
int candidate = Integer.parseInt(args[2]);

DatagramSocket sock = new DatagramSocket();
sock.connect(destAddr,destPort);

VoteMsg vote = new VoteMsg(false,false,candidate,0);

VoteMsgCoder coder = new VoteMsgTextCoder();

byte[] encodedVote = coder.toWire(vote);
System.out.println("Sending Text-Encoded Request ("+encodedVote.length+" bytes):");
System.out.println(vote);

DatagramPacket message = new DatagramPacket(encodedVote, encodedVote.length);

sock.send(message);

message = new DatagramPacket(new byte[VoteMsgTextCoder.MAX_WIRE_LENGTH],VoteMsgTextCoder.MAX_WIRE_LENGTH);
sock.receive(message);
encodedVote = Arrays.copyOfRange(message.getData(),0,message.getLength());

System.out.println("Received Text-Encoded Response ("+encodedVote.length + " bytes):");
vote = coder.fromWire(encodedVote);
System.out.println(vote);
}
}