Subversion Repositories fpga-cf

/**
 * 
 */
package edu.byu.cc.plieber.fpgaenet.fcp;
 
import java.io.IOException;
import java.net.DatagramPacket;
import java.util.LinkedList;
import java.util.concurrent.LinkedBlockingQueue;
 
;
 
/**
 * @author Peter Lieber
 * 
 */
class FCPSendThread extends Thread {
	private FCPProtocol protocol;
	protected LinkedBlockingQueue<FCPPacket> sendQueue;
	volatile protected boolean done = false;
	volatile protected boolean sendTimeout = false;
	volatile protected boolean ioException = false;
	protected long whileCount = 0;
 
	public FCPSendThread(FCPProtocol protocol) {
		this.protocol = protocol;
		this.sendQueue = new LinkedBlockingQueue<FCPPacket>();
	}
 
	@Override
	public void run() {
		while (true) {
			try {
				if (done == true)
					return;
				while (sendQueue.isEmpty()) {
					if (done == true)
						return;
				}
				FCPPacket fcppacket = sendQueue.take();
				if (fcppacket.command == 0 || fcppacket.command == 4) {
					fcppacket.seq = ++protocol.snd_cur;
					if (fcppacket.seq > 65000) {
						this.reconnect();
						fcppacket.seq = ++protocol.snd_cur;
					}
				}
				else
					fcppacket.seq = 0;
 
				//System.out.println("Sending Packet: " + fcppacket.toString());
				DatagramPacket packet = fcppacket.wrapInDatagram(fcppacket.dest, fcppacket.dstPort);
 
				// parse
				if ( true ) {//protocol.sendWindow <= 20) {
					if ((fcppacket.command == 0 || fcppacket.command == 4 || fcppacket.command == 2)
							&& fcppacket.seq > protocol.snd_last_ack + protocol.sendWindow) {
						int numResendsLeft = 5;
						long timeouttime = System.currentTimeMillis() + protocol.timeout;
						long newWhileCnt = whileCount + 1;
						while ((fcppacket.command == 0 || fcppacket.command == 4)
								&& fcppacket.seq > protocol.snd_last_ack + protocol.sendWindow) {
							whileCount = newWhileCnt;
							//System.out.println("Waiting");
							if (System.currentTimeMillis() > timeouttime) {
								if (numResendsLeft <= 0)
									throw new FCPException("Communication Error: Resend Limit Reached");
								numResendsLeft--;
								protocol.socket.send(protocol.packetOutbox.get(protocol.snd_last_ack + 1).wrapInDatagram());
								System.out.println("Resent: " + (protocol.snd_last_ack + 1));
								timeouttime = System.currentTimeMillis() + protocol.timeout;
							}
							if (done == true)
								return;
						}
					}
				} 
 
				if (fcppacket.command == 0 || fcppacket.command == 4)
					protocol.packetOutbox.put(fcppacket.seq, fcppacket);
				protocol.socket.send(packet);
			} catch (IOException e) {
				this.ioException = true;
				return;
			} catch (FCPException e) {
				this.sendTimeout = true;
				return;
			} catch (InterruptedException e) {
 
			}
		}
	}
 
	private void reconnect() {
		FCPPacket fp = new FCPPacket();
		fp.version = 0;
		fp.command = 2;
		fp.port = 0;
		fp.seq = 0;
		fp.len = 0;
		fp.dest = protocol.connectedAddress;
		fp.dstPort = protocol.connectedPort;
		try {
			protocol.connected = false;
			protocol.packetOutbox.clear();
			protocol.socket.send(fp.wrapInDatagram());
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		while (!protocol.connected);
	}
}