/*
 * Written by Dawid Kurzyniec and released to the public domain, as explained
 * at http://creativecommons.org/licenses/publicdomain
 */

package edu.emory.mathcs.util.net.inproc;

import java.io.*;
import java.net.*;
import java.util.*;

import edu.emory.mathcs.util.io.*;
import edu.emory.mathcs.backport.java.util.concurrent.*;
import edu.emory.mathcs.util.concurrent.*;

/**
 * Abstraction of a server socket which can be accessed only from within a
 * process. While this
 * class fully adheres to the socket API, it is a server socket that accepts
 * connections only from appropriate {@link InProcSocket "sockets"} within the
 * same process. The class can be used to create local in-process bindings
 * within APIs that assume remote access. For instance, when used as an RMI
 * transport, in-process sockets can interconnect local objects while
 * maintaining remote invocation semantics (pass-by-value etc.) yet avoiding
 * security risks associated with network sockets and offering a bit better
 * performance than a loopback network interface.
 *
 * @see InProcSocket
 *
 * @author Dawid Kurzyniec
 * @version 1.0
 */
public abstract class InProcServerSocket extends ServerSocket {

    final static Map bindings = new HashMap();
    final static Random random = new Random();

    boolean bound;
    volatile boolean closed;
    int localPort;
    int soTimeout = 0;

    BlockingQueue connectionQueue;

    public InProcServerSocket() throws IOException {
        super();
    }

    public InProcServerSocket(int port) throws IOException {
        this(port, 50);
    }

    public InProcServerSocket(int port, int backlog) throws IOException {
        super();
        bind(new InProcSocketAddress(port), backlog);
    }

    public InProcServerSocket(InProcSocketAddress addr) throws IOException {
        this(addr, 50);
    }

    public InProcServerSocket(InProcSocketAddress addr, int backlog) throws IOException {
        super();
        bind(addr, backlog);
    }

    public synchronized void bind(SocketAddress endpoint, int backlog) throws IOException {
        ensureNotClosed();
        if (isBound()) throw new SocketException("Already bound");
        if (endpoint == null) endpoint = new InProcSocketAddress(0);
        if (backlog <= 0) throw new IllegalArgumentException("Backlog must be positive");
        if (!(endpoint instanceof InProcSocketAddress))
            throw new IllegalArgumentException("Unsupported address type");
        int port = ((InProcSocketAddress)endpoint).port;
        try {
            /** @todo security (checkListen) check? */
            synchronized (bindings) {
                if (port != 0) {
                    if (bindings.containsKey(new Integer(port))) {
                        throw new BindException("InProc port " + port + " already in use");
                    }
                    else {
                        // bail out to bindins.put
                    }
                }
                else {
                    boolean ok = false;
                    for (int i=0; i<100; i++) {
                        port = random.nextInt() & 0x7FFFFFFF;
                        // try to fit into the 2-byte range
                        if (i<4) port = port & 0x3FFF;
                        else if (i<8) port = port & 0xFFFF;
                        if (port > 1024 && !bindings.containsKey(new Integer(port))) {
                            ok = true;
                            break;
                        }
                    }
                    if (!ok) {
                        throw new BindException("InProc: Could not find available port to " +
                                                "listen on");
                    }
                }

                // at this point, we have established an available port number
                bindings.put(new Integer(port), this);
            }
            localPort = port;
            bound = true;
            connectionQueue = new DynamicArrayBlockingQueue(4, backlog);
        }
        catch (SecurityException e) {
            bound = false;
            throw e;
        }
        catch (IOException e) {
            bound = false;
            throw e;
        }
    }

    public boolean isClosed() {
        return closed;
    }

    public boolean isBound() {
        return bound;
    }

    public InetAddress getInetAddress() {
        // not an Inet socket, but for security managers it is better to report
        // that it is a "local" socket
        return InProcSocket.inprocInetAddr;
    }

    public int getLocalPort() {
        if (!isBound()) return -1;
        return localPort;
    }

    public SocketAddress getLocalSocketAddress() {
        if (!isBound()) return null;
        return new InProcSocketAddress(getLocalPort());
    }
//
//    public Socket accept() throws IOException {
//        Socket socket = delegate.accept();
//        return wrapAcceptedSocket(socket);
//    }
//

    public Socket accept() throws IOException {
        ensureNotClosed();
        if (!isBound()) throw new SocketException("Socket is not bound yet");
        int soTimeout = this.soTimeout;
        ConnReq req;
        InProcSocket.Channel ch;
        do {
            try {
                if (soTimeout > 0) {
                    req = (ConnReq)connectionQueue.poll(soTimeout, TimeUnit.MILLISECONDS);
                }
                else {
                    req = (ConnReq)connectionQueue.take();
                }
            }
            catch (InterruptedException e) {
                throw new InterruptedIOException(e.toString());
            }
            if (req == null) {
                throw new SocketTimeoutException("Timeout on InProcServerSocket.accept");
            }
            if (req == TERMINATOR) {
                throw new SocketException("Socket closed");
            }
            ch = req.accept();
        }
        while (ch == null);
        return new InProcSocket(ch, localPort);
    }


