path: root/src/org/gnunet/mesh/Mesh.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.mesh;

import org.gnunet.construct.Construct;
import org.gnunet.mq.ClientMessageQueue;
import org.gnunet.mq.Envelope;
import org.gnunet.mq.MessageQueue;
import org.gnunet.requests.Request;
import org.gnunet.requests.RequestQueue;
import org.gnunet.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

/**
 * Mesh API
 *
 * @author Florian Dold
 */
public class Mesh {
    /**
     * Class logger.
     */
    private static final Logger logger = LoggerFactory
            .getLogger(Mesh.class);

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

    /**
     * For tunnels created by the server, the bit in this
     * mask is always set.
     */
    private static final int TUNNEL_ID_SERV = 0xB0000000;

    /**
     * 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 mesh service
     */
    private final Client client;

    /**
     * Message queue for the client.
     */
    private final ClientMessageQueue client_mq;

    /**
     * Called whenever a tunnel was destroyed.
     */
    private TunnelEndHandler tunnelEndHandler;

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

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

    /**
     * Handler for inbound tunnels.
     */
    private InboundTunnelHandler inboundTunnelHandler;

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

    /**
     * Counter for generating fresh tunnel ID's
     * when creating new tunnels.
     */
    int next_tid = 1;

    /**
     * A tunnel to a remote peer.
     * @param <T> type of context data for the tunnel
     */
    public class Tunnel<T> extends MessageQueue {
        private T context;
        private final int opt;
        public final PeerIdentity peer;
        public final int port;
        protected int tunnelId;
        private boolean receive_done_expected = false;
        int ack_count = 0;

        /**
         * Create a new tunnel (we're initiator and will be allowed to add/remove peers
         * and to broadcast).
         *
         * @param context tunnel context
         * @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 Tunnel(PeerIdentity peer, int port, boolean nobuffer, boolean reliable, T context)
        {
            this(peer, 0, port, nobuffer, reliable);
            TunnelCreateMessage tcm = new TunnelCreateMessage();
            tcm.otherEnd = peer;
            tcm.opt = opt;
            tcm.port = port;
            tcm.tunnel_id = tunnelId;
            client_mq.send(tcm);
        }

        /**
         * Private tunnel constructor, for creating tunnel objects for
         * incoming tunnels.
         *
         * @param peer
         * @param tunnelId
         * @param port
         * @param nobuffer
         * @param reliable
         */
        private Tunnel(PeerIdentity peer, int tunnelId, int port, boolean nobuffer, boolean reliable) {
            int my_opt = 0;
            if (reliable)
                my_opt |= OPTION_RELIABLE;
            if (nobuffer)
                my_opt |= OPTION_NOBUFFER;
            if (0 == tunnelId)
                this.tunnelId = ((next_tid++) | TUNNEL_ID_CLI) & ~TUNNEL_ID_SERV;
            else
                this.tunnelId = tunnelId;
            this.peer = peer;
            this.port = port;
            this.opt = my_opt;
        }

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

        public void destroy() {
            TunnelDestroyMessage m = new TunnelDestroyMessage();
            m.tunnel_id = tunnelId;
            client_mq.send(m);
        }

        @Override
        protected void sendImmediate(Envelope ev) {
            if (ack_count <= 0)
                throw new AssertionError();
            DataMessage m = new DataMessage();
            m.payload = Construct.toBinary(GnunetMessage.fromBody(ev.message));
            Envelope mesh_ev = new Envelope(m);
            client_mq.send(mesh_ev);
            ack_count -= 1;
        }

        public T getContext() {
            return context;
        }

        public void setContext(T newContext) {
            context = newContext;
        }

    }

    private class MeshMessageReceiver extends RunaboutMessageReceiver {
        public void visit(TunnelCreateMessage m) {
            Tunnel t = new Tunnel(m.otherEnd, m.tunnel_id, m.port,
                    (m.opt & OPTION_NOBUFFER) != 0, (m.opt & OPTION_NOBUFFER) != 0);
            if (inboundTunnelHandler != null) {
                inboundTunnelHandler.onInboundTunnel(t, m.otherEnd);
            }
        }

        public void visit(DataMessage m) {
            Tunnel t = tunnelMap.get(m.tid);
            if (t != null)
            {
                if (t.receive_done_expected)
                    logger.warn("got unexpected message from service");
                t.receive_done_expected = true;
                messageReceiver.visitAppropriate(Construct.parseAs(m.payload, GnunetMessage.class).body);
            }
        }

        public void visit(LocalAckMessage m) {
            Tunnel t = tunnelMap.get(m.tid);
            if (t != null)
                t.ack_count += 1;
        }

        public void visit(TunnelDestroyMessage m) {
            Tunnel t = tunnelMap.get(m.tunnel_id);
            if (t == null) {
                logger.warn("server got confused with tunnel IDs on destroy, ignoring message");
                return;
            }
            t.destroy();
            tunnelEndHandler.onTunnelEnd(t);
        }

        @Override
        public void handleError() {
            if (tunnelEndHandler != null) {
                for (Tunnel t : tunnelMap.values()) {
                    tunnelEndHandler.onTunnelEnd(t);
                }
            }
            tunnelMap.clear();
            client.reconnect();
            ClientConnectMessage ccm = new ClientConnectMessage();
            ccm.apps_list = ports;
            client_mq.send(ccm);
        }
    }

    /**
     * Connect to the mesh service.
     *
     * @param cfg                  configuration to use
     * @param inboundTunnelHandler function called when an *inbound* tunnel is created
     * @param tunnelEndHandler     function called when an *inbound* tunnel is destroyed by the
     *                             remote peer, it is *not* called if Tunnel.destroy
     *                             is called on the tunnel
     */
    public Mesh(Configuration cfg, InboundTunnelHandler inboundTunnelHandler,
                TunnelEndHandler tunnelEndHandler, MeshRunabout messageReceiver, int... ports) {
        this.tunnelEndHandler = tunnelEndHandler;
        this.messageReceiver = messageReceiver;
        this.ports = ports;
        this.inboundTunnelHandler = inboundTunnelHandler;

        client = new Client("mesh", cfg);
        client_mq = new ClientMessageQueue(client, new MeshMessageReceiver());
        ClientConnectMessage ccm = new ClientConnectMessage();
        ccm.apps_list = ports;
        client_mq.send(ccm);
    }

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