package jp.progressive.rudp;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import jp.progressive.rudp.Receiver.ReceivedCallback;
import jp.progressive.rudp.RudpRemote.RemoteDisconnectException;

import org.apache.commons.lang3.Pair;

// RUDP実装
/**
 * スレッドセーフ
 */
public class RudpSocket {
    private UdpSocket udp;
    private Map<SocketAddress, RudpRemote> remotes;

    public RudpSocket() {
	this.udp = new UdpSocket();
	this.remotes = new HashMap<SocketAddress, RudpRemote>();
    }

    public void open(int port) throws IOException {
	udp.open(port);
    }

    public Receiver getReceiver(ReceivedCallback callback) {
	return new Receiver(this, callback);
    }

    public boolean isOpen() {
	return udp.isOpen();
    }

    public void send(byte[] bytes, SocketAddress remote) throws IOException {
	synchronized (this) {
	    udp.send(getRudpRemote(remote).getRudpDatagram(bytes), remote);
	}
    }

    public boolean needsSync() {
	synchronized (remotes) {
	    for (Entry<SocketAddress, RudpRemote> remote : remotes.entrySet()) {
		if (remote.getValue().needsSync()) {
		    return true;
		}
	    }
	}
	return false;
    }

    // syncスレッド
    public void sync() throws IOException {
	synchronized (remotes) {
	    for (Entry<SocketAddress, RudpRemote> remote : remotes.entrySet()) {
		RudpRemote rudpRemote = remote.getValue();
		if (rudpRemote.needsSync()) {
		    synchronized (this) {
			udp.send(rudpRemote.getRudpSyncDatagram(),
				remote.getKey());
		    }
		}
	    }
	}
    }

    // receiveスレッド
    /**
     * 取得するまでブロックされる
     */
    public Pair<SocketAddress, List<ByteBuffer>> receive() throws IOException {
	Pair<SocketAddress, ByteBuffer> received = udp.receive(); // ブロック
	SocketAddress remote = received.left;
	List<ByteBuffer> list;
	try {
	    list = getRudpRemote(remote).getDatagrams(received.right);
	} catch (RemoteDisconnectException e) {
	    synchronized (remotes) {
		remotes.remove(remote);
	    }
	    return receive();
	}
	if (list.isEmpty()) {
	    return receive();
	}
	return new Pair<SocketAddress, List<ByteBuffer>>(remote, list);
    }

    public void shutdown() throws IOException {
	synchronized (remotes) {
	    for (SocketAddress remote : remotes.keySet()) {
		udp.send(
			ByteBuffer.wrap(new byte[] { 0, 0, 0,
				SpecialtySequenceDefine.DISCONNECT }), remote);
	    }
	}
	udp.close();
    }

    private final RudpRemote getRudpRemote(SocketAddress addr) {
	synchronized (remotes) {
	    if (!remotes.containsKey(addr)) {
		remotes.put(addr, new RudpRemote());
	    }
	    return remotes.get(addr);
	}
    }
}
