- mq for java

- started with consensus
- started with new mesh

26 files changed, 525 insertions, 485 deletions

diff --git a/src/org/gnunet/consensus/ConcludeCallback.java b/src/org/gnunet/consensus/ConcludeCallback.java
new file mode 100644
index 0000000..7660c49
--- /dev/null
+++ b/src/org/gnunet/consensus/ConcludeCallback.java
@@ -0,0 +1,5 @@
+package org.gnunet.consensus;
+
+public interface ConcludeCallback {
+    void onConcludeDone();
+}
diff --git a/src/org/gnunet/consensus/ConcludeDoneMessage.java b/src/org/gnunet/consensus/ConcludeDoneMessage.java
new file mode 100644
index 0000000..3b016cf
--- /dev/null
+++ b/src/org/gnunet/consensus/ConcludeDoneMessage.java
@@ -0,0 +1,12 @@
+package org.gnunet.consensus;
+
+
+import org.gnunet.construct.UnionCase;
+
+/**
+ * Notify the client that conclude has finished.
+ * Direction: service -> client
+ */
+@UnionCase(525)
+public class ConcludeDoneMessage {
+}
diff --git a/src/org/gnunet/consensus/ConcludeMessage.java b/src/org/gnunet/consensus/ConcludeMessage.java
new file mode 100644
index 0000000..7b43928
--- /dev/null
+++ b/src/org/gnunet/consensus/ConcludeMessage.java
@@ -0,0 +1,19 @@
+package org.gnunet.consensus;
+
+import org.gnunet.construct.FillWith;
+import org.gnunet.construct.UInt16;
+import org.gnunet.construct.UInt8;
+import org.gnunet.construct.UnionCase;
+import org.gnunet.util.GnunetMessage;
+
+/**
+ * Notify the client of a new element.
+ *
+ * Direction: service -> client
+ *
+ * @author Florian Dold
+ */
+@UnionCase(524)
+public class ConcludeMessage implements GnunetMessage.Body {
+    /* empty body */
+}
\ No newline at end of file
diff --git a/src/org/gnunet/consensus/Consensus.java b/src/org/gnunet/consensus/Consensus.java
new file mode 100644
index 0000000..e6fab01
--- /dev/null
+++ b/src/org/gnunet/consensus/Consensus.java
@@ -0,0 +1,120 @@
+package org.gnunet.consensus;
+
+import org.gnunet.mq.ClientMessageQueue;
+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;
+
+/**
+ * Multi-peer set reconciliation.
+ */
+public class Consensus {
+    /**
+     * Class logger.
+     */
+    private static final Logger logger = LoggerFactory
+            .getLogger(Consensus.class);
+
+    /**
+     * Callback for new elements arriving from the service.
+     * Also used to notify of consensus failure.
+     */
+    private final NewElementCallback newElementCallback;
+
+    /**
+     * Client connected to the consensus service.
+     */
+    private Client client;
+
+    /**
+     * Message queue for 'client'.
+     */
+    private MessageQueue client_mq;
+
+    /**
+     * Called when conclude has finished.
+     */
+    private ConcludeCallback concludeCallback;
+
+    /**
+     * Message dispatch for messages from the consensus service.
+     */
+    private class ConsensusMessageReceiver extends RunaboutMessageReceiver {
+        public void visit(ConcludeDoneMessage m) {
+            if (null == concludeCallback)
+            {
+                logger.error("unexpected conclude done message");
+                return;
+            }
+            concludeCallback.onConcludeDone();
+        }
+
+        public void visit(NewElementMessage m) {
+            ConsensusElement element = new ConsensusElement();
+            element.element_type = m.element_type;
+            element.data = m.element_data;
+            newElementCallback.onNewElement(element);
+        }
+
+        @Override
+        public void handleError() {
+            newElementCallback.onNewElement(null);
+        }
+    }
+
+    /**
+     * Create a consensus session.  The set being reconciled is initially
+     * empty.  Only reconcile with other peers after
+     * GNUNET_CONSENSUS_reconcile has been called.
+     *
+     * @param num_peers number of peers in the session
+     * @param peers array of peers participating in this consensus session
+     *              Inclusion of the local peer is optional.
+     * @param sessionId session identifier
+     *                   Allows a group of peers to have more than consensus session.
+     * @param newElementCallback callback, called when a new element is added to the set by
+     *                    another peer
+     */
+    public Consensus(Configuration cfg, int num_peers, PeerIdentity[] peers, HashCode sessionId,
+                     NewElementCallback newElementCallback) {
+        client = new Client("consensus", cfg);
+        client_mq = new ClientMessageQueue(client, new ConsensusMessageReceiver());
+        this.newElementCallback = newElementCallback;
+    }
+
+    public void insertElement (ConsensusElement element, final InsertDoneCallback idc) {
+        InsertElementMessage m = new InsertElementMessage();
+        m.element_data = element.data;
+        m.element_type = element.element_type;
+        Envelope ev = new Envelope(m);
+        ev.notifySent(new NotifySentHandler() {
+            @Override
+            public void onSent() {
+                idc.onInsertDone();
+            }
+        });
+        client_mq.send(ev);
+    }
+
+    public void conclude(ConcludeCallback concludeCallback) {
+        if (null == concludeCallback)
+            throw new AssertionError("conclude with empty callback");
+        if (null != this.concludeCallback)
+            throw new AssertionError("called conclude twice");
+        this.concludeCallback = concludeCallback;
+    }
+
+    /**
+     * Destroy a consensus handle (free all state associated with
+     * it, no longer call any of the callbacks).
+     */
+    public void destroy() {
+        client_mq.destroy();
+        client_mq = null;
+        client.disconnect();
+        client = null;
+    }
+}
diff --git a/src/org/gnunet/consensus/ConsensusElement.java b/src/org/gnunet/consensus/ConsensusElement.java
new file mode 100644
index 0000000..846e72a
--- /dev/null
+++ b/src/org/gnunet/consensus/ConsensusElement.java
@@ -0,0 +1,16 @@
+package org.gnunet.consensus;
+
+
+public class ConsensusElement {
+    /**
+     * Type of the element.
+     * 0 <= element_type <= 2^16
+     */
+    int element_type;
+
+    /**
+     * Data for the element.
+     * 0 <= data.length <= 2^16
+     */
+    byte[] data;
+}
diff --git a/src/org/gnunet/consensus/InsertDoneCallback.java b/src/org/gnunet/consensus/InsertDoneCallback.java
new file mode 100644
index 0000000..f0e86ca
--- /dev/null
+++ b/src/org/gnunet/consensus/InsertDoneCallback.java
@@ -0,0 +1,5 @@
+package org.gnunet.consensus;
+
+public interface InsertDoneCallback {
+    void onInsertDone();
+}
diff --git a/src/org/gnunet/consensus/InsertElementMessage.java b/src/org/gnunet/consensus/InsertElementMessage.java
new file mode 100644
index 0000000..fd0ff67
--- /dev/null
+++ b/src/org/gnunet/consensus/InsertElementMessage.java
@@ -0,0 +1,23 @@
+package org.gnunet.consensus;
+
+import org.gnunet.construct.FillWith;
+import org.gnunet.construct.UInt16;
+import org.gnunet.construct.UInt8;
+import org.gnunet.construct.UnionCase;
+import org.gnunet.util.GnunetMessage;
+
+/**
+ * Send an element to the service, insert it into the consensus set.
+ *
+ * Direction: client -> service
+ *
+ * @author Florian Dold
+ */
+@UnionCase(521)
+public class InsertElementMessage implements GnunetMessage.Body {
+    @UInt16
+    public int element_type;
+    @FillWith
+    @UInt8
+    public byte[] element_data;
+}
\ No newline at end of file
diff --git a/src/org/gnunet/consensus/NewElementCallback.java b/src/org/gnunet/consensus/NewElementCallback.java
new file mode 100644
index 0000000..4b07a71
--- /dev/null
+++ b/src/org/gnunet/consensus/NewElementCallback.java
@@ -0,0 +1,5 @@
+package org.gnunet.consensus;
+
+public interface NewElementCallback {
+    void onNewElement(ConsensusElement element);
+}
diff --git a/src/org/gnunet/consensus/NewElementMessage.java b/src/org/gnunet/consensus/NewElementMessage.java
new file mode 100644
index 0000000..deb3634
--- /dev/null
+++ b/src/org/gnunet/consensus/NewElementMessage.java
@@ -0,0 +1,20 @@
+package org.gnunet.consensus;
+
+import org.gnunet.construct.*;
+import org.gnunet.util.GnunetMessage;
+
+/**
+ * Notify the client of a new element.
+ *
+ * Direction: service -> client
+ *
+ * @author Florian Dold
+ */
+@UnionCase(523)
+public class NewElementMessage implements GnunetMessage.Body {
+    @UInt16
+    public int element_type;
+    @FillWith
+    @UInt8
+    public byte[] element_data;
+}
\ No newline at end of file
diff --git a/src/org/gnunet/construct/MsgMap.txt b/src/org/gnunet/construct/MsgMap.txt
index 719e92a..30e7077 100644
--- a/src/org/gnunet/construct/MsgMap.txt
+++ b/src/org/gnunet/construct/MsgMap.txt
@@ -25,7 +25,7 @@ org.gnunet.util.GnunetMessage$Body|153=org.gnunet.dht.MonitorStartStop
 org.gnunet.util.GnunetMessage$Body|155=org.gnunet.dht.ClientPutConfirmationMessage
 org.gnunet.util.GnunetMessage$Body|262=org.gnunet.mesh.OriginMessage
 org.gnunet.util.GnunetMessage$Body|323=org.gnunet.nse.UpdateMessage
