path: root/src/main/java/org/gnunet/cadet/Cadet.java

/*
 This file is part of GNUnet.
 Copyright (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.cadet;

import org.gnunet.construct.Construct;
import org.gnunet.cadet.messages.ClientConnectMessage;
import org.gnunet.cadet.messages.LocalAckMessage;
import org.gnunet.cadet.messages.TunnelCreateMessage;
import org.gnunet.cadet.messages.TunnelDestroyMessage;
import org.gnunet.mq.Envelope;
import org.gnunet.mq.MessageQueue;
import org.gnunet.mq.NotifySentHandler;
import org.gnunet.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * Connection to the cadet service.
 */
public class Cadet {
    /**
     * Class logger.
     */
    private static final Logger logger = LoggerFactory
            .getLogger(Cadet.class);

    /**
     * For tunnels created by the client, the bit in this
     * mask is always set.
     */
    private static final long TUNNEL_ID_CLI = 0x80000000L;

    /**
     * For tunnels created by the service, the bit in this
     * mask is always set.
     */
    private static final long TUNNEL_ID_SERV = 0xB0000000L;

    /**
     * Disable buffering on intermediate nodes (for minimum latency).
     * Yes/No.
     */
    private static final int OPTION_NOBUFFER = 1;

    /**
     * Enable tunnel reliability, lost messages will be retransmitted.
     * Yes/No.
     */
    private static final int OPTION_RELIABLE = 2;

    /**
     * Client connected to the cadet service
     */
    private final Client client;

    /**
     * Called whenever a tunnel was destroyed.
     */
    private ChannelEndHandler channelEndHandler;

    /**
     * Message handler for messages received through
     * a tunnel.
     */
    private CadetRunabout messageReceiver;

    /**
     * Ports that we listen on.
     */
    private int[] ports;

    /**
     * Handler for inbound tunnels.
     */
    private InboundChannelHandler inboundChannelHandler;

    /**
     * Mapping from the tunnel's ID to the tunnel object.
     */
    private Map<Long,Channel> tunnelMap = new HashMap<Long,Channel>();

    /**
     * Counter for generating fresh tunnel ID's
     * when creating new tunnels.
     */
    private long nextTid = 1;

    /**
     * A tunnel to a remote peer.
     */
    public class Channel extends MessageQueue {
        private final int opt;
        final PeerIdentity peer;
        final int port;
        protected long tunnelId;
        private boolean receiveDoneExpected = false;
        int ackCount = 0;
        boolean destroyedByService;

        /**
         * Canceler for the currently submitted envelope.
         */
        private Cancelable envelopeCanceler;

        /**
         * Create a new tunnel (we're initiator and will be allowed to add/remove peers
         * and to broadcast).
         *
         * @param peer peer identity the tunnel should go to
         * @param port Port number.
         * @param nobuffer Flag for disabling buffering on relay nodes.
         * @param reliable Flag for end-to-end reliability.
         */
        public Channel(PeerIdentity peer, int port, boolean nobuffer, boolean reliable) {
            this(peer, 0, port, nobuffer, reliable);
            TunnelCreateMessage tcm = new TunnelCreateMessage();
            tcm.otherEnd = peer;
            tcm.opt = opt;
            tcm.port = port;
            tcm.tunnelId = tunnelId;
            client.send(tcm);
        }


        /**
         * Private tunnel constructor, for creating tunnel objects for
         * incoming tunnels.
         *
         * @param peer
         * @param tunnelId
         * @param port
         * @param nobuffer
         * @param reliable
         */
        public Channel(PeerIdentity peer, long tunnelId, int port, boolean nobuffer, boolean reliable) {
            int myOpt = 0;
            if (reliable)
                myOpt |= OPTION_RELIABLE;
            if (nobuffer)
                myOpt |= OPTION_NOBUFFER;
            if (0 == tunnelId) {
                this.tunnelId = nextTid++;
                this.tunnelId &= ~TUNNEL_ID_SERV;
                this.tunnelId |= TUNNEL_ID_CLI;
            }
            else {
                this.tunnelId = tunnelId;
            }
            this.peer = peer;
            this.port = port;
            this.opt = myOpt;
            logger.debug("registering tunnel {}", this.tunnelId);
            tunnelMap.put(this.tunnelId, this);
        }

        public void receiveDone() {
            if (!receiveDoneExpected)
                throw new AssertionError("unexpected call to receiveDone");
            LocalAckMessage am = new LocalAckMessage();
            am.tid = tunnelId;
            client.send(am);
            receiveDoneExpected = false;
        }

        public void destroy() {
            if (!destroyedByService) {
                TunnelDestroyMessage m = new TunnelDestroyMessage();
                m.tunnelId = tunnelId;
                m.reserved = new byte[32];
                client.send(m);
            }
            tunnelMap.remove(tunnelId);
        }

        @Override
        protected void submit(Envelope ev) {
            logger.debug("submitting data message on tunnel {}", tunnelId);
            if (ackCount <= 0)
                throw new AssertionError();
            DataMessage m = new DataMessage();
            m.payload = Construct.toBinary(GnunetMessage.fromBody(ev.message));
            m.tid = tunnelId;
            Envelope cadetEv = new Envelope(m);
            cadetEv.notifySent(new NotifySentHandler() {
                @Override
                public void onSent() {
                    envelopeCanceler = null;
                    reportMessageSent();
                }
            });
            client.send(cadetEv);
            envelopeCanceler = cadetEv;
            ackCount -= 1;
        }

        @Override
        protected void retract() {
            if (envelopeCanceler == null)
                throw new AssertionError();
            envelopeCanceler.cancel();
            envelopeCanceler = null;
        }

