rudpsender.java

基于UDP的可靠邮件系统

package org.hwmhere.udp.send;

import java.io.*;
import java.net.*;

import org.hwmhere.udp.UDPConstant;
import org.hwmhere.util.math.BaseType;

/**
 * @author hwm
 * 
 */
public class RUDPSender {

	private byte[] sendData;// = new byte[UDPConstant.DATA_SIZE];

	private int dataLength;

	private byte[] echoData = new byte[UDPConstant.ECHO_LENGTH];

	private byte[] tempSendData;

	private DatagramSocket sendSocket;

	private InetAddress host;

	private int port;

	/**
	 * 用于发送报文序号的计数 如果初始为0,在RDT2.0中只加不减 只有当一个数据报成功接收并返回ACK后才加一
	 * 
	 * @see #upSequence()
	 * @see #downSequence()
	 */
	private int sequence = 0;

	/**
	 * @param host
	 * @param port
	 * @throws SocketException
	 * @throws UnknownHostException
	 */

	public RUDPSender(InetAddress host, int port, byte[] data) throws Exception {

		if (data.length > UDPConstant.DATA_SIZE) {
			throw new Exception("Data to send is to long.");
		}

		sendData = data;
		dataLength = data.length;

		this.host = host;
		this.port = port;

		sendSocket = new DatagramSocket();
		sendSocket.setSoTimeout(30000);
	}

	public void send() {
		while (!isOver()) {
			try {
				// INFORMation
				System.out.println("Send Datagram " + sequence);
				DatagramPacket sendPacket = getSendPacket();
				// 发送
				sendSocket.send(sendPacket);
				DatagramPacket receivePacket = new DatagramPacket(echoData,
						echoData.length);
				try {
					sendSocket.receive(receivePacket);
					// 成功接受,返回了结果
					if (isACK(receivePacket)) {
						// INFORMation
						System.out.println("Receive Datagram " + sequence
								+ "'s replay");
						upSequence();
					}
				} catch (SocketTimeoutException te) {
					// INFORMation
					System.out.println("Time Out.Continue.");
					continue;
				}
			} catch (IOException e) {
				e.printStackTrace();
				close();
				return;
			}
		}
		// INFORMation
		System.out.println("OK Send Successful!");
		close();
	}

	// TODO:从发送的数据中构造一个需要发送的数据，根据RDT算法
	public DatagramPacket getSendPacket() {

		byte[] tempSequenceByte = getSequenceByte();

		ByteArrayOutputStream bos = new ByteArrayOutputStream();

		bos.write(tempSequenceByte[0]);
		bos.write(tempSequenceByte[1]);
		bos.write(getFlageByte());

		if ((sequence + 1) * UDPConstant.DATA_SEGMENT_UTIL <= dataLength) {
			bos.write(sendData, sequence * UDPConstant.DATA_SEGMENT_UTIL,
					UDPConstant.DATA_SEGMENT_UTIL);
		} else {
			bos.write(sendData, sequence * UDPConstant.DATA_SEGMENT_UTIL,
					dataLength - (sequence * UDPConstant.DATA_SEGMENT_UTIL));
		}
		tempSendData = bos.toByteArray();
		System.out.print("发送数据包长度:" + tempSendData.length);
		return new DatagramPacket(tempSendData, tempSendData.length, host, port);
	}

	public void close() {
		sendSocket.close();
	}

	// TODO:返回是否是ACK
	public boolean isACK(DatagramPacket receivePacket) {
		String answer = new String(receivePacket.getData());
		answer.trim();

		// TODO:getData可能还需要进一步的处理,取出字符串的部分
		if (answer.startsWith(UDPConstant.ACK)) {
			System.out.println("From Server: receive ACK");
			return true;
		} else {
			System.out.println("From Server: receive NAK");
			return false;
		}

	}

	private byte getFlageByte() {

		if (isLast())
			return (byte) 1;
		else
			return (byte) 0;

	}

	/**
	 * @return byte[4]
	 */
	private byte[] getSequenceByte() {

		return BaseType.toLH(sequence);

	}

	// TODO:发送完毕?
	public boolean isOver() {

		if (sequence * UDPConstant.DATA_SEGMENT_UTIL >= dataLength)
			return true;
		return false;
	}

	public boolean isLast() {

		if ((sequence + 1) * UDPConstant.DATA_SEGMENT_UTIL >= dataLength)
			return true;
		return false;
	}

	public DatagramSocket getsendSocket() {
		return sendSocket;
	}

	public void setsendSocket(DatagramSocket sendSocket) {
		this.sendSocket = sendSocket;
	}

	public byte[] getechoData() {
		return echoData;
	}

	public void setechoData(byte[] echoData) {
		this.echoData = echoData;
	}

	public byte[] getSendData() {
		return sendData;
	}

	public void setSendData(byte[] sendData) {
		this.sendData = sendData;
	}

	private void upSequence() {
		sequence++;
	}

	private void downSequence() {
		sequence--;
	}

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

		while (true) {
			InetAddress host = InetAddress.getByName("localhost");

			BufferedReader infromUser = new BufferedReader(
					new InputStreamReader(System.in));

			String sentence = infromUser.readLine();
			byte[] sendData = sentence.getBytes();

			RUDPSender sender = new RUDPSender(host, 6790, sendData);

			sender.send();
		}

	}

}