-org.gnunet.util.GnunetMessage$Body|260=org.gnunet.mesh.UnicastMessage
+org.gnunet.util.GnunetMessage$Body|260=org.gnunet.mesh.DataMessage
 org.gnunet.util.GnunetMessage$Body|261=org.gnunet.mesh.MulticastMessage
 org.gnunet.util.GnunetMessage$Body|321=org.gnunet.nse.StartMessage
 org.gnunet.util.GnunetMessage$Body|144=org.gnunet.dht.ClientGetStopMessage
diff --git a/src/org/gnunet/mesh/ClientConnectMessage.java b/src/org/gnunet/mesh/ClientConnectMessage.java
index 57edede..1a56ebb 100644
--- a/src/org/gnunet/mesh/ClientConnectMessage.java
+++ b/src/org/gnunet/mesh/ClientConnectMessage.java
@@ -12,18 +12,6 @@ import org.gnunet.util.GnunetMessage;
  */
 @UnionCase(272)
 public class ClientConnectMessage implements GnunetMessage.Body {
-    @UInt16
-    public int applications_length;
-    @UInt16
-    public int types_length;
-    /**
-     * List of applications that this client claims to provide.
-     */
-    @VariableSizeIntegerArray(lengthField = "applications_length", signed = false, bitSize = 32)
+    @IntegerFill(signed = false, bitSize = 32)
     public int[] apps_list;
-    /**
-     * Message types that this client understands.
-     */
-    @VariableSizeIntegerArray(lengthField = "types_length", signed = false, bitSize = 16)
-    public int[] types_list;
 }
