package jp.progressive.rudp;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.apache.commons.collections15.Closure;
import org.apache.commons.lang3.Pair;

public class Receiver {
    private RudpSocket rudp;
    private ReceivedCallback callback;
    private Thread syncThread;
    private Thread receiveThread;

    Receiver(RudpSocket rudp, ReceivedCallback callback) {
	this.rudp = rudp;
	this.callback = callback;
	ThreadFactory factory = Executors.defaultThreadFactory();
	this.syncThread = factory.newThread(new SyncRunner());
	this.receiveThread = factory.newThread(new ReceiveRunner());
    }

    public void start() {
	syncThread.start();
	receiveThread.start();
    }

    public void shutdown() throws IOException {
	syncThread.interrupt();
	receiveThread.interrupt();
	try {
	    syncThread.join();
	    receiveThread.join();
	} catch (InterruptedException e) {
	}
    }

    class SyncRunner implements Runnable {
	public void run() {
	    try {
		while (!Thread.interrupted()) {
		    rudp.sync();
		    Thread.sleep(3000);
		}
		rudp.sync();
	    } catch (ClosedChannelException e) {
		// ソケットが閉じた
	    } catch (IOException e) {
		e.printStackTrace();
	    } catch (InterruptedException e) {
	    }
	}
    }

    class ReceiveRunner implements Runnable {
	public void run() {
	    try {
		while (!Thread.interrupted()) {
		    callback.execute(rudp.receive());
		}
	    } catch (ClosedChannelException e) {
		// ソケットが閉じた
	    } catch (IOException e) {
		e.printStackTrace();
	    }
	}
    }

    public static interface ReceivedCallback extends
	    Closure<Pair<SocketAddress, List<ByteBuffer>>> {
    }
}
