public static final byte FIN = (byte) (VERSION | longToUbyte(16));
 public static final byte SELECTIVE_ACK = longToUbyte(1);
 public static final byte STATE = (byte) (VERSION | longToUbyte(32));
 public static UtpPacket extractUtpPacket(DatagramPacket dgpkt) {
		
	UtpPacket pkt = new UtpPacket();
	byte[] pktb = dgpkt.getData();
	pkt.setFromByteArray(pktb, dgpkt.getLength(), dgpkt.getOffset());
	return pkt;
}
 public static boolean isDataPacket(DatagramPacket udpPacket) {
	return isPacketType(udpPacket, DATA);
}
 public static boolean isFinPacket(DatagramPacket udpPacket) {
	return isPacketType(udpPacket, FIN);
}
 public static boolean isResetPacket(DatagramPacket udpPacket) {
	return isPacketType(udpPacket, RESET);
}
 public static boolean isStatePacket(DatagramPacket udpPacket) {
	return isPacketType(udpPacket, STATE);
}
 public static boolean isSynPkt(UtpPacket packet) {
		
	if (packet == null) {
		return false;
	}
		
	return packet.getTypeVersion() == SYN;
		
}
 public static final long MAX_USHORT = 65535;
 public static int longToUint(long longvalue) {
	if (longvalue > MAX_UINT) {
		throw new ByteOverflowException(getExceptionText(MAX_UINT, longvalue));
	} else if (longvalue < 0) {
		throw new SignedNumberException(getExceptionText(MAX_UINT, longvalue));
	}
	return (int) (longvalue & 0xFFFFFFFF);
}
 public static short longToUshort(long longvalue) {
	if (longvalue > MAX_USHORT) {
		throw new ByteOverflowException(getExceptionText(MAX_USHORT, longvalue));
	} else if (longvalue < 0) {
		throw new SignedNumberException(getExceptionText(MAX_USHORT, longvalue));
	}
	return (short) (longvalue & 0xFFFF);
}
 public abstract class UtpSocketChannel {

	/* ID for outgoing packets */
	private long connectionIdSending;

	/* timestamping utility */
	protected MicroSecondsTimeStamp timeStamper = new MicroSecondsTimeStamp();

	/* current sequenceNumber */
	private int sequenceNumber;

	/* ID for incomming packets */
	private long connectionIdRecieving;

	/*
	 * address of the remote sockets which this socket is connected to
	 */
	protected SocketAddress remoteAddress;

	/* current ack Number */
	protected int ackNumber;

	/* reference to the underlying UDP Socket */
	protected DatagramSocket dgSocket;

	/*
	 * Connection Future Object - need to hold a reference here i case
	 * connection the initial connection attempt fails. So it will be updates
	 * once the reattempts success.
	 */
	protected UtpConnectFutureImpl connectFuture = null;

	/* lock for the socket state* */
	protected final ReentrantLock stateLock = new ReentrantLock();

	/* logger */
	private static final Logger log = LoggerFactory
			.getLogger(UtpSocketChannel.class);

	/* Current state of the socket */
	protected volatile UtpSocketState state = null;

	/* Sequencing begin */
	protected static int DEF_SEQ_START = 1;

	/**
	 * Opens a new Socket and binds it to any available port
	 * 
	 * @return {@link UtpSocketChannel}
	 * @throws IOException
	 *             see {@link DatagramSocket#DatagramSocket()}
	 */
	public static UtpSocketChannel open() throws IOException {
		UtpSocketChannelImpl c = new UtpSocketChannelImpl();
		try {
			c.setDgSocket(new DatagramSocket());
			c.setState(CLOSED);
		} catch (IOException exp) {
			throw new IOException("Could not open UtpSocketChannel: "
					+ exp.getMessage());
		}
		return c;
	}

	/**
	 * Connects this Socket to the specified address
	 * 
	 * @param address
	 * @return {@link UtpConnectFuture}
	 */
	public UtpConnectFuture connect(SocketAddress address) {
		stateLock.lock();
		try {
			try {
				connectFuture = new UtpConnectFutureImpl();
			} catch (InterruptedException e) {
				e.printStackTrace();
				return null;
			}
			try {
				/* underlying impl does it's connection setup */
				connectImpl(connectFuture);
				
				/* fill packet, set initial variables and send packet */
				setRemoteAddress(address);
				setupConnectionId();
				setSequenceNumber(DEF_SEQ_START);

				UtpPacket synPacket = UtpPacketUtils.createSynPacket();
				synPacket
						.setConnectionId(longToUshort(getConnectionIdRecieving()));
				synPacket.setTimestamp(timeStamper.utpTimeStamp());
				sendPacket(synPacket);
				setState(SYN_SENT);
				printState("[Syn send] ");

				incrementSequenceNumber();
				startConnectionTimeOutCounter(synPacket);
			} catch (IOException exp) {
				// DO NOTHING, let's try later with reconnect runnable
			}
		} finally {
			stateLock.unlock();
		}

		return connectFuture;
	}

	/**
	 * Writes the content of the Buffer to the Channel.
	 * 
	 * @param src
	 *            the buffer, it is expected to be in Reading-Mode.
	 * @return {@link UtpWriteFuture} which will be updated by the channel
	 */
	public abstract UtpWriteFuture write(ByteBuffer src);
	
	/**
	 * Reads the incomming data to the channel.
	 * 
	 * @param src
	 *            the buffer.
	 * @return {@link UtpWriteFuture} which will be updated by the channel
	 */
	public abstract UtpReadFuture read(ByteBuffer dst);
	
	
	/**
	 * Closes the channel. Also unbinds the socket if the socket is not shared,
	 * for example with the server.
	 */
	public abstract UtpCloseFuture close();

	/**
	 * @return The Connection ID for Outgoing Packets
	 */
	public long getConnectionIdsending() {
		return connectionIdSending;
	}
	
	/**
	 * @return true if the channel is open.
	 */
	public boolean isOpen() {
		return state != CLOSED;
	}
	
	public abstract boolean isReading();

	public abstract boolean isWriting();
	
	/**
	 * @return true, if this socket is connected to another socket. 
	 */
	public boolean isConnected() {
		return getState() == UtpSocketState.CONNECTED;
	}
	
	public long getConnectionIdRecieving() {
		return connectionIdRecieving;
	}
	
	public UtpSocketState getState() {
		return state;
	}
	
	public SocketAddress getRemoteAdress() {
		return remoteAddress;
	}

	public int getAckNumber() {
		return ackNumber;
	}
	
	public int getSequenceNumber() {
		return sequenceNumber;
	}
	
	public DatagramSocket getDgSocket() {
		return dgSocket;
	}
	
	/* signal the implementation to start the reConnect Timer */
	protected abstract void startConnectionTimeOutCounter(UtpPacket synPacket);

	/* debug method to print id's, seq# and ack# */
	protected void printState(String msg) {
		String state = "[ConnID Sending: " + connectionIdSending + "] "
				+ "[ConnID Recv: " + connectionIdRecieving + "] [SeqNr. "
				+ sequenceNumber + "] [AckNr: " + ackNumber + "]";
		log.debug(msg + state);

	}
	
	/* implementation to cancel all operations and close the socket */
	protected abstract void abortImpl();

	/* if implementation must something do during the connection */
	protected abstract void connectImpl(UtpConnectFutureImpl future);

	/*
	 * increments current sequence number unlike TCP, uTp is sequencing its
	 * packets, not bytes but the Seq# is only 16 bits, so overflows are likely
	 * to happen Seq# == 0 not possible.
	 */
	protected void incrementSequenceNumber() {
		int seqNumber = getSequenceNumber() + 1;
		if (seqNumber > MAX_USHORT) {
			seqNumber = 1;
		}
		setSequenceNumber(seqNumber);
	}

	/*
	 * general methods to send packets need to be public in the impl, but also
	 * need to be accessed from this class.
	 */
	protected abstract void sendPacket(UtpPacket packet) throws IOException;

	protected abstract void sendPacket(DatagramPacket pkt) throws IOException;

	/*
	 * sets up connection ids. incomming = rnd outgoing = incomming + 1
	 */
	private void setupConnectionId() {
		Random rnd = new Random();
		int max = (int) (MAX_USHORT - 1);
		long rndInt = rnd.nextInt(max);
		setConnectionIdRecieving(rndInt);
		setConnectionIdsending(rndInt + 1);

	}


	/* set connection ID for outgoing packets */
	protected void setConnectionIdsending(long connectionIdSending) {
		this.connectionIdSending = connectionIdSending;
	}

	/**
	 * Creates an Ack-Packet.
	 * 
	 * @param pkt
	 *            the packet to be Acked
	 * @param timedifference
	 *            Between t1 and t2. t1 beeing the time the remote socket has
	 *            sent this packet, t2 when this socket recieved the packet.
	 *            Differences is <b>unsigned</b> and in µs resolution.
	 * @param advertisedWindow
	 *            How much space this socket has in its temporary recieve buffer
	 * @return ack packet
	 */
	protected abstract UtpPacket createAckPacket(UtpPacket pkt, int timedifference,
			long advertisedWindow);

	/*
	 * setting the ack current ack number to the sequence number of the packet.
	 * needs to be accessed from this class and publicly from the
	 * implementation, hence protected and abstract.
	 */
	protected abstract void setAckNrFromPacketSqNr(UtpPacket utpPacket);

	/*
	 * creates a datapacket, increments seq. number the packet is ready to be
	 * filled with data.
	 */
	protected abstract UtpPacket createDataPacket();

	protected void setConnectionIdRecieving(long connectionIdRecieving) {
		this.connectionIdRecieving = connectionIdRecieving;
	}

	protected void setState(UtpSocketState state) {
		this.state = state;
	}


	protected void setSequenceNumber(int sequenceNumber) {
		this.sequenceNumber = sequenceNumber;
	}

	protected abstract void setRemoteAddress(SocketAddress remoteAdress);

	protected abstract void setAckNumber(int ackNumber);

	protected abstract void setDgSocket(DatagramSocket dgSocket);

} public enum UtpSocketState {
	
		/**
		 * Indicates that a syn packet has been send.
		 */
		SYN_SENT,
		
		/**
		 * Indicates that a connection has been established.
		 */
		CONNECTED,
		
		/**
		 * Indicates that no connection is established.
		 */
		CLOSED,
		
		/**
		 * Indicates that a SYN has been recieved but could not be acked.
		 */
		SYN_ACKING_FAILED, 
		
		/**
		 * Indicates that the fin Packet has been send.
		 */
		FIN_SEND,
		
		/**
		 * Indicates that a fin was recieved. 
		 */
		GOT_FIN,
		
}
 public class UtpCloseFuture extends UtpBlockableFuture {

	public UtpCloseFuture() throws InterruptedException {
		super();
	}
}
 public abstract class UtpWriteFuture extends UtpBlockableFuture{
	
	public UtpWriteFuture() throws InterruptedException {
		super();
	}

	protected volatile int bytesWritten;
	
	/**
	 * @return number of bytes that was written.
	 */
	public int getBytesSend() {
		return bytesWritten;
	}
	
}
 public class UtpAlgConfiguration {

	public static final int MAX_CONNECTION_ATTEMPTS = 5;
	public static final int CONNECTION_ATTEMPT_INTERVALL_MILLIS = 5000;
	
	public static long MINIMUM_DELTA_TO_MAX_WINDOW_MICROS = 1000000;
	// ack every second packets
	public static int SKIP_PACKETS_UNTIL_ACK = 2;
	

	/**
	 * TWEAKING SECTION
	 */

	/**
	 * Auto ack every packet that is smaller than ACK_NR from ack packet. 
	 * Some Implementations like libutp do this. 
	 */
	public static boolean AUTO_ACK_SMALLER_THAN_ACK_NUMBER = true;
	/**
	 * if oldest mindelay sample is older than that, update it. 
	 */
	public static long MINIMUM_DIFFERENCE_TIMESTAMP_MICROSEC = 120000000L;

	/**
	 * timeout 
	 */
	public static int MINIMUM_TIMEOUT_MILLIS = 500;

	/**
	 * Packet size modus
	 */
	public static PacketSizeModus PACKET_SIZE_MODE = PacketSizeModus.CONSTANT_1472;

	/**
	 * maximum packet size should be dynamically set once path mtu discovery
	 * implemented.
	 */

	public volatile static int MAX_PACKET_SIZE = 1472;

	/**
	 * minimum packet size.
	 */
	public volatile static int MIN_PACKET_SIZE = 150;

	/**
	 * Minimum path MTU
	 */
	public volatile static int MINIMUM_MTU = 576;

	/**
	 * Maximal window increase per RTT - increase to allow uTP throttle up
	 * faster.
	 */
	public volatile static int MAX_CWND_INCREASE_PACKETS_PER_RTT = 3000;

	/**
	 * maximal buffering delay
	 */
	public volatile static int C_CONTROL_TARGET_MICROS = 100000;

	/**
	 * activate burst sending
	 */
	public volatile static boolean SEND_IN_BURST = true;

	/**
	 * Reduce burst sending artificially
	 */
	public volatile static int MAX_BURST_SEND = 5;

	/**
	 * Minimum number of acks past seqNr=x to trigger a resend of seqNr=x;
	 */
	public volatile static int MIN_SKIP_PACKET_BEFORE_RESEND = 3;

	public volatile static long MICROSECOND_WAIT_BETWEEN_BURSTS = 28000;

	public volatile static long TIME_WAIT_AFTER_LAST_PACKET = 3000000;

	public volatile static boolean ONLY_POSITIVE_GAIN = false;
	
	public volatile static boolean DEBUG = false;
	

	/**
	 * @return information about the algorithm. This is only used for debugging 
	 */
	public static String getString() {
		String toReturn = "";
		toReturn += "MINIMUM_TIMEOUT_MILLIS: " + MINIMUM_TIMEOUT_MILLIS + " ";
		toReturn += "PACKET_SIZE_MODE: " + PACKET_SIZE_MODE + " ";
		toReturn += "MAX_PACKET_SIZE: " + MAX_PACKET_SIZE + " ";
		toReturn += "MIN_PACKET_SIZE: " + MIN_PACKET_SIZE + " ";
		toReturn += "MINIMUM_MTU: " + MINIMUM_MTU + " ";
		toReturn += "MAX_CWND_INCREASE_PACKETS_PER_RTT: " + MAX_CWND_INCREASE_PACKETS_PER_RTT + " ";
		toReturn += "C_CONTROL_TARGET_MICROS: " + C_CONTROL_TARGET_MICROS + " ";
		toReturn += "SEND_IN_BURST: " + SEND_IN_BURST + " ";
		toReturn += "MAX_BURST_SEND: " + MAX_BURST_SEND + " ";
		toReturn += "MIN_SKIP_PACKET_BEFORE_RESEND: " + MIN_SKIP_PACKET_BEFORE_RESEND + " ";
		toReturn += "MICROSECOND_WAIT_BETWEEN_BURSTS: " + MICROSECOND_WAIT_BETWEEN_BURSTS + " ";
		toReturn += "TIME_WAIT_AFTER_FIN_MICROS: " + TIME_WAIT_AFTER_LAST_PACKET + " ";
		toReturn += "ONLY_POSITIVE_GAIN: " + ONLY_POSITIVE_GAIN + " ";
		toReturn += "DEBUG: " + DEBUG + " ";
		return toReturn;
	}

	
} public class ConnectionTimeOutRunnable implements Runnable {

	private UtpPacket synPacket;
	private UtpSocketChannelImpl channel;
	private volatile ReentrantLock lock;

	public ConnectionTimeOutRunnable(UtpPacket packet,
			UtpSocketChannelImpl channel, ReentrantLock lock) {
		this.synPacket = packet;
		this.lock = lock;
		this.channel = channel;
	}

	@Override
	public void run() { 
		channel.resendSynPacket(synPacket);
	}


}
 public class UtpConnectFutureImpl extends UtpConnectFuture {
	
	public UtpConnectFutureImpl() throws InterruptedException {
		super();
	}
	/**
	 * Releases semaphore and sets return values.
	 * @param exp if exception occured, null is possible.
	 */
	public void finished(IOException exp) {
		this.exception = exp;
		this.isDone = true;
		semaphore.release();
	}

}
 public class UtpReadFutureImpl extends UtpReadFuture {

	public UtpReadFutureImpl() throws InterruptedException {
		super(); 
	}

	/**
	 * Releasing semaphore and running the listener if set.
	 * @param exp
	 * @param buffer
	 */
	public void finished(IOException exp, ByteBuffer buffer) {
		this.buffer = buffer;
		this.exception = exp;
		isDone = true;
		semaphore.release();
		listenerLock.lock();
		try {
			if (listener != null) {
				listener.setByteBuffer(buffer);
				listener.setIOException(exp);
				if (listener.createExtraThread()) {
					Thread listenerThread = new Thread(listener, listener.getThreadName());
					listenerThread.start();
				} else {
					listener.run();
				}
			}
		} finally {
			listenerLock.unlock();
		}
	}
}
 public class UtpReadingRunnable extends Thread implements Runnable {
	
	private static final int PACKET_DIFF_WARP = 50000;
	private IOException exp;
	private ByteBuffer buffer;
	private UtpSocketChannelImpl channel;
	private SkippedPacketBuffer skippedBuffer = new SkippedPacketBuffer();
	private boolean exceptionOccured = false;
	private boolean graceFullInterrupt;
	private boolean isRunning;
	private MicroSecondsTimeStamp timeStamper;
	private long totalPayloadLength = 0;
	private long lastPacketTimestamp;
	private int lastPayloadLength;
	private UtpReadFutureImpl readFuture;
	private long nowtimeStamp;
	private long lastPackedRecieved;
	private long startReadingTimeStamp;
	private boolean gotLastPacket = false;
	// in case we ack every x-th packet, this is the counter. 
	private int currentPackedAck = 0;
	
	private static final Logger log = LoggerFactory.getLogger(UtpReadingRunnable.class);
	
	public UtpReadingRunnable(UtpSocketChannelImpl channel, ByteBuffer buff, MicroSecondsTimeStamp timestamp, UtpReadFutureImpl future) {
		this.channel = channel;
		this.buffer = buff;
		this.timeStamper = timestamp;
		this.readFuture = future;
		lastPayloadLength = UtpAlgConfiguration.MAX_PACKET_SIZE;
		this.startReadingTimeStamp = timestamp.timeStamp();
	}
	
	public int getBytesRead() {
		return 0;
	}

	public IOException getException() {
		return exp;
	}

	public boolean hasExceptionOccured() {
		return false;
	}

	@Override
	public void run() {
//		buffer.flip();

		isRunning = true;
		IOException exp = null;
		while(continueReading()) {
			BlockingQueue<UtpTimestampedPacketDTO> queue = channel.getDataGramQueue();
			try {
				UtpTimestampedPacketDTO timestampedPair = queue.poll(UtpAlgConfiguration.TIME_WAIT_AFTER_LAST_PACKET/2, TimeUnit.MICROSECONDS);
				nowtimeStamp = timeStamper.timeStamp();
				if (timestampedPair != null) {
					currentPackedAck++;
//					log.debug("Seq: " + (timestampedPair.utpPacket().getSequenceNumber() & 0xFFFF));
					lastPackedRecieved = timestampedPair.stamp();
					if (isLastPacket(timestampedPair)) {
						gotLastPacket  = true;
						log.debug("GOT LAST PACKET");
						lastPacketTimestamp = timeStamper.timeStamp();
					}
					if (isPacketExpected(timestampedPair.utpPacket())) {
						handleExpectedPacket(timestampedPair);								
					} else {
						handleUnexpectedPacket(timestampedPair);
					}
					if (ackThisPacket()) {
						currentPackedAck = 0;
					}
				}
				
				/*TODO: How to measure Rtt here for dynamic timeout limit?*/
				if (isTimedOut()) {
					if (!hasSkippedPackets()) {
						gotLastPacket  = true;
						log.debug("ENDING READING, NO MORE INCOMMING DATA");
					} else {
						log.debug("now: " + nowtimeStamp + " last: " + lastPackedRecieved + " = " + (nowtimeStamp - lastPackedRecieved));
						log.debug("now: " + nowtimeStamp + " start: " + startReadingTimeStamp + " = " + (nowtimeStamp - startReadingTimeStamp));
						throw new IOException();						
					}
				}
					
			} catch (IOException ioe) {
				exp = ioe;
				exp.printStackTrace();
				exceptionOccured = true;
			} catch (InterruptedException iexp) {
				iexp.printStackTrace();
				exceptionOccured = true;
			} catch (ArrayIndexOutOfBoundsException aexp) {
				aexp.printStackTrace();
				exceptionOccured = true;
				exp = new IOException();
			}
		}
		isRunning = false;
		readFuture.finished(exp, buffer);
		
		
		log.debug("Buffer position: " + buffer.position() + " buffer limit: " + buffer.limit());
		log.debug("PAYLOAD LENGHT " + totalPayloadLength);
		log.debug("READER OUT");

		channel.returnFromReading();

	}
	
	private boolean isTimedOut() {
		//TODO: extract constants...
		/* time out after 4sec, when eof not reached */
		boolean timedOut = nowtimeStamp - lastPackedRecieved >= 4000000;
		/* but if remote socket has not recieved synack yet, he will try to reconnect
		 * await that aswell */
		boolean connectionReattemptAwaited = nowtimeStamp - startReadingTimeStamp >= 4000000;
		return timedOut && connectionReattemptAwaited;
	}

	private boolean isLastPacket(UtpTimestampedPacketDTO timestampedPair) {
//		log.debug("WindowSize: " + (timestampedPair.utpPacket().getWindowSize() & 0xFFFFFFFF));
//		log.debug("got Packet: " + (timestampedPair.utpPacket().getSequenceNumber() & 0xFFFF) + " and will ack? " + ackThisPacket() + " acksizeCounter " + currentPackedAck);
		return (timestampedPair.utpPacket().getWindowSize() & 0xFFFFFFFF) == 0;
	}

	private void handleExpectedPacket(UtpTimestampedPacketDTO timestampedPair) throws IOException {
//		log.debug("handling expected packet: " + (timestampedPair.utpPacket().getSequenceNumber() & 0xFFFF));
		if (hasSkippedPackets()) {
			buffer.put(timestampedPair.utpPacket().getPayload());
			int payloadLength = timestampedPair.utpPacket().getPayload().length;
			lastPayloadLength = payloadLength;
			totalPayloadLength += payloadLength;
			Queue<UtpTimestampedPacketDTO> packets = skippedBuffer.getAllUntillNextMissing();
			int lastSeqNumber = 0;
			if (packets.isEmpty()) {
				lastSeqNumber = timestampedPair.utpPacket().getSequenceNumber() & 0xFFFF;
			}
			UtpPacket lastPacket = null;
			for (UtpTimestampedPacketDTO p : packets) {
				buffer.put(p.utpPacket().getPayload());
				payloadLength += p.utpPacket().getPayload().length;
				lastSeqNumber = p.utpPacket().getSequenceNumber() & 0xFFFF;
				lastPacket = p.utpPacket();
			}
			skippedBuffer.reindex(lastSeqNumber);
			channel.setAckNumber(lastSeqNumber);
			//if still has skipped packets, need to selectively ack
			if (hasSkippedPackets()) {
				if (ackThisPacket()) {
//					log.debug("acking expected, had, still have");
					SelectiveAckHeaderExtension headerExtension = skippedBuffer.createHeaderExtension();
					channel.selectiveAckPacket(headerExtension, getTimestampDifference(timestampedPair), getLeftSpaceInBuffer());								
				}

			} else {
				if (ackThisPacket()) {
//					log.debug("acking expected, has, than has nomore");
					channel.ackPacket(lastPacket, getTimestampDifference(timestampedPair), getLeftSpaceInBuffer());					
				}
			}
		} else {
			if (ackThisPacket()) {
//				log.debug("acking expected, nomore");
				channel.ackPacket(timestampedPair.utpPacket(), getTimestampDifference(timestampedPair), getLeftSpaceInBuffer());				
			} else {
				channel.setAckNumber(timestampedPair.utpPacket().getSequenceNumber() & 0xFFFF);
			}
			buffer.put(timestampedPair.utpPacket().getPayload());
			totalPayloadLength += timestampedPair.utpPacket().getPayload().length;
		}
	}
	
	private boolean ackThisPacket() {
		if (currentPackedAck >= UtpAlgConfiguration.SKIP_PACKETS_UNTIL_ACK) {
			return true;
		} 
		return false;
	}

	/**
	 * Returns the average space available in the buffer in Bytes.
	 * @return bytes
	 */
	public long getLeftSpaceInBuffer() throws IOException {
		return (skippedBuffer.getFreeSize()) * lastPayloadLength;
	}

	private int getTimestampDifference(UtpTimestampedPacketDTO timestampedPair) {
		int difference = timeStamper.utpDifference(timestampedPair.utpTimeStamp(), timestampedPair.utpPacket().getTimestamp());
		return difference;
	}

	private void handleUnexpectedPacket(UtpTimestampedPacketDTO timestampedPair) throws IOException {
		int expected = getExpectedSeqNr();
//		log.debug("handling unexpected packet: " + (timestampedPair.utpPacket().getSequenceNumber() & 0xFFFF));
		int seqNr = timestampedPair.utpPacket().getSequenceNumber() & 0xFFFF;
		if (skippedBuffer.isEmpty()) {
			skippedBuffer.setExpectedSequenceNumber(expected);
		}
		//TODO: wrapping seq nr: expected can be 5 e.g.
		// but buffer can recieve 65xxx, which already has been acked, since seq numbers wrapped. 
		// current implementation puts this wrongly into the buffer. it should go in the else block
		// possible fix: alreadyAcked = expected > seqNr || seqNr - expected > CONSTANT;
		boolean alreadyAcked = expected > seqNr || seqNr - expected > PACKET_DIFF_WARP;
		
		boolean saneSeqNr = expected == skippedBuffer.getExpectedSequenceNumber();
//		log.debug("saneSeqNr: " + saneSeqNr + " alreadyAcked: " + alreadyAcked + " will ack: " + ackThisPacket());
		if (saneSeqNr && !alreadyAcked) {
			skippedBuffer.bufferPacket(timestampedPair);
			// need to create header extension after the packet is put into the incomming buffer. 
			SelectiveAckHeaderExtension headerExtension = skippedBuffer.createHeaderExtension();
			if (ackThisPacket()) {
//				log.debug("acking unexpected snae");
				channel.selectiveAckPacket(headerExtension, getTimestampDifference(timestampedPair), getLeftSpaceInBuffer());									
			}
		} else if (ackThisPacket()){
			SelectiveAckHeaderExtension headerExtension = skippedBuffer.createHeaderExtension();
//			log.debug("acking unexpected  nonsane");
			channel.ackAlreadyAcked(headerExtension, getTimestampDifference(timestampedPair), getLeftSpaceInBuffer());
		}
	}
	
	/**
	 * True if this packet is expected. 
	 * @param utpPacket packet
	 * @return
	 */
	public boolean isPacketExpected(UtpPacket utpPacket) {
		int seqNumberFromPacket = utpPacket.getSequenceNumber() & 0xFFFF;
//		log.debug("Expected Sequence Number == " + getExpectedSeqNr());
		return getExpectedSeqNr() == seqNumberFromPacket;
	}
	
	private int getExpectedSeqNr() {
		int ackNumber = channel.getAckNumber();
		if (ackNumber == UnsignedTypesUtil.MAX_USHORT) {
			return 1;
		} 
		return ackNumber + 1;
	}

	private boolean hasSkippedPackets() {
		return !skippedBuffer.isEmpty();
	}

	public void graceFullInterrupt() {
		this.graceFullInterrupt = true;
	}

	private boolean continueReading() {
		return !graceFullInterrupt && !exceptionOccured 
				&& (!gotLastPacket || hasSkippedPackets() || !timeAwaitedAfterLastPacket());
	}
		
	private boolean timeAwaitedAfterLastPacket() {
		return (timeStamper.timeStamp() - lastPacketTimestamp) > UtpAlgConfiguration.TIME_WAIT_AFTER_LAST_PACKET 
				&& gotLastPacket;
	}

	public boolean isRunning() {
		return isRunning;
	}

	public MicroSecondsTimeStamp getTimestamp() {
		return timeStamper;
	}

	public void setTimestamp(MicroSecondsTimeStamp timestamp) {
		this.timeStamper = timestamp;
	}
}
 public interface UtpPacketRecievable {
	/**
	 * Recieve that packet. 
	 * @param packet
	 */
	public void recievePacket(DatagramPacket packet);

}
 public class UtpRecieveRunnable extends Thread implements Runnable {


	private DatagramSocket socket;
	private UtpPacketRecievable packetReciever;
	private boolean graceFullInterrupt = false;
	
	private final static Logger log = LoggerFactory.getLogger(UtpRecieveRunnable.class);
	

	public UtpRecieveRunnable(DatagramSocket socket, UtpPacketRecievable queueable) {
		this.socket = socket;
		this.packetReciever = queueable;
	}
		
	public void graceFullInterrupt() {
		super.interrupt();
		graceFullInterrupt = true;
		socket.close();
	}
	
	private boolean continueThread() {
		return !graceFullInterrupt;
	}
	
	@Override
	public void run() {
		while(continueThread()) {
			byte[] buffer = new byte[MAX_UDP_HEADER_LENGTH + MAX_UTP_PACKET_LENGTH];
			DatagramPacket dgpkt = new DatagramPacket(buffer, buffer.length);
			try {
				//the packet will be filled here
				socket.receive(dgpkt);
				//hand packet to the queue
				packetReciever.recievePacket(dgpkt);
//				(new UtpPassPacket(dgpkt, queueable)).start();

			} catch (IOException exp) {
				if (graceFullInterrupt) {
					log.debug("socket closing");
					break;					
				} else {
					exp.printStackTrace();					
				}
			}
		}
		log.debug("RECIEVER OUT");
		
	}

}
 public class UtpWriteFutureImpl extends UtpWriteFuture {
		
	
	public UtpWriteFutureImpl() throws InterruptedException {
		super();
	}

	public void finished(IOException exp, int bytesWritten) {
		this.setBytesSend(bytesWritten);
		this.exception = exp;
		isDone = true;
		semaphore.release();
	}

	public void setBytesSend(int position) {
		bytesWritten = position;
	}

}
 public class UtpWritingRunnable extends Thread implements Runnable {
	
	private ByteBuffer buffer;
	private volatile boolean graceFullInterrupt;
	private UtpSocketChannelImpl channel;
	private boolean isRunning = false;
	private UtpAlgorithm algorithm;
	private IOException possibleException = null;
	private MicroSecondsTimeStamp timeStamper;
	private UtpWriteFutureImpl future;
	
	private final static Logger log = LoggerFactory.getLogger(UtpWritingRunnable.class);
	
	public UtpWritingRunnable(UtpSocketChannelImpl channel, ByteBuffer buffer, 
			MicroSecondsTimeStamp timeStamper, UtpWriteFutureImpl future) {
		this.buffer = buffer;
		this.channel = channel;
		this.timeStamper = timeStamper;
		this.future = future;
		algorithm = new UtpAlgorithm(timeStamper, channel.getRemoteAdress());
	}


	@Override
	public void run() {
		algorithm.initiateAckPosition(channel.getSequenceNumber());
		algorithm.setTimeStamper(timeStamper);
		algorithm.setByteBuffer(buffer);
		isRunning = true;
		IOException possibleExp = null;
		boolean exceptionOccured = false;
		buffer.flip();
		int durchgang = 0;
		while(continueSending()) {
			try {
				if(!checkForAcks()) {
					graceFullInterrupt = true;
					break;
				}
				
				Queue<DatagramPacket> packetsToResend = algorithm.getPacketsToResend();
				for (DatagramPacket datagramPacket : packetsToResend) {
					datagramPacket.setSocketAddress(channel.getRemoteAdress());
					channel.sendPacket(datagramPacket);
				}
				
			} catch (IOException exp) {
				exp.printStackTrace();
				graceFullInterrupt = true;
				possibleExp = exp;
				exceptionOccured = true;
				break;
			}
			if (algorithm.isTimedOut()) {
				graceFullInterrupt = true;
				possibleExp = new IOException("timed out");
				log.debug("timed out");
				exceptionOccured = true;
			}
//			if(!checkForAcks()) {
//				graceFullInterrupt = true;
//				break;
//			}
			while (algorithm.canSendNextPacket() && !exceptionOccured && !graceFullInterrupt && buffer.hasRemaining()) {
				try {
					channel.sendPacket(getNextPacket());	
				} catch (IOException exp) {
					exp.printStackTrace();
					graceFullInterrupt = true;
					possibleExp = exp;
					exceptionOccured = true;
					break;
				}
			}
//			if (!buffer.hasRemaining() && !finSend) {
//				UtpPacket fin = channel.getFinPacket();
//				log.debug("Sending FIN");
//				try {
//					channel.finalizeConnection(fin);
//					algorithm.markFinOnfly(fin);
//				} catch (IOException exp) {
//					exp.printStackTrace();
//					graceFullInterrupt = true;
//					possibleExp = exp;
//					exceptionOccured = true;
//				}
//				finSend = true;
//			}
			uptadeFuture();
			durchgang++;
			if (durchgang % 1000 == 0) {
				log.debug("buffer position: " + buffer.position() + " buffer limit: " + buffer.limit());
			}
		}

		if (possibleExp != null) {
			exceptionOccured(possibleExp);
		}
		isRunning = false;
		algorithm.end(buffer.position(), !exceptionOccured);
		future.finished(possibleExp, buffer.position());
		log.debug("WRITER OUT");
		channel.removeWriter();
	}
	
	private void uptadeFuture() {
		future.setBytesSend(buffer.position());
		
	}


	private boolean checkForAcks() {
		BlockingQueue<UtpTimestampedPacketDTO> queue = channel.getDataGramQueue();
		try {
			waitAndProcessAcks(queue);
		} catch (InterruptedException ie) {
			return false;
		}
		return true;
	}
	
	private void waitAndProcessAcks(BlockingQueue<UtpTimestampedPacketDTO> queue) throws InterruptedException {
		long waitingTimeMicros = algorithm.getWaitingTimeMicroSeconds();
		UtpTimestampedPacketDTO temp = queue.poll(waitingTimeMicros, TimeUnit.MICROSECONDS);
		if (temp != null) {
			algorithm.ackRecieved(temp);
			algorithm.removeAcked();
			if (queue.peek() != null) {
				processAcks(queue);
			}
		}
	}

	private void processAcks(BlockingQueue<UtpTimestampedPacketDTO> queue) throws InterruptedException {
		UtpTimestampedPacketDTO pair;
		while((pair = queue.poll()) != null) {
			algorithm.ackRecieved(pair);
			algorithm.removeAcked();		
		}
	}

	private DatagramPacket getNextPacket() throws IOException {
		int packetSize = algorithm.sizeOfNextPacket();
		int remainingBytes = buffer.remaining();
		
		if (remainingBytes < packetSize) {
			packetSize = remainingBytes;
		}
		
		
		byte[] payload = new byte[packetSize];
		buffer.get(payload);
		UtpPacket utpPacket = channel.getNextDataPacket();
		utpPacket.setPayload(payload);
		
		int leftInBuffer = buffer.remaining();
		if (leftInBuffer > UnsignedTypesUtil.MAX_UINT) {
			leftInBuffer =  (int) (UnsignedTypesUtil.MAX_UINT & 0xFFFFFFFF);
		}
		utpPacket.setWindowSize(leftInBuffer);
//		log.debug("Sending Pkt: " + utpPacket.toString());
		byte[] utpPacketBytes = utpPacket.toByteArray();
		DatagramPacket udpPacket = new DatagramPacket(utpPacketBytes, utpPacketBytes.length, channel.getRemoteAdress());
		algorithm.markPacketOnfly(utpPacket, udpPacket);
		return udpPacket;	
	}


	private void exceptionOccured(IOException exp) {
		possibleException = exp;
	}
	
	public boolean hasExceptionOccured() {
		return possibleException != null;
	}
	
	public IOException getException() {
		return possibleException;
	}
	
	private boolean continueSending() {
		return !graceFullInterrupt && !allPacketsAckedSendAndAcked();
	}
	
	private boolean allPacketsAckedSendAndAcked() {

//		return finSend && algorithm.areAllPacketsAcked() && !buffer.hasRemaining();
		return algorithm.areAllPacketsAcked() && !buffer.hasRemaining();
	}


	public void graceFullInterrupt() {
		graceFullInterrupt = true;
	}

	public int getBytesSend() {
		return buffer.position();
	}
	
	public boolean isRunning() {
		return isRunning;
	}
}
 public class MicroSecondsTimeStamp {

	private static long initDateMillis = System.currentTimeMillis();
	private static long startNs = System.nanoTime();
	
	/**
	 * Returns a uTP time stamp for packet headers. 
	 * @return timestamp
	 */
	public int utpTimeStamp() {
		int returnStamp;
		//TODO: if performance issues, try bitwise & operator since constant MAX_UINT equals 0xFFFFFF
		// (see http://en.wikipedia.org/wiki/Modulo_operation#Performance_issues )
		long stamp = timeStamp() % UnsignedTypesUtil.MAX_UINT;
		try {
			returnStamp  = longToUint(stamp);
		} catch (ByteOverflowException exp) {
			stamp = stamp % MAX_UINT;
			returnStamp = longToUint(stamp);
		}
		return returnStamp;
	}
	
	/**
	 * Calculates the Difference of the uTP timestamp between now and parameter. 
	 * @param othertimestamp timestamp
	 * @return difference 
	 */
	public int utpDifference(int othertimestamp) {
		return utpDifference(utpTimeStamp(), othertimestamp);
	}
	/**
	 * calculates the utp Difference of timestamps (this - other)
	 * @param thisTimeStamp
	 * @param othertimestamp
	 * @return difference. 
	 */
	public int utpDifference(int thisTimeStamp, int othertimestamp) {
		int nowTime = thisTimeStamp;
		long nowTimeL = nowTime & 0xFFFFFFFF;
		long otherTimeL = othertimestamp & 0xFFFFFFFF;
		long differenceL = nowTimeL - otherTimeL;
		// TODO: POSSIBLE BUG NEGATIVE DIFFERENCE
		if (differenceL < 0) {
			differenceL += MAX_UINT;
		}
		return longToUint(differenceL);
	}
	

	/**
	 * @return timestamp with micro second resulution
	 */
	public long timeStamp() {
		
		long currentNs = System.nanoTime();
		long deltaMs = (currentNs - startNs)/1000;
		return (initDateMillis*1000 + deltaMs);
	}
	
	
	public long getBegin() {
		return (initDateMillis * 1000);
	}
	
	public long getStartNs() {
		return startNs;
	}
	
	public long getInitDateMs() {
		return initDateMillis;
	}

}
 public class SelectiveAckHeaderExtension extends UtpHeaderExtension {

	private byte nextExtension;
	private byte[] bitMask;
	
	/* Bit mappings */
	public static byte[] BITMAP = { 1, 2, 4, 8, 16, 32, 64, (byte) 128};
	
	/**
	 * true if the ACK+number-th bit is set to 1 in the bitmask
	 * @param bitmask bitmask
	 * @param number number
	 * @return true if bit is set, otherwise false. 
	 */
	public static boolean isBitMarked(byte bitmask, int number) {
		if (number < 2 || number > 9 ) {
			return false;
		} else {
			boolean returnvalue = (BITMAP[number - 2] & bitmask) == BITMAP[number - 2];
			return returnvalue;
		}
	}
	
	@Override
	public byte getNextExtension() {
		return nextExtension;
	}
	@Override
	public void setNextExtension(byte nextExtension) {
		this.nextExtension = nextExtension;
	}
	@Override
	public byte getLength() {
		return longToUbyte(bitMask.length);
	}

	@Override
	public byte[] getBitMask() {
		return bitMask;
	}
	@Override
	public void setBitMask(byte[] bitMask) {
		this.bitMask = bitMask;
	}
	
	@Override
	public boolean equals(Object obj) {
		if (obj == null) {
			return false;
		} 
		if (this == obj) {
			return true;
		}
		if (!(obj instanceof SelectiveAckHeaderExtension)) {
			return false;
		} 
		SelectiveAckHeaderExtension s = (SelectiveAckHeaderExtension) obj;
		return Arrays.equals(toByteArray(), s.toByteArray());
	}
	
	@Override
	public byte[] toByteArray() {
		//TODO: not create a new byte array
		byte[] array = new byte[2 + bitMask.length];
		array[0] = nextExtension;
		array[1] = longToUbyte(bitMask.length);
		for (int i = 0; i < bitMask.length; i++) {
			array[i + 2] = bitMask[i];
		}
		return array;
	}
}
 public abstract class UtpHeaderExtension {
	
	public static UtpHeaderExtension resolve(byte b) {
		if (b == longToUbyte(1)) {
			return new SelectiveAckHeaderExtension();
		} else {
			return null;
		}
	}

	public abstract byte getNextExtension();
	public abstract void setNextExtension(byte nextExtension) ;
	
	public abstract byte getLength();
	
	public abstract byte[] getBitMask();
	public abstract void setBitMask(byte[] bitMask);
	
	public abstract byte[] toByteArray();

}
 public class UtpPacket {
	
	private byte typeVersion;
	private byte firstExtension;
	private short connectionId;
	private int timestamp;
	private int timestampDifference;
	private int windowSize;
	private short sequenceNumber;
	private short ackNumber;
	private UtpHeaderExtension[] extensions;
	private byte[] payload;
	
	
	public byte[] getPayload() {
		return payload;
	}
	
	public void setPayload(byte[] payload) {
		this.payload = payload;
	}
	
	public void setExtensions(UtpHeaderExtension[] extensions) {
		this.extensions = extensions;
	}
	
	public UtpHeaderExtension[] getExtensions() {
		return this.extensions;
	}
	
	public byte getFirstExtension() {
		return firstExtension;
	}
	
	public void setFirstExtension(byte firstExtension) {
		this.firstExtension = firstExtension;
	}
	
	public byte getTypeVersion() {
		return typeVersion;
	}

	public void setTypeVersion(byte typeVersion) {
		this.typeVersion = typeVersion;
	}

	public short getConnectionId() {
		return connectionId;
	}

	public void setConnectionId(short connectionId) {
		this.connectionId = connectionId;
	}

	public int getTimestamp() {
		return timestamp;
	}

	public void setTimestamp(int timestamp) {
		this.timestamp = timestamp;
	}

	public int getTimestampDifference() {
		return timestampDifference;
	}

	public void setTimestampDifference(int timestampDifference) {
		this.timestampDifference = timestampDifference;
	}

	public int getWindowSize() {
		return windowSize;
	}

	public void setWindowSize(int windowSize) {
		this.windowSize = windowSize;
	}

	public short getSequenceNumber() {
		return sequenceNumber;
	}

	public void setSequenceNumber(short sequenceNumber) {
		this.sequenceNumber = sequenceNumber;
	}

	public short getAckNumber() {
		return ackNumber;
	}

	public void setAckNumber(short ackNumber) {
		this.ackNumber = ackNumber;
	}

	/**
	 * Returns a byte array version of this packet.
	 */
	public byte[] toByteArray() {
		
		if (!hasExtensions()) {
			return joinByteArray(getExtensionlessByteArray(), getPayload());
		}
		

		int offset = DEF_HEADER_LENGTH;
		int headerLength = offset + getTotalLengthOfExtensions();
		byte[] header = new byte[headerLength];
		byte[] extensionlessArray = getExtensionlessByteArray();
		
		for (int i = 0; i< extensionlessArray.length; i++) {
			header[i] = extensionlessArray[i];
		}
		
		for (int i = 0; i < extensions.length; i++) {
			byte[] extenionBytes = extensions[i].toByteArray();
			for (int j = 0; j < extenionBytes.length; j++) {
				header[offset++] = extenionBytes[j];
			}
		}
		return joinByteArray(header, getPayload());
		
	}

	private byte[] getExtensionlessByteArray() {
		byte[] array = {typeVersion, firstExtension, (byte) (connectionId >> 8), (byte) (connectionId),
		(byte) (timestamp >> 24), (byte) (timestamp >> 16), (byte) (timestamp >> 8), (byte) (timestamp),
		(byte) (timestampDifference >> 24), (byte) (timestampDifference >> 16), (byte) (timestampDifference >> 8), (byte) (timestampDifference),
		(byte) (windowSize >> 24), (byte) (windowSize >> 16), (byte) (windowSize >> 8), (byte) (windowSize),
		(byte) (sequenceNumber >> 8), (byte) (sequenceNumber), (byte) (ackNumber >> 8), (byte) (ackNumber), };
		return array;
	}

	private boolean hasExtensions() {
		return !((extensions == null || extensions.length == 0) && firstExtension == 0);
	}

	private int getTotalLengthOfExtensions() {
		if (!hasExtensions()) {
			return 0;
		}
		int length = 0;
		for (int i = 0; i < extensions.length; i++) {
			length += 2 + extensions[i].getBitMask().length;
		}
		return length;
	}
	
	/**
	 * returns the total packet length.
	 * @return bytes.
	 */
	public int getPacketLength() {
		byte[] pl = getPayload();
		int plLength = pl != null ? pl.length : 0;
		return DEF_HEADER_LENGTH + getTotalLengthOfExtensions() + plLength;
	}
	
	/**
	 * Sets the packet data from the array
	 * @param array the array
	 * @param length total packet length
	 * @param offset the packet starts here
	 */
	public void setFromByteArray(byte[] array, int length, int offset) {
		if (array == null) {
			return;
		}
		
		typeVersion = array[0];
		firstExtension = array[1];
		connectionId = UnsignedTypesUtil.bytesToUshort(array[2], array[3]);
		timestamp = UnsignedTypesUtil.bytesToUint(array[4], array[5], array[6], array[7]);
		timestampDifference = UnsignedTypesUtil.bytesToUint(array[8], array[9], array[10], array[11]);
		windowSize = UnsignedTypesUtil.bytesToUint(array[12], array[13], array[14], array[15]);
		sequenceNumber = UnsignedTypesUtil.bytesToUshort(array[16], array[17]);
		ackNumber = UnsignedTypesUtil.bytesToUshort(array[18], array[19]);
		
		int utpOffset = offset + UtpPacketUtils.DEF_HEADER_LENGTH;
		if (firstExtension != 0) {
			utpOffset += loadExtensions(array);
		}
		
		payload = new byte[length - utpOffset];
		System.arraycopy(array, utpOffset, payload, 0, length - utpOffset);
		
	}
	
	
	private int loadExtensions(byte[] array) {
		byte extensionType = array[1];
		int extensionStartIndex = 20;
		int totalLength = 0;
		
		ArrayList<UtpHeaderExtension> list = new ArrayList<UtpHeaderExtension>();
		UtpHeaderExtension extension = UtpHeaderExtension.resolve(extensionType);
	
		while (extension != null) {
			int extensionLength = array[extensionStartIndex+1] & 0xFF;
			byte[] bitmask = new byte[extensionLength];
			System.arraycopy(array, extensionStartIndex+2, bitmask, 0, extensionLength);
			extension.setNextExtension(array[extensionStartIndex]);
			extension.setBitMask(bitmask);
			list.add(extension);
			totalLength = extensionLength+2;
			int nextPossibleExtensionIndex = extensionLength+2+extensionStartIndex;
			// packet is enough big
			if (array.length > nextPossibleExtensionIndex) {
				extension = UtpHeaderExtension.resolve(array[nextPossibleExtensionIndex]);
				extensionStartIndex = nextPossibleExtensionIndex;				
			} else { // packet end reached. 
				extension = null;
			}
		}

		UtpHeaderExtension[] extensions = list.toArray(new UtpHeaderExtension[list.size()]);
		setExtensions(extensions);
		return totalLength;
	}



	@Override
	public boolean equals(Object obj) {
		if (obj == null) {
			return false;
		} 
		if (this == obj) {
			return true;
		}
		if (!(obj instanceof UtpPacket)) {
			return false;
		} else {
			UtpPacket pkt = (UtpPacket) obj;
			
			byte[] their = pkt.toByteArray();
			byte[] mine = this.toByteArray();
			
			if (their.length != mine.length) {
				return false;
			}
			for (int i = 0; i < mine.length; i++) {
				if (mine[i] != their[i]) {
					return false;
				}
			}
			
			return true;
		}
		
	}
	@Override
	public String toString() {
		String ret = "[Type: " + (typeVersion & 0xFF) + "] " + 
					"[FirstExt: " + (firstExtension & 0xFF) + "] " +
					"[ConnId: "	+ (connectionId & 0xFFFF) + "] " + 
					"[Wnd: " + (windowSize & 0xFFFFFFFF) + " " + 
					"[Seq: " + (sequenceNumber & 0xFFFF) + "] " + 
					"[Ack: " +	(ackNumber & 0xFFFF) + "] ";
		
		if (extensions != null) {
			for (int i = 0; i< extensions.length; i++) {
				ret += "[Ext_" + i + ": " + (extensions[i].getNextExtension() & 0xFF) + " " + extensions[i].getLength() + "] ";
			}
		}
		return ret;
		

	}
	
	
    @Override
    public int hashCode() {
      int code = 0;
      //TB: multiply with prime and xor
      //TODO: check if hashCode is needed
		code = typeVersion + 10*firstExtension + 100* connectionId 
				+ 1000*timestamp + 10000*timestampDifference + 100000*windowSize
				+ 1000000*sequenceNumber + 10000000*ackNumber + 1000000000*toByteArray().length;
      return code;
    }
	
}
 public class UtpPacketUtils {
	
	public static final byte VERSION = longToUbyte(1);

	public static final byte DATA = (byte) (VERSION | longToUbyte(0));
	public static final byte FIN = (byte) (VERSION | longToUbyte(16));
	public static final byte STATE = (byte) (VERSION | longToUbyte(32));
	public static final byte RESET = (byte) (VERSION | longToUbyte(48));
	public static final byte SYN = (byte) (VERSION | longToUbyte(64));

	
	public static final byte NO_EXTENSION = longToUbyte(0);
	public static final byte SELECTIVE_ACK = longToUbyte(1);
	
	public static final int MAX_UTP_PACKET_LENGTH = 1500;
	public static final int MAX_UDP_HEADER_LENGTH = 48;
	public static final int DEF_HEADER_LENGTH = 20;
	
	private static final Logger log = LoggerFactory.getLogger(UtpPacketUtils.class);
	
	public static byte[] joinByteArray(byte[] array1, byte[] array2) {
		
		int length1 = array1 == null ? 0 : array1.length;
		int length2 = array2 == null ? 0 : array2.length;

		
		int totalLength = length1 + length2;
		byte[] returnArray = new byte[totalLength];
		
		int i = 0;
		for (; i < length1; i++) {
			returnArray[i] = array1[i];
		}
		
		for (int j = 0; j < length2; j++) {
			returnArray[i] = array2[j];
			i++;
		}
		
		return returnArray;
		
	}
	
	/**
	 * Creates an Utp-Packet to initialize a connection
	 * Following values will be set:
	 * <ul>
	 * <li>Type and Version</li>
	 * <li>Sequence Number</li>
	 * </ul>
	 * @return {@link UtpPacket}
	 */
	public static UtpPacket createSynPacket() {
		
		UtpPacket pkt = new UtpPacket();
		pkt.setTypeVersion(SYN);
		pkt.setSequenceNumber(longToUbyte(1));
		byte[] pl = { 1, 2, 3, 4, 5, 6};
		pkt.setPayload(pl);
		return pkt;
	}
	
	public static UtpPacket extractUtpPacket(DatagramPacket dgpkt) {
		
		UtpPacket pkt = new UtpPacket();
		byte[] pktb = dgpkt.getData();
		pkt.setFromByteArray(pktb, dgpkt.getLength(), dgpkt.getOffset());
		return pkt;
	}

	public static boolean isSynPkt(UtpPacket packet) {
		
		if (packet == null) {
			return false;
		}
		
		return packet.getTypeVersion() == SYN;
		
	}
	
	
	private static boolean isPacketType(DatagramPacket packet, byte flag) {
		if (packet == null) {
			return false;
		}
		
		byte[] data = packet.getData();
		
		if (data != null && data.length >= DEF_HEADER_LENGTH) {
			return data[0] == flag;
		}
		
		return false;
	}

	public static boolean isSynPkt(DatagramPacket packet) {
		return isPacketType(packet, SYN);
	}

	public static boolean isResetPacket(DatagramPacket udpPacket) {
		return isPacketType(udpPacket, RESET);
	}

	public static boolean isDataPacket(DatagramPacket udpPacket) {
		return isPacketType(udpPacket, DATA);
	}

	public static boolean isStatePacket(DatagramPacket udpPacket) {
		return isPacketType(udpPacket, STATE);
	}

	public static boolean isFinPacket(DatagramPacket udpPacket) {
		return isPacketType(udpPacket, FIN);
	}

}

