1 files changed, 42 insertions, 26 deletions

diff --git a/src/main/java/org/gnunet/mq/MessageQueue.java b/src/main/java/org/gnunet/mq/MessageQueue.java
index 4df3ae4..cab1003 100644
--- a/src/main/java/org/gnunet/mq/MessageQueue.java
+++ b/src/main/java/org/gnunet/mq/MessageQueue.java
@@ -11,10 +11,19 @@ import java.util.LinkedList;
 public abstract class MessageQueue {
     private LinkedList<Envelope> queued_envelopes = new LinkedList<Envelope>();
     private LinkedList<Envelope> prefered_queued_envelopes = new LinkedList<Envelope>();
-    protected Envelope current_envelope;
+    protected Envelope currentEnvelope;
+    private boolean readyForSubmit;
 
+    /**
+     * Submit a message. Once the message can't be canceled,
+     * reportMessageSent must be called.
+     * @param ev envelope to submit
+     */
     protected abstract void submit(Envelope ev);
 
+    /**
+     * Cancel a message that already has been submitted.
+     */
     protected abstract void retract();
 
     public void send(GnunetMessage.Body body) {
@@ -25,6 +34,16 @@ public abstract class MessageQueue {
         sendPrefered(new Envelope(body));
     }
 
+    public void send(Envelope ev) {
+        queued_envelopes.addLast(ev);
+        trySubmitNext();
+    }
+
+    public void sendPrefered(Envelope ev) {
+        prefered_queued_envelopes.addLast(ev);
+        trySubmitNext();
+    }
+
     private Envelope pollNextEnvelope() {
         if (!prefered_queued_envelopes.isEmpty())
             return prefered_queued_envelopes.removeFirst();
@@ -33,36 +52,33 @@ public abstract class MessageQueue {
         return null;
     }
 
-    public void send(Envelope ev) {
-        if (null == current_envelope) {
-            current_envelope = ev;
-            submit(current_envelope);
-        } else {
-            queued_envelopes.addLast(ev);
+    protected void trySubmitNext() {
+        if (currentEnvelope != null || !readyForSubmit) {
+            return;
+        }
+        Envelope ev = pollNextEnvelope();
+        if (ev == null) {
+            return;
         }
+        currentEnvelope = ev;
+        readyForSubmit = false;
+        submit(currentEnvelope);
     }
 
-    public void sendPrefered(Envelope ev) {
-        if (null == current_envelope) {
-            current_envelope = ev;
-            submit(current_envelope);
-        } else {
-            prefered_queued_envelopes.addLast(ev);
+    protected void reportReadyForSubmit() {
+        if (readyForSubmit) {
+            throw new AssertionError("message queue reported 'ready for submit' twice");
         }
+        readyForSubmit = true;
+        trySubmitNext();
     }
 
     protected void reportMessageSent() {
-        if (null == current_envelope)
+        if (null == currentEnvelope)
             throw new AssertionError();
-        current_envelope.invokeSentNotification();
-        next();
-    }
-
-    private void next() {
-        current_envelope = pollNextEnvelope();
-        if (current_envelope == null)
-            return;
-        submit(current_envelope);
+        currentEnvelope.invokeSentNotification();
+        currentEnvelope = null;
+        trySubmitNext();
     }
 
     /**
@@ -71,11 +87,11 @@ public abstract class MessageQueue {
      * @param ev the envelope to cancel
      */
     /* pkg-private */ void cancelEnvelope(Envelope ev) {
-        if (null == current_envelope)
+        if (null == currentEnvelope)
             throw new AssertionError();
-        if (ev == current_envelope) {
+        if (ev == currentEnvelope) {
             retract();
-            next();
+            trySubmitNext();
         } else {
             queued_envelopes.remove(ev);
             prefered_queued_envelopes.remove(ev);