    private static final ConnReq TERMINATOR = new ConnReq();

    public synchronized void close() throws IOException {
        closed = true;
        while (true) {
            ConnReq connReq = (ConnReq)connectionQueue.poll();
            if (connReq == null) break;
            connReq.refuse();
        }
        // abort possibly blocked accept
        try {
            connectionQueue.put(TERMINATOR);
        }
        catch (InterruptedException e) {
            throw new RuntimeException("FATAL: Blocked when putting into empty queue");
        }
        if (isBound()) {
            synchronized (bindings) {
                bindings.remove(new Integer(localPort));
            }
        }
    }

//    public ServerSocketChannel getChannel() {
//        return null;
//    }
//
    public void setSoTimeout(int timeout) throws SocketException {
        if (timeout < 0) throw new IllegalArgumentException("Timeout must be non-negative");
        ensureNotClosed();
        soTimeout = timeout;
    }

    public int getSoTimeout() throws IOException {
        ensureNotClosed();
        return soTimeout;
    }

    public void setReuseAddress(boolean on) throws SocketException {
        ensureNotClosed();
        // no op
    }

    public boolean getReuseAddress() throws SocketException {
        ensureNotClosed();
        return true;
    }

    public String toString() {
        if (!isBound()) return "InProcServerSocket[unbound]";
        return "InProcServerSocket[port=" + localPort + "]";
    }

//    public void setReceiveBufferSize (int size) throws SocketException {
//        delegate.setReceiveBufferSize(size);
//    }
//
//    public int getReceiveBufferSize() throws SocketException{
//        return delegate.getReceiveBufferSize();
//    }
//

    private void ensureNotClosed() throws SocketException {
        if (isClosed()) throw new SocketException("Socket is closed");
    }

    static InProcSocket.Channel connect(int port, int timeout) throws IOException {
        InProcServerSocket srvsock;
        synchronized (bindings) {
            srvsock = (InProcServerSocket)bindings.get(new Integer(port));
        }
        if (srvsock == null) {
            throw new IOException("Connection refused (server not listening)");
        }
        boolean success;
        ConnReq connreq = new ConnReq();
        synchronized (srvsock) {
            if (srvsock.isClosed()) throw new IOException("Connection refused (server closed)");
            success = srvsock.connectionQueue.offer(connreq);
        }
        if (!success) {
            throw new IOException("Connection refused (queue full, try again later)");
        }
        return connreq.awaitOrCancel(timeout);
    }

    private static class ConnReq {
        boolean accepted  = false;
        boolean cancelled = false;
        boolean refused = false;
        InProcSocket.Channel srvChannel;
        InProcSocket.Channel cliChannel;

        ConnReq() {}
        public synchronized boolean cancel() {
            if (accepted || refused) return false;
            cancelled = true;
            notifyAll();
            return true;
        }
        public synchronized InProcSocket.Channel accept() {
            if (cancelled) return null;
            if (accepted || refused) throw new IllegalStateException("Already responded");
            accepted = true;
            /** @todo bufsize */
            BufferedPipe upstream = new BufferedPipe();
            BufferedPipe downstream = new BufferedPipe();
            srvChannel = new InProcSocket.Channel((TimedInput)upstream.sink(),
                                                 downstream.source());
            cliChannel = new InProcSocket.Channel((TimedInput)downstream.sink(),
                                                 upstream.source());
            notifyAll();
            return srvChannel;
        }
        public synchronized void refuse() {
            if (cancelled) return;
            if (accepted || refused) throw new IllegalStateException("Already responded");
            refused = true;
            notifyAll();
        }
        public synchronized InProcSocket.Channel awaitOrCancel(int timeout) throws IOException {
            try {
                if (timeout == 0) {
                    while (!accepted && !cancelled && !refused) {
                        wait();
                    }
                }
                else {
                    long endtime = timeout + System.currentTimeMillis();
                    while (!accepted && !cancelled && !refused) {
                        long todo = endtime - System.currentTimeMillis();
                        if (todo > 0) {
                            wait(todo);
                        }
                        else {
                            cancel();
                            break;
                        }
                    }
                }

                if (cancelled) throw new SocketTimeoutException("Connection timed out");
                if (refused) throw new IOException("Connection refused (server closing)");
                // otherwise must be accepted
                return cliChannel;
            }
            catch (InterruptedException e) {
                if (accepted) return cliChannel;
                cancel();
                throw new InterruptedIOException("Connect interrupted");
            }
        }
    }
}