import static net.utp4j.channels.UtpSocketState.CLOSED;
import static net.utp4j.channels.UtpSocketState.CONNECTED;
import static net.utp4j.channels.UtpSocketState.SYN_ACKING_FAILED;
import static net.utp4j.data.UtpPacketUtils.FIN;
import static net.utp4j.data.UtpPacketUtils.SELECTIVE_ACK;
import static net.utp4j.data.UtpPacketUtils.STATE;
import static net.utp4j.data.UtpPacketUtils.extractUtpPacket;
import static net.utp4j.data.UtpPacketUtils.isDataPacket;
import static net.utp4j.data.UtpPacketUtils.isFinPacket;
import static net.utp4j.data.UtpPacketUtils.isResetPacket;
import static net.utp4j.data.UtpPacketUtils.isStatePacket;
import static net.utp4j.data.UtpPacketUtils.isSynPkt;
import static net.utp4j.data.bytes.UnsignedTypesUtil.MAX_USHORT;
import static net.utp4j.data.bytes.UnsignedTypesUtil.longToUint;
import static net.utp4j.data.bytes.UnsignedTypesUtil.longToUshort;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.DatagramSocket;
import java.net.SocketAddress;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.SortedMap;
import java.util.TreeMap;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import net.utp4j.channels.UtpSocketChannel;
import net.utp4j.channels.UtpSocketState;
import net.utp4j.channels.futures.UtpCloseFuture;
import net.utp4j.channels.futures.UtpWriteFuture;
import net.utp4j.channels.impl.alg.UtpAlgConfiguration;
import net.utp4j.channels.impl.conn.ConnectionTimeOutRunnable;
import net.utp4j.channels.impl.conn.UtpConnectFutureImpl;
import net.utp4j.channels.impl.read.UtpReadFutureImpl;
import net.utp4j.channels.impl.read.UtpReadingRunnable;
import net.utp4j.channels.impl.recieve.UtpPacketRecievable;
import net.utp4j.channels.impl.recieve.UtpRecieveRunnable;
import net.utp4j.channels.impl.write.UtpWriteFutureImpl;
import net.utp4j.channels.impl.write.UtpWritingRunnable;
import net.utp4j.data.MicroSecondsTimeStamp;
import net.utp4j.data.SelectiveAckHeaderExtension;
import net.utp4j.data.UtpHeaderExtension;
import net.utp4j.data.UtpPacket;
import net.utp4j.data.UtpPacketUtils;

		} else {
			sendResetPacket(udpPacket.getSocketAddress());
		}
	}

	private void sendResetPacket(SocketAddress addr) {
		log.debug("Sending RST packet");

	}
	
	private boolean acceptSyn(DatagramPacket udpPacket) {
		UtpPacket pkt = extractUtpPacket(udpPacket);
		return getState() == CLOSED 
				|| (getState() == CONNECTED && isSameAddressAndId(
						pkt.getConnectionId(), udpPacket.getSocketAddress()));
	}

	@Override
	protected void setAckNrFromPacketSqNr(UtpPacket utpPacket) {
		short ackNumberS = utpPacket.getSequenceNumber();
		setAckNumber(ackNumberS & 0xFFFF);
	}

	private void setConnectionIdsFromPacket(UtpPacket utpPacket) {
		short connId = utpPacket.getConnectionId();
		int connIdSender = (connId & 0xFFFF);
		int connIdRec = (connId & 0xFFFF) + 1;
		setConnectionIdsending(connIdSender);
		setConnectionIdRecieving(connIdRec);

	}

	@Override
	protected void connectImpl(UtpConnectFutureImpl future) {
		reciever = new UtpRecieveRunnable(getDgSocket(), this);
		reciever.start();

	}

	@Override
	protected void abortImpl() {
		if (reciever != null) {
			reciever.graceFullInterrupt();
		} else if (server != null) {
			server.unregister(this);
		}
	}

	@Override
	public UtpWriteFuture write(ByteBuffer src) {
		UtpWriteFutureImpl future = null;
		try {
			future = new UtpWriteFutureImpl();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		writer = new UtpWritingRunnable(this, src, timeStamper, future);
		writer.start();
		return future;
	}

	public BlockingQueue<UtpTimestampedPacketDTO> getDataGramQueue() {
		return queue;
	}
	
	/**
	 * Returns a Data packet with specified header fields already set. 
	 * @return
	 */
	public UtpPacket getNextDataPacket() {
		return createDataPacket();
	}

	/**
	 * Returns predefined fin packet
	 * @return fin packet.
	 */
	public UtpPacket getFinPacket() {
		UtpPacket fin = createDataPacket();
		fin.setTypeVersion(FIN);
		return fin;
	}

	@Override
	public UtpReadFutureImpl read(ByteBuffer dst) {
		UtpReadFutureImpl readFuture = null;
		try {
			readFuture = new UtpReadFutureImpl();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		reader = new UtpReadingRunnable(this, dst, timeStamper, readFuture);
		reader.start();
		return readFuture;
	}
	
	/**
	 * Creates an Selective Ack packet
	 * @param headerExtension the header extension where the SACK data is stored
	 * @param timestampDifference timestamp difference for the ack pcket.
	 * @param advertisedWindow remaining buffer size.
	 * @throws IOException
	 */
	public void selectiveAckPacket(SelectiveAckHeaderExtension headerExtension,
			int timestampDifference, long advertisedWindow) throws IOException {
		UtpPacket sack = createSelectiveAckPacket(headerExtension,
				timestampDifference, advertisedWindow);
		sendPacket(sack);

	}

	private UtpPacket createSelectiveAckPacket(SelectiveAckHeaderExtension headerExtension,
			int timestampDifference, long advertisedWindow) {
		UtpPacket ackPacket = new UtpPacket();
		ackPacket.setAckNumber(longToUshort(getAckNumber()));
		ackPacket.setTimestampDifference(timestampDifference);
		ackPacket.setTimestamp(timeStamper.utpTimeStamp());
		ackPacket.setConnectionId(longToUshort(getConnectionIdsending()));
		ackPacket.setWindowSize(longToUint(advertisedWindow));


		ackPacket.setFirstExtension(SELECTIVE_ACK);
