path: root/src/org/gnunet/util/Server.java

/*
 This file is part of GNUnet.
 (C) 2011, 2012 Christian Grothoff (and other contributing authors)

 GNUnet is free software; you can redistribute it and/or modify
 it under the terms of the GNU General Public License as published
 by the Free Software Foundation; either version 3, or (at your
 option) any later version.

 GNUnet is distributed in the hope that it will be useful, but
 WITHOUT ANY WARRANTY; without even the implied warranty of
 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 General Public License for more details.

 You should have received a copy of the GNU General Public License
 along with GNUnet; see the file COPYING.  If not, write to the
 Free Software Foundation, Inc., 59 Temple Place - Suite 330,
 Boston, MA 02111-1307, USA.
 */

package org.gnunet.util;

import org.grothoff.Runabout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;

public class Server {
    private static final Logger logger = LoggerFactory
            .getLogger(Server.class);

    private final RelativeTime idleTimeout;
    private final boolean requireFound;
    private List<ServerSocketChannel> listenSockets;
    private List<ClientHandle> clients = new LinkedList<ClientHandle>();

    private MessageRunabout receivedMessagehandler;
    private List<DisconnectHandler> disconnectHandlers = new LinkedList<DisconnectHandler>();
    private ArrayList<Class> expectedMessages;

    private boolean shutdownRequested;
    private Cancelable acceptTask;

    public interface DisconnectHandler {
        void onDisconnect(ClientHandle clientHandle);
    }


    public class ClientHandle {
        private RelativeTime clientTimeout;
        private Connection connection;

        private int referenceCount = 0;
        private Connection.ReceiveHandle currentReceive;

        private boolean isMonitor;

        private ClientHandle(SocketChannel accept) {
            connection = new Connection(accept);
            clientTimeout = idleTimeout;
            // start receiving
            receiveDone(true);
        }


        /**
         * Notify us when the server has enough space to transmit
         * a message of the given size to the given client.
         *
         * @param size        requested amount of buffer space
         * @param timeout     after how long should we give up (and call
         *                    notify with buf NULL and size 0)?
         * @param transmitter callback
         * @return a handle to cancel the notification
         */
        public Cancelable notifyTransmitReady(int size, RelativeTime timeout, MessageTransmitter transmitter) {
            return connection.notifyTransmitReady(0, timeout, transmitter);
        }

        /**
         * Resume receiving from this client, we are done processing the
         * current request.  This function must be called from within each
         * message handler (or its respective continuations).
         * <p/>
         * The server does not automatically continue to receive messages to
         * support flow control.
         *
         * @param keepClient false if connection to the client should be closed
         */
        public void receiveDone(boolean keepClient) {
            if (keepClient) {
                currentReceive = connection.receive(RelativeTime.FOREVER, new MessageReceiver() {
                    @Override
                    public void process(GnunetMessage.Body msg) {
                        if (msg instanceof UnknownMessageBody) {
                            if (requireFound) {
                                logger.info("disconnecting client sending unknown message");
                                disconnect();
                            }
                            // otherwise, just ignore it
                        }
                        if (receivedMessagehandler == null) {
                            throw new AssertionError("received message, but no handler installed");
                        }
                        receivedMessagehandler.setSender(ClientHandle.this);
                        receivedMessagehandler.visitAppropriate(msg);
                    }

                    @Override
                    public void handleError() {
                        logger.warn("error receiving from client");
                        disconnect();
                    }
                });
            } else {
                disconnect();
            }

        }

        /**
         * Change the idle timeout of this particular client.
         */
        public void setTimeout(RelativeTime newTimeout) {
        }

        public void disconnect() {
            connection.disconnect();
            Server.this.clients.remove(this);
            for (DisconnectHandler dh : disconnectHandlers) {
                dh.onDisconnect(this);
            }
        }

