package jp.progressive.rudp;

import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.builder.ReflectionToStringBuilder;

class RudpRemote {
    private int lastSeqNo;// sequence 自分がこれまでに送ったsequence
    private int myAckNo;// 相手から受け取ったsequence 受け取りが完了したもの
    private int remoteAckNo;// acknowledgement 動作完了 相手に送ったものがどこまで伝わったか
    private Map<Integer, byte[]> msgCache;

    RudpRemote() {
	this.lastSeqNo = 0;
	this.myAckNo = 0;
	this.remoteAckNo = 0;
	this.msgCache = Collections
		.synchronizedMap(new HashMap<Integer, byte[]>());
    }

    @Override
    public String toString() {
	return ReflectionToStringBuilder.toString(this);
    }

    synchronized boolean needsSync() {
	synchronized (msgCache) {
	    if (msgCache.isEmpty()) {
		return false;
	    }
	    return true;
	}
    }

    synchronized ByteBuffer getRudpSyncDatagram() {
	assert needsSync();
	return getRudpDatagram();
    }

    synchronized ByteBuffer getRudpDatagram(byte[] nextSequence) {
	assert nextSequence != null;
	lastSeqNo++;
	synchronized (msgCache) {
	    msgCache.put(lastSeqNo, nextSequence);
	}
	return getRudpDatagram();
    }

    private synchronized ByteBuffer getRudpDatagram() {
	int remoteAckNo = this.remoteAckNo;
	ByteBuffer buffer = ByteBuffer
		.allocate(7 + 16 * (lastSeqNo - remoteAckNo));
	assert buffer.capacity() < 65536;
	if (buffer.capacity() > 65536)
	    System.exit(-1);
	// SEQ_NO
	DatagramUtils.putUInt24(buffer, lastSeqNo);
	// ACK_NO
	DatagramUtils.putUInt24(buffer, myAckNo);
	// MESSAGE
	synchronized (msgCache) {
	    for (int i = lastSeqNo; i > remoteAckNo; i--) {
		byte[] seq = msgCache.get(i);
		assert seq != null;
		buffer.put((byte) seq.length);
		buffer.put(seq);
	    }
	}
	buffer.put((byte) '\0');
	buffer.flip();
	return buffer;
    }

    synchronized List<ByteBuffer> getDatagrams(ByteBuffer buffer)
	    throws RemoteDisconnectException {
	// SEQ_NO
	int seqNo = DatagramUtils.getUInt24(buffer);
	if (seqNo == 0) {
	    switch (buffer.get()) {
	    case SpecialtySequenceDefine.DISCONNECT:
		throw new RemoteDisconnectException();
	    default:
		throw new RuntimeException();
	    }
	}
	// ACK_NO
	int ackNo = DatagramUtils.getUInt24(buffer);
	// 相手への送信が確認されたものを削除
	synchronized (msgCache) {
	    for (int i = remoteAckNo + 1; i <= ackNo; i++) {
		msgCache.remove(i);
	    }
	}
	if (remoteAckNo < ackNo) {
	    remoteAckNo = ackNo;
	}
	return getNewBuffers(buffer, seqNo);
    }

    private synchronized List<ByteBuffer> getNewBuffers(ByteBuffer buffer,
	    int newNo) {
	// 新しくないシーケンスは無視
	if (myAckNo >= newNo) {
	    return Collections.emptyList();
	}
	List<ByteBuffer> list = new ArrayList<ByteBuffer>();
	// 新しい物から順にリストに突っ込む
	for (int i = newNo; i > myAckNo; i--) {
	    int length = buffer.get();
	    if (length == 0) {
		// 足りない＝データが異常
		return Collections.emptyList();
	    }
	    ByteBuffer buf = buffer.slice();
	    buf.limit(length);
	    list.add(buf);
	    buffer.position(buffer.position() + length);
	}
	myAckNo = newNo;
	return list;
    }

    public static class RemoteDisconnectException extends Exception {
	private static final long serialVersionUID = -6709853433771244102L;
    }
}