diff --git a/src/org/gnunet/mesh/ConnectPeerByTypeMessage.java b/src/org/gnunet/mesh/ConnectPeerByTypeMessage.java
deleted file mode 100644
index fe8bfe0..0000000
--- a/src/org/gnunet/mesh/ConnectPeerByTypeMessage.java
+++ /dev/null
@@ -1,18 +0,0 @@
-package org.gnunet.mesh;
-
-import org.gnunet.construct.UInt32;
-import org.gnunet.construct.UnionCase;
-import org.gnunet.util.GnunetMessage;
-/**
- * ...
- *
- * @author Florian Dold
- */
-@UnionCase(277)
-public class ConnectPeerByTypeMessage implements GnunetMessage.Body {
-    @UInt32
-    public int tunnelId;
-
-    @UInt32
-    public int applicationType;
-}
diff --git a/src/org/gnunet/mesh/UnicastMessage.java b/src/org/gnunet/mesh/DataMessage.java
index 141d49a..92546c8 100644
--- a/src/org/gnunet/mesh/UnicastMessage.java
+++ b/src/org/gnunet/mesh/DataMessage.java
@@ -10,17 +10,9 @@ import org.gnunet.util.PeerIdentity;
  * @author Florian Dold
  */
 @UnionCase(260)
-public class UnicastMessage implements GnunetMessage.Body {
+public class DataMessage implements GnunetMessage.Body {
     @UInt32
     public int tid;
-    @UInt32
-    public int ttl;
-    @UInt32
-    public int pid;
-    @NestedMessage
-    public PeerIdentity oid;
-    @NestedMessage
-    public PeerIdentity destination;
     @FillWith
     @UInt8
     public byte[] payload;
diff --git a/src/org/gnunet/mesh/Mesh.java b/src/org/gnunet/mesh/Mesh.java
index 6f14451..091cb17 100644
--- a/src/org/gnunet/mesh/Mesh.java
+++ b/src/org/gnunet/mesh/Mesh.java
@@ -20,15 +20,17 @@
 
 package org.gnunet.mesh;
 
-import com.google.common.collect.Maps;
 import org.gnunet.construct.Construct;
-import org.gnunet.requests.FixedMessageRequest;
+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;
 
 /**
@@ -37,345 +39,188 @@ import java.util.Map;
  * @author Florian Dold
  */
 public class Mesh {
+    /**
+     * Class logger.
+     */
     private static final Logger logger = LoggerFactory
             .getLogger(Mesh.class);
 
     /**
-     * How many messages can we send to the service until we have to wait for an ACK from it?
+     * For tunnels created by the client, the bit in this
+     * mask is always set.
      */
-    private static final int INITIAL_WINDOW_SIZE = 8;
+    private static final int TUNNEL_ID_CLI = 0x80000000;
+
     /**
-     * Requests queued to be sent to the mesh service.
+     * For tunnels created by the server, the bit in this
+     * mask is always set.
      */
-    private RequestQueue requestQueue;
+    private static final int TUNNEL_ID_SERV = 0xB0000000;
+
     /**
-     * Called whenever a tunnel was destroyed.
+     * Disable buffering on intermediate nodes (for minimum latency).
+     * Yes/No.
      */
-    private TunnelEndHandler tunnelEndHandler;
-    private MeshRunabout messageReceiver;
-    private int[] applications;
-    private InboundTunnelHandler inboundTunnelHandler;
+    private static final int OPTION_NOBUFFER = 1;
 
-    private final static int LOCAL_TUNNEL_ID_CLI = 0x80000000;
-    private final static int LOCAL_TUNNEL_ID_SERV = 0xB0000000;
+    /**
+     * Enable tunnel reliability, lost messages will be retransmitted.
+     * Yes/No.
+     */
+    private static final int OPTION_RELIABLE = 2;
 
-    private int nextTunnelId = LOCAL_TUNNEL_ID_CLI;
+    /**
+     * Client connected to the mesh service
+     */
+    private final Client client;
 
     /**
-     * Stores all tunnels created by this client, referenced by their local tunnel id.
+     * Message queue for the client.
      */
-    private Map<Integer, Tunnel> tunnelMap = Maps.newTreeMap();
+    private final ClientMessageQueue client_mq;
 
+    /**
+     * Called whenever a tunnel was destroyed.
+     */
+    private TunnelEndHandler tunnelEndHandler;
 
-    public class OriginTunnel extends Tunnel {
-        public DisconnectHandler disconnectHandler;
-        public ConnectHandler connectHandler;
+    /**
+     * Message handler for messages received through
+     * a tunnel.
+     */
+    private MeshRunabout messageReceiver;
 
-        public void addPeer(PeerIdentity peerIdentity) {
-            throw new UnsupportedOperationException("not implemented");
-        }
+    /**
+     * Ports that we listen on.
+     */
+    private int[] ports;
 
-        /**
-         * Request that the given peer isn't added to this tunnel in calls to
-         * connect_by_* calls, (due to misbehaviour, bad performance, ...).
-         *
-         * @param peerIdentity peer identity of the peer which should be blacklisted
-         *                     for the tunnel.
-         */
-        public void blacklist(PeerIdentity peerIdentity) {
-            throw new UnsupportedOperationException("not implemented");
-        }
+    /**
+     * Handler for inbound tunnels.
+     */
+    private InboundTunnelHandler inboundTunnelHandler;
 
-        /**
-         * Request that the given peer isn't blacklisted anymore from this tunnel,
-         * and therefore can be added in future calls to connect*.
-         * The peer must have been previously blacklisted for this tunnel.
-         *
-         * @param peerIdentity peer identity of the peer which shouldn't be blacklisted
-         *                     for the tunnel anymore.
-         */
-        public void unblacklist(PeerIdentity peerIdentity) {
-            throw new UnsupportedOperationException("not implemented");
-        }
+    /**
+     * Mapping from the tunnel's ID to the tunnel object.
+     */
+    private Map<Integer,Tunnel> tunnelMap = new HashMap<>();
 
-        /**
-         * Request that the mesh should try to connect to a peer supporting the given
-         * message type.
-         *
-         * @param appType application type that must be supported by the peer
-         *                (MESH should discover peer in proximity handling this type)
-         */
-        public void requestConnectByType(int appType) {
-            ConnectPeerByTypeMessage m = new ConnectPeerByTypeMessage();
-            m.applicationType = appType;
-            m.tunnelId = tunnelId;
-            requestQueue.add(new FixedMessageRequest(m));
-        }
+    /**
+     * Counter for generating fresh tunnel ID's
+     * when creating new tunnels.
+     */
+    int next_tid = 1;
 
-        /**
-         * Request that the mesh should try to connect to a peer matching the
-         * description given in the service string.
-         *
-         * @param description string describing the destination node requirements
-         */
-        public void requestConnectByString(String description) {
-            throw new UnsupportedOperationException("not implemented");
-        }
+    /**
+     * A tunnel to a remote peer.
+     * @param <T> type of context data for the tunnel
+     */
+    public class Tunnel<T> extends MessageQueue {
+        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;
 
         /**
-         * Request that a peer should be added to the tunnel.  The connect handler
-         * will be called when the peer connects
+         * Create a new tunnel (we're initiator and will be allowed to add/remove peers
+         * and to broadcast).
          *
-         * @param peer peer to add
+         * @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 void requestConnectAdd(PeerIdentity peer) {
-            throw new UnsupportedOperationException("not implemented");
+        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);
         }
 
         /**
-         * Request that a peer should be removed from the tunnel.  The existing
-         * disconnect handler will be called ONCE if we were connected.
+         * Private tunnel constructor, for creating tunnel objects for
+         * incoming tunnels.
          *
-         * @param peer peer to remove
+         * @param peer
+         * @param tunnelId
+         * @param port
+         * @param nobuffer
+         * @param reliable
          */
-        public void requestConnectDel(PeerIdentity peer) {
-            throw new UnsupportedOperationException("not implemented");
-        }
-
-        private void registerWithService() {
-            requestQueue.add(new TunnelCreateRequest(this));
-        }
-    }
-
-    public static class TunnelTransmitRequest extends Request {
-
-        public boolean doNotSend = false;
-
-        static class Sink implements Connection.MessageSink {
-            byte[] payload;
-
-            @Override
-            public void send(GnunetMessage.Body m) {
-                if (payload != null) {
-                    throw new AssertionError("only one payload allowed per transmitter");
-                }
-                payload = Construct.toBinary(GnunetMessage.fromBody(m));
-            }
-        }
-
-        public PeerIdentity target;
-        public MessageTransmitter transmitter;
-        public Tunnel tunnel;
-
-        @Override
-        public void transmit(Connection.MessageSink sink) {
-            if (doNotSend) {
-                return;
-            }
-
-            Sink s = new Sink();
-            transmitter.transmit(s);
-
-            if (tunnel.tunnelId >= Mesh.LOCAL_TUNNEL_ID_SERV) {
-                // we are not the origin, thus can only send to origin
-                OriginMessage m = new OriginMessage();
-                m.tid = tunnel.tunnelId;
-                m.pid = tunnel.nextSentPacketId++;
-                m.payload = s.payload;
-                m.oid = new PeerIdentity();
-                m.sender = m.oid;
-                sink.send(m);
-            } else if (target == null) {
-                // multicast, we are origin
-                MulticastMessage m = new MulticastMessage();
-                m.tid = tunnel.tunnelId;
-                m.pid = tunnel.nextSentPacketId++;
-                m.payload = s.payload;
-                sink.send(m);
-            } else {
-                // unicast
-                System.out.println("sending unicast");
-                UnicastMessage m = new UnicastMessage();
-                m.destination = target;
-                m.oid = new PeerIdentity();
-                m.tid = tunnel.tunnelId;
-                m.pid = tunnel.nextSentPacketId++;
-                m.payload = s.payload;
-                sink.send(m);
-            }
+        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 class Tunnel {
-        protected int tunnelId;
-
-        private int nextSentPacketId = 0;
-        private int maxSentPacketId = INITIAL_WINDOW_SIZE - 1;
-
-        private TunnelTransmitRequest waitingTunnelTransmitRequest;
-
-        /**
-         * Ask the mesh to call "notify" once it is ready to transmit the
-         * given number of bytes to the specified tunnel or target.
-         * Only one call can be active at any time, to issue another request,
-         * wait for the callback or cancel the current request.
-         *
-         * @param maxdelay    how long can the message wait?
-         * @param target      destination for the message
-         *                    NULL for multicast to all tunnel targets
-         * @param notify_size how many bytes of buffer space does notify want?
-         * @param transmitter handler to call when buffer space is available;
-         *                    will be called with NULL on timeout or if the overall queue
-         *                    for this peer is larger than queue_size and this is currently
-         *                    the message with the lowest priority
-         * @return non-NULL if the notify callback was queued,
-         *         NULL if we can not even queue the request (insufficient
-         *         memory); if NULL is returned, "notify" will NOT be called.
-         */
-        public Cancelable notifyTransmitReady(RelativeTime maxdelay, PeerIdentity target, int notify_size, MessageTransmitter transmitter) {
-            if (waitingTunnelTransmitRequest != null) {
-                throw new AssertionError();
-            }
 
-            final TunnelTransmitRequest request = new TunnelTransmitRequest();
-            request.target = target;
-            request.transmitter = transmitter;
-            request.tunnel = this;
-            request.setDeadline(maxdelay.toAbsolute());
-
-            if (nextSentPacketId <= maxSentPacketId) {
-
-                final Cancelable cancel = requestQueue.add(request);
-
-                return new Cancelable() {
-                    @Override
-                    public void cancel() {
-                        cancel.cancel();
-                    }
-                };
-            } else {
-                // we have to wait until we get a local ack from the service
-                waitingTunnelTransmitRequest = request;
-
-                return new Cancelable() {
-                    @Override
-                    public void cancel() {
-                        request.doNotSend = true;
-                    }
-                };
-            }
+        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() {
-            // todo
-        }
-
-        private void onAckUpdated() {
-            if (waitingTunnelTransmitRequest == null) {
-                return;
-            }
-            requestQueue.add(waitingTunnelTransmitRequest);
-            waitingTunnelTransmitRequest = null;
-        }
-    }
-
-
-    /**
-     * A request to initialize the connection with the mesh service.
-     */
-    public class ClientConnectRequest extends Request {
-        @Override
-        public void transmit(Connection.MessageSink sink) {
-            System.out.println("transmit called " + this);
-
-            ClientConnectMessage ccm = new ClientConnectMessage();
-            ccm.applications_length = applications.length;
-            ccm.apps_list = applications;
-            int[] types;
-            if (messageReceiver != null) {
-                types = RunaboutUtil.getRunaboutMessageTypes(messageReceiver);
-            } else {
-                types = new int[0];
-            }
-            ccm.types_list = types;
-            ccm.types_length = types.length;
-
-            sink.send(ccm);
-        }
-    }
-
-    public static class TunnelCreateRequest extends Request {
-        public OriginTunnel tunnel;
-
-        public TunnelCreateRequest(OriginTunnel rootTunnel) {
-            tunnel = rootTunnel;
+            TunnelDestroyMessage m = new TunnelDestroyMessage();
+            m.tunnel_id = tunnelId;
+            client_mq.send(m);
         }
 
         @Override
-        public void transmit(Connection.MessageSink sink) {
-            TunnelCreateMessage tcm = new TunnelCreateMessage();
-            tcm.tunnel_id = tunnel.tunnelId;
-            sink.send(tcm);
+        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;
         }
     }
 
 
     private class MeshMessageReceiver extends RunaboutMessageReceiver {
-        public void visit(PeerAddMessage b) {
-            Tunnel r = tunnelMap.get(b.tunnelId);
-            if (r == null || !(r instanceof OriginTunnel)) {
-                logger.warn("server got confused with tunnel IDs on peer add, ignoring message");
-                return;
-            }
-            OriginTunnel ot = (OriginTunnel) r;
-            if (ot.connectHandler != null) {
-                ot.connectHandler.onConnect(ot, b.peer);
-            }
-        }
-
-        public void visit(PeerDeleteMessage b) {
-            Tunnel r = tunnelMap.get(b.tunnelId);
-            if (r == null || !(r instanceof OriginTunnel)) {
-                logger.warn("server got confused with tunnel IDs on peer delete, ignoring message");
-                return;
-            }
-            OriginTunnel ot = (OriginTunnel) r;
-            if (ot.disconnectHandler != null) {
-                ot.disconnectHandler.onDisconnect(b.peer);
-            }
-        }
-
         public void visit(TunnelCreateMessage m) {
-            Tunnel t = new Tunnel();
-            t.tunnelId = m.tunnel_id;
+            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(UnicastMessage m) {
-            messageReceiver.setSender(m.oid);
-            messageReceiver.visitAppropriate(Construct.parseAs(m.payload, GnunetMessage.class).body);
-        }
-
-        public void visit(MulticastMessage m) {
-            messageReceiver.setSender(m.oid);
-            messageReceiver.visitAppropriate(Construct.parseAs(m.payload, GnunetMessage.class).body);
-        }
-
-        public void visit(OriginMessage m) {
-            messageReceiver.setSender(m.sender);
-            messageReceiver.visitAppropriate(Construct.parseAs(m.payload, GnunetMessage.class).body);
+        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) {
-                logger.warn("server got confused with tunnel IDs on ack, ignoring message");
-                return;
-            }
-            t.nextSentPacketId = m.maxPid;
-            t.onAckUpdated();
+            if (t != null)
+                t.ack_count += 1;
         }
 
         public void visit(TunnelDestroyMessage m) {
@@ -405,59 +250,28 @@ public class Mesh {
      *                             is called on the tunnel
      */
     public Mesh(Configuration cfg, InboundTunnelHandler inboundTunnelHandler,
-                TunnelEndHandler tunnelEndHandler, MeshRunabout messageReceiver, int... applications) {
+                TunnelEndHandler tunnelEndHandler, MeshRunabout messageReceiver, int... ports) {
         this.tunnelEndHandler = tunnelEndHandler;
         this.messageReceiver = messageReceiver;
-        this.applications = applications;
+        this.ports = ports;
         this.inboundTunnelHandler = inboundTunnelHandler;
 
-        Client client = new Client("mesh", cfg);
-        requestQueue = new RequestQueue(client, new MeshMessageReceiver());
-
-        requestQueue.add(new ClientConnectRequest());
-    }
-
-    /**
-     * Create a new tunnel (we're initiator and will be allowed to add/remove peers
-     * and to broadcast).
-     *
-     * @param connectHandler    callback for when a new peer connects to the tunnel, either because the origin added him,
-     *                          or the client joined the tunnel
-     * @param disconnectHandler callback for when when a peer is disconnected
-     */
-    public OriginTunnel createTunnel(ConnectHandler connectHandler, DisconnectHandler disconnectHandler) {
-        OriginTunnel tunnel = new OriginTunnel();
-        tunnel.connectHandler = connectHandler;
-        tunnel.disconnectHandler = disconnectHandler;
-        tunnel.tunnelId = nextTunnelId++;
-        tunnelMap.put(tunnel.tunnelId, tunnel);
-        tunnel.registerWithService();
-        return tunnel;
-    }
-
-
-    /**
-     * Announce to ther peer the availability of services described by the regex,
-     * in order to be reachable to other peers via connect_by_string.
-     * <p/>
-     * Note that the first 8 characters are considered to be part of a prefix,
-     * (for instance 'gnunet://'). If you put a variable part in there (*, +. ()),
-     * all matching strings will be stored in the DHT.
-     *
-     * @param regex string with the regular expression describing local services.
-     */
-    public void announceRegex(String regex) {
-        throw new UnsupportedOperationException("not implemented");
+        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. The registered inbound tunnel cleaner will be
-     * called should any inbound tunnels still exist.
+     * 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() {
-        requestQueue.destroy();
+        client_mq.destroy();
+        client.disconnect();
     }
 }
diff --git a/src/org/gnunet/mesh/MulticastMessage.java b/src/org/gnunet/mesh/MulticastMessage.java
deleted file mode 100644
index 2ca344d..0000000
--- a/src/org/gnunet/mesh/MulticastMessage.java
+++ /dev/null
@@ -1,27 +0,0 @@
-package org.gnunet.mesh;
-
-import org.gnunet.construct.*;
-import org.gnunet.util.GnunetMessage;
-import org.gnunet.util.PeerIdentity;
-
-/**
- * ...
- *
- * @author Florian Dold
- */
-@UnionCase(261)
-public class MulticastMessage implements GnunetMessage.Body {
-    /**
-     * Tunnel ID
-     */
-    @UInt32
-    public int tid;
-    @UInt32
-    public int ttl;
-    @UInt32
-    public int pid;
-    @NestedMessage
-    public PeerIdentity oid;
-    @FillWith @UInt8
-    public byte[] payload;
-}
diff --git a/src/org/gnunet/mesh/OriginMessage.java b/src/org/gnunet/mesh/OriginMessage.java
deleted file mode 100644
index 143eaeb..0000000
--- a/src/org/gnunet/mesh/OriginMessage.java
+++ /dev/null
@@ -1,27 +0,0 @@
-package org.gnunet.mesh;
-
-import org.gnunet.construct.*;
-import org.gnunet.util.GnunetMessage;
-import org.gnunet.util.PeerIdentity;
-
-/**
- * ...
- *
- * @author Florian Dold
- */
-@UnionCase(262)
-public class OriginMessage implements GnunetMessage.Body {
-    @UInt32
-    public int tid;
-    @UInt32
-    public int ttl;
-    @UInt32
-    public int pid;
-    @NestedMessage
-    public PeerIdentity oid;
-    @NestedMessage
-    public PeerIdentity sender;
-    @FillWith
-    @UInt8
-    public byte[] payload;
-}
diff --git a/src/org/gnunet/mesh/PeerAddMessage.java b/src/org/gnunet/mesh/PeerAddMessage.java
deleted file mode 100644
index 099165c..0000000
--- a/src/org/gnunet/mesh/PeerAddMessage.java
+++ /dev/null
@@ -1,23 +0,0 @@
-package org.gnunet.mesh;
-
-import org.gnunet.construct.NestedMessage;
-import org.gnunet.construct.UInt32;
-import org.gnunet.construct.UnionCase;
-import org.gnunet.util.GnunetMessage;
-import org.gnunet.util.PeerIdentity;
-
-/**
- * Message used for two things (bad!)
- * (1) client->server: request that a certain peer is added to a tunnel
- * (2) server->client: notify the client that a new peer has joined the tunnel
- *
- * @author Florian Dold
- */
-@UnionCase(275)
-public class PeerAddMessage implements GnunetMessage.Body {
-    @UInt32
-    public int tunnelId;
-
-    @NestedMessage
-    public PeerIdentity peer;
-}
diff --git a/src/org/gnunet/mesh/PeerDeleteMessage.java b/src/org/gnunet/mesh/PeerDeleteMessage.java
deleted file mode 100644
index 045e565..0000000
--- a/src/org/gnunet/mesh/PeerDeleteMessage.java
+++ /dev/null
@@ -1,23 +0,0 @@
-package org.gnunet.mesh;
-
-import org.gnunet.construct.NestedMessage;
-import org.gnunet.construct.UInt32;
-import org.gnunet.construct.UnionCase;
-import org.gnunet.util.GnunetMessage;
-import org.gnunet.util.PeerIdentity;
-
-/**
- * Message used for two things (bad!)
- * (1) client->server: request that a certain peer is added to a tunnel
- * (2) server->client: notify the client that a new peer has joined the tunnel
- *
- * @author Florian Dold
- */
-@UnionCase(276)
-public class PeerDeleteMessage implements GnunetMessage.Body {
-    @UInt32
-    public int tunnelId;
-
-    @NestedMessage
-    public PeerIdentity peer;
-}
diff --git a/src/org/gnunet/mesh/TunnelCreateMessage.java b/src/org/gnunet/mesh/TunnelCreateMessage.java
index 7b63bc6..eaa4d6c 100644
--- a/src/org/gnunet/mesh/TunnelCreateMessage.java
+++ b/src/org/gnunet/mesh/TunnelCreateMessage.java
@@ -7,11 +7,7 @@ import org.gnunet.util.GnunetMessage;
 import org.gnunet.util.PeerIdentity;
 
 /**
- * Message used to
- * a) request the service to create a new tunnel with the given tunnel id
- * b) notify the client of a newly created tunnel
- *
- * todo: this is bad design, split into two messages in the C code!
+ * FIXME
  *
  * @author Florian Dold
  */
@@ -20,9 +16,12 @@ public class TunnelCreateMessage implements GnunetMessage.Body {
     @UInt32
     public int tunnel_id;
 
-    /**
-     * Only present if sent from server to client (purpose b)
-     */
-    @NestedMessage(optional = true)
+    @NestedMessage(optional = false)
     public PeerIdentity otherEnd;
+
+    @UInt32
+    public int port;
+
+    @UInt32
+    public int opt;
 }
diff --git a/src/org/gnunet/mq/ClientMessageQueue.java b/src/org/gnunet/mq/ClientMessageQueue.java
new file mode 100644
index 0000000..bfb0466
--- /dev/null
+++ b/src/org/gnunet/mq/ClientMessageQueue.java
@@ -0,0 +1,48 @@
+package org.gnunet.mq;
+
+
+import org.gnunet.construct.Construct;
+import org.gnunet.util.*;
+
+/**
+ * Message queue for org.util.Connection
+ */
+public class ClientMessageQueue extends MessageQueue {
+    private final Client client;
+    private final RunaboutMessageReceiver receiver;
+
+
+    public ClientMessageQueue(Client client, RunaboutMessageReceiver receiver) {
+        this.client = client;
+        this.receiver = receiver;
+    }
+
+
+    public ClientMessageQueue(Client client) {
+        this(client, null);
+    }
+
+
+    @Override
+    protected void sendImmediate(final Envelope ev) {
+        int size = Construct.getSize(ev.message);
+        client.notifyTransmitReady(RelativeTime.FOREVER, false, size, new MessageTransmitter() {
+            @Override
+            public void transmit(Connection.MessageSink sink) {
+                sink.send(ev.message);
+                reportMessageSent();
+            }
+
+            @Override
+            public void handleError() {
+                // FIXME
+            }
+        });
+    }
+
+
+    @Override
+    public void destroy() {
+
+    }
+}
diff --git a/src/org/gnunet/mq/Envelope.java b/src/org/gnunet/mq/Envelope.java
new file mode 100644
index 0000000..09c0c2c
--- /dev/null
+++ b/src/org/gnunet/mq/Envelope.java
@@ -0,0 +1,34 @@
+package org.gnunet.mq;
+
+import org.gnunet.util.GnunetMessage;
+
+/**
+ * Container for a message to be sent by a message queue.
+ */
+public class Envelope {
+    public final GnunetMessage.Body message;
+    private MessageQueue parent_queue;
+    private NotifySentHandler notify_sent_handler;
+
+    public Envelope(GnunetMessage.Body message) {
+        this.message = message;
+    }
+
+    public void notifySent(NotifySentHandler h) {
+        this.notify_sent_handler = h;
+    }
+
+    public void injectSent() {
+        if (notify_sent_handler != null)
+            notify_sent_handler.onSent();
+    }
+
+    public void cancel() {
+        // TODO
+    }
+
+    /* pkg-private */ void invokeSentNotification() {
+        if (null != notify_sent_handler)
+            notify_sent_handler.onSent();
+    }
+}
diff --git a/src/org/gnunet/mq/MessageQueue.java b/src/org/gnunet/mq/MessageQueue.java
new file mode 100644
index 0000000..de08edf
--- /dev/null
+++ b/src/org/gnunet/mq/MessageQueue.java
@@ -0,0 +1,41 @@
+package org.gnunet.mq;
+
+
+import org.gnunet.util.GnunetMessage;
+
+import java.util.LinkedList;
+
+/**
+ * General-purpose message queue
+ */
+public abstract class MessageQueue {
+    private LinkedList<Envelope> queued_envelopes = new LinkedList<>();
+    protected Envelope current_envelope;
+
+    protected abstract void sendImmediate(Envelope ev);
+
+    public void send(GnunetMessage.Body body) {
+        send(new Envelope(body));
+    }
+
+    public void send(Envelope ev) {
+        if (null == current_envelope) {
+            current_envelope = ev;
+            sendImmediate(current_envelope);
+        } else {
+            queued_envelopes.addLast(ev);
+        }
+    }
+
+    protected void reportMessageSent() {
+        if (null == current_envelope)
+            throw new AssertionError();
+        current_envelope.invokeSentNotification();
+        if (queued_envelopes.isEmpty())
+            return;
+        current_envelope = queued_envelopes.pop();
+        sendImmediate(current_envelope);
+    }
+
+    public abstract void destroy();
+}
diff --git a/src/org/gnunet/mq/NotifySentHandler.java b/src/org/gnunet/mq/NotifySentHandler.java
new file mode 100644
index 0000000..7ec13b2
--- /dev/null
+++ b/src/org/gnunet/mq/NotifySentHandler.java
@@ -0,0 +1,6 @@
+package org.gnunet.mq;
+
+
+public interface NotifySentHandler {
+    void onSent();
+}
diff --git a/src/org/gnunet/requests/RequestQueue.java b/src/org/gnunet/requests/RequestQueue.java
index e514303..6f7102d 100644
--- a/src/org/gnunet/requests/RequestQueue.java
+++ b/src/org/gnunet/requests/RequestQueue.java
@@ -28,7 +28,6 @@ import java.util.LinkedList;
  * Generic queues for Requests to be sent to the service.
  */
 public class RequestQueue {
-    // todo: implement more efficiently (attributes instead of multiple queues)
 
     /**
      * Requests to be transmitted to the service.
diff --git a/src/org/gnunet/transport/RequestConnectMessage.java b/src/org/gnunet/transport/RequestConnectMessage.java
index 9bd7a5b..63bbc39 100644
--- a/src/org/gnunet/transport/RequestConnectMessage.java
+++ b/src/org/gnunet/transport/RequestConnectMessage.java
@@ -21,5 +21,4 @@ public class RequestConnectMessage implements GnunetMessage.Body {
      */
     @NestedMessage
     public PeerIdentity peer;
-
 }
diff --git a/src/org/gnunet/transport/Transport.java b/src/org/gnunet/transport/Transport.java
index f79c340..5ccbc77 100644
--- a/src/org/gnunet/transport/Transport.java
+++ b/src/org/gnunet/transport/Transport.java
@@ -1,6 +1,7 @@
 package org.gnunet.transport;
 
 import org.gnunet.hello.HelloMessage;
+import org.gnunet.mq.Envelope;
 import org.gnunet.util.*;
 
 /**
@@ -25,7 +26,19 @@ public class Transport {
      *         NULL on failure (cb will not be called)
      */
     Cancelable tryConnect(PeerIdentity target, TryConnectCallback cb) {
-        throw new UnsupportedOperationException();
+        RequestConnectMessage m = new RequestConnectMessage();
+        m.peer = target;
+        m.reserved = 0;
+        final Envelope ev = new Envelope(m);
+        ev.notifySent(null /* FIXME */);
+        //client_mq.send(ev);
+
+        return new Cancelable() {
+            @Override
+            public void cancel() {
+                ev.cancel();
+            }
+        };
     }