        /**
         * Disable the warning the server issues if a message is not acknowledged
         * in a timely fashion.  Use this call if a client is intentionally delayed
         * for a while.  Only applies to the current message.
         */
        public void disableReceiveDoneWarning() {
            // todo
        }

        public void keep() {
            referenceCount++;
        }

        public void drop() {
            referenceCount--;
            if (referenceCount == 0 && shutdownRequested) {
                disconnect();
            }
        }

        public void markMonitor() {
            this.isMonitor = true;
        }
    }


    abstract static class MessageRunabout extends Runabout {
        private ClientHandle currentSender;

        /**
         * Allows implementors of MessageRunabout to get the Client that sent the message
         * currently visited.
         *
         * @return handle of the client whose message is currently being visited
         */
        public final ClientHandle getSender() {
            return currentSender;
        }

        private void setSender(ClientHandle clientHandle) {
            currentSender = clientHandle;
        }
    }


    private void doAccept(final ServerSocketChannel srv) {
        Scheduler.TaskConfiguration b = new Scheduler.TaskConfiguration(RelativeTime.FOREVER,
                new Scheduler.Task() {
                    @Override
                    public void run(Scheduler.RunContext ctx) {
                        acceptTask = null;
                        try {
                            SocketChannel cli = srv.accept();

                            if (cli != null) {
                                logger.debug("client connected");
                                cli.configureBlocking(false);
                                ClientHandle clientHandle = new ClientHandle(cli);
                                clients.add(clientHandle);
                            }

                        } catch (IOException e) {
                            throw new RuntimeException("accept failed", e);
                        }
                        doAccept(srv);
                    }
                });
        b.selectAccept(srv);
        acceptTask = b.schedule();
    }


    /**
     * Create a server listening on all specified addresses.
     *
     * @param addresses    addresses to bind on
     * @param idleTimeout  time after a client will be disconnected if idle
     * @param requireFound allow unknown messages to be received without disconnecting the client in response
     */
    public Server(List<SocketAddress> addresses, RelativeTime idleTimeout, boolean requireFound) {
        this.idleTimeout = idleTimeout;
        this.requireFound = requireFound;
        listenSockets = new ArrayList<ServerSocketChannel>(addresses.size());
        try {
            for (SocketAddress addr : addresses) {
                ServerSocketChannel socket = ServerSocketChannel.open();
                socket.configureBlocking(false);
                socket.socket().bind(addr);
                logger.debug("socket listening on {}", addr.toString());
                listenSockets.add(socket);
                doAccept(socket);
            }
        } catch (IOException e) {
            throw new RuntimeException("could not bind");
        }
    }

    public Server(RelativeTime idleTimeout, boolean requireFound) {
        this.idleTimeout = idleTimeout;
        this.requireFound = requireFound;
    }

    public void addAcceptSocket(ServerSocketChannel sock) {
        doAccept(sock);
    }

    /**
     * Pass messages that the runabout can handle to it.
     * There can only be one runabout per message type.
     * (Discrepancy with the C-API, could be changed in the future)
     *
     * @param msgRunabout handler
     */
    public void setHandler(MessageRunabout msgRunabout) {
        receivedMessagehandler = msgRunabout;
        expectedMessages = RunaboutUtil.getRunaboutVisitees(msgRunabout);
    }

    public Cancelable notifyDisconnect(final DisconnectHandler disconnectHandler) {
        this.disconnectHandlers.add(disconnectHandler);
        return new Cancelable() {
            @Override
            public void cancel() {
                Server.this.disconnectHandlers.remove(disconnectHandler);
            }
        };
    }

    /**
     * Stop the listen socket and get ready to shutdown the server
     * once only 'monitor' clients are left.
     */
    public void stopListening() {
        shutdownRequested = true;
        // todo: shut down if only monitor clients left
    }

    public void destroy() {
        for (ClientHandle h : new ArrayList<ClientHandle>(clients)) {
            h.disconnect();
        }
        acceptTask.cancel();
    }

}