source: simplerunner/src/main/java/geniusweb/simplerunner/BasicConnection.java@ 52

package geniusweb.simplerunner;

import java.io.IOException;
import java.net.URI;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;

import geniusweb.connection.ConnectionEnd;
import geniusweb.references.Reference;
import tudelft.utilities.listener.Listener;

/**
 * A basic connection that implements connection with direct calls
 * 
 *
 * @param <IN>  the type of the incoming data
 * @param <OUT> the type of outgoing data
 */
public class BasicConnection<IN, OUT> implements ConnectionEnd<IN, OUT> {
        private List<Listener<IN>> listeners = new CopyOnWriteArrayList<Listener<IN>>();

        private final Info<OUT> EOS = new EOS<OUT>();
        private final Reference reference;
        private final URI uri;
        // to be initialized
        private Listener<OUT> handler = null;
        private ArrayBlockingQueue<Info<OUT>> messages = new ArrayBlockingQueue<>(
                        4);
        private Throwable error;

        /**
         * 
         * @param reference Reference that was used to create this connection.
         * @param uri       the URI of the remote endpoint that makes up the
         *                  connection. This is a URI that uniquely identifies the
         *                  remote object
         */
        public BasicConnection(Reference reference, URI uri) {
                this.reference = reference;
                this.uri = uri;
        }

        /**
         * To be called to hook up the other side that will handle a send action
         * from us. Must be called first.
         * 
         * @param newhandler a Listener&lt;OUT&gt; that can handle send actions.
         * 
         */
        public void init(Listener<OUT> newhandler) {
                if (handler != null) {
                        throw new IllegalStateException("already initialized");
                }
                this.handler = newhandler;

                Thread handlerThread = new Thread(new Runnable() {

                        @Override
                        public void run() {
                                try {
                                        while (true) {
                                                Info<OUT> mess = messages.take();
                                                if (mess instanceof EOS)
                                                        break;
                                                handler.notifyChange(((Data<OUT>) mess).get());
                                        }
                                } catch (Exception e) {
                                        setError(e);
                                }
                                handler = null;
                                System.out.println("BasicConnection closed");
                        }
                });
                handlerThread.start();
        }

        /**
         * Error condition occurs. Record error and close connection
         * 
         * @param e
         */
        protected synchronized void setError(Throwable e) {
                if (error == null) {
                        // maybe log instead?
                        e.printStackTrace();
                        error = e;
                        close();
                }
        }

        @Override
        public void send(OUT data) throws IOException {
                if (handler == null) {
                        throw new IllegalStateException(
                                        "BasicConnection has not been initialized or was closed");
                }
                try {
                        messages.put(new Data<OUT>(data));
                } catch (InterruptedException e) {
                        setError(e);
                }
        }

        @Override
        public Reference getReference() {
                return reference;
        }

        @Override
        public URI getRemoteURI() {
                return uri;
        }

        @Override
        public void close() {
                System.out.println("flushing and terminating " + this);
                if (handler == null || messages.contains(EOS))
                        return;
                try {
                        messages.put(EOS);
                } catch (InterruptedException e) {
                        setError(e);
                }
        }

        @Override
        public String toString() {
                return "BasicConnection[" + reference + "]";
        }

        @Override
        public Throwable getError() {
                return error;
        }

        /**
         * 
         * @return true iff this connection is open. Returns false also when then
         *         handler is in the close-down process
         */
        public boolean isOpen() {
                return handler != null && !messages.contains(EOS);
        }

        /****************** implements listenable ****************/
        // override because notifyListeners should throw exceptions.
        @Override
        public void addListener(Listener<IN> l) {
                listeners.add(l);
        }

        @Override
        public void removeListener(Listener<IN> l) {
                listeners.remove(l);
        }

        public void notifyListeners(IN data) {
                for (Listener<IN> l : listeners) {
                        l.notifyChange(data);
                }
        }

}

/**
 * Wrapper around data so that we can put Null end EOS in a
 * {@link ArrayBlockingQueue}
 *
 * @param <S> the type of contained data.
 */

interface Info<S> {

}

class Data<S> implements Info<S> {
        private final S data;

        public Data(S data) {
                this.data = data;
        }

        public S get() {
                return data;
        }
}

/**
 * End of stream.
 * 
 * @author wouter
 *
 */
class EOS<S> implements Info<S> {
}