        void handleAck() {
            ackCount++;
            logger.debug("got ack for tunnel id " + tunnelId);
            if (ackCount == 1) {
                reportReadyForSubmit();
            }
        }
    }

    private class CadetMessageReceiver extends RunaboutMessageReceiver {
        public void visit(TunnelCreateMessage m) {
            Channel t = new Channel(m.otherEnd, m.tunnelId, m.port,
                    (m.opt & OPTION_NOBUFFER) != 0, (m.opt & OPTION_NOBUFFER) != 0);
            logger.debug("inbound tunnel {}", m.tunnelId);
            if (inboundChannelHandler != null) {
                inboundChannelHandler.onInboundChannel(t, m.otherEnd);
            }
        }

        public void visit(DataMessage m) {
            Channel t = tunnelMap.get(m.tid);
            if (t != null)
            {
                if (t.receiveDoneExpected)
                    logger.warn("got unexpected message from service");
                t.receiveDoneExpected = true;
                messageReceiver.setSender(t);
                GnunetMessage gnunetMessage = Construct.parseAs(m.payload, GnunetMessage.class);
                logger.debug("received message of size {} and type {}",
                        gnunetMessage.header.messageSize, gnunetMessage.header.messageType);
                messageReceiver.visitAppropriate(gnunetMessage.body);
                messageReceiver.setSender(null);
            }
        }

        public void visit(LocalAckMessage m) {
            logger.debug("got LocalAckMessage for {}", m.tid);
            Channel t = tunnelMap.get(m.tid);
            if (t != null) {
                t.handleAck();
            } else {
                logger.warn("tunnel id for local ack not found");
            }
        }

        public void visit(TunnelDestroyMessage m) {
            Channel t = tunnelMap.get(m.tunnelId);
            if (null == t) {
                logger.warn("server got confused with tunnel IDs on destroy, ignoring message");
                return;
            }
            t.destroyedByService = true;
            logger.debug("tunnel destroyed by service");
            t.destroy();
            channelEndHandler.onChannelEnd(t);
        }

        public void visit(RejectMessage m) {
            // FIXME: C code indicates that the nack/reject message might change ...
            Channel t = tunnelMap.get(m.tunnelId);
            if (null == t) {
                logger.warn("server got confused with tunnel IDs on destroy, ignoring message");
                return;
            }
            t.destroyedByService = true;
            logger.debug("tunnel destroyed by service (nack/reject)");
            t.destroy();
            channelEndHandler.onChannelEnd(t);
        }

        @Override
        public void handleError() {
            logger.warn("lost connection to cadet service, reconnecting");
            if (null != channelEndHandler) {
                for (Channel t : tunnelMap.values()) {
                    channelEndHandler.onChannelEnd(t);
                }
            }
            tunnelMap.clear();
            client.reconnect();
            ClientConnectMessage ccm = new ClientConnectMessage();
            ccm.applicationList = ports;
            client.send(ccm);
        }
    }

    /**
     * Connect to the cadet service, listening to the given ports.
     *
     * @param cfg configuration to use
     * @param inboundChannelHandler called when an inbound channel is established
     * @param channelEndHandler called when a tunnel is destroyed (either by the client calling Tunnel.destroy(),
     *                         or by the service)
     * @param messageReceiver runabout for messages we are interested in
     * @param ports ports to listen on
     */
    public Cadet(Configuration cfg, InboundChannelHandler inboundChannelHandler,
                ChannelEndHandler channelEndHandler, CadetRunabout messageReceiver, int... ports) {
        if (null == channelEndHandler) {
            throw new AssertionError("tunnel end handler may not be null");
        }
        this.channelEndHandler = channelEndHandler;
        this.messageReceiver = messageReceiver;
        this.ports = ports;
        this.inboundChannelHandler = inboundChannelHandler;
        client = new Client("cadet", cfg);
        client.installReceiver(new CadetMessageReceiver());
        ClientConnectMessage ccm = new ClientConnectMessage();
        ccm.applicationList = ports;
        String portList = "";
        for (int p : ports) {
            portList += "" + p + " ";
        }
        client.send(ccm);

        logger.debug("cadet handle created, listening on ports {}", portList);
    }

    /**
     * Connect to the cadet service. Use this constructor if you are not interested in inbound channels.
     *
     * @param cfg configuration to use
     * @param channelEndHandler called when a tunnel is destroyed (either by the client calling Channel.destroy(),
     *                         or by the service)
     */
    public Cadet(Configuration cfg, ChannelEndHandler channelEndHandler, CadetRunabout messageReceiver) {
        this(cfg, null, channelEndHandler, messageReceiver);
    }

    /**
     * Connect to the cadet service. Use this constructor if you are not interested in inbound tunnels
     * and don't want to receive messages.
     *
     * @param cfg configuration to use
     * @param channelEndHandler called when a tunnel is destroyed (either by the client calling Tunnel.destroy(),
     *                         or by the service)
     */
    public Cadet(Configuration cfg, ChannelEndHandler channelEndHandler) {
        this(cfg, null, channelEndHandler, null);
    }


    /**
     * Create a channel to a peer over the given port, with the given options.
     *
     * @param peer peer to create a channel to
     * @param port port to use
     * @param nobuffer true if messages should be buffered
     * @param reliable true if transmission should be reliable
     * @return a channel
     */
    public Channel createChannel(PeerIdentity peer, int port, boolean nobuffer, boolean reliable) {
        logger.debug("creating tunnel to peer {} over port {}", peer.toString(), port);
        return new Channel(peer, port, nobuffer, reliable);
    }

    /**
     * Disconnect from the cadet service.
     * All tunnels will be destroyed.
     * All tunnel disconnect callbacks will be called on any still connected peers, notifying
     * about their disconnection.
     */
    public void destroy() {
        client.disconnect();
    }
}