- simplify request containers

- statistics bugs
- fix missing annotation in consensus api
- work on voting

4 files changed, 200 insertions, 172 deletions

diff --git a/src/main/java/org/gnunet/requests/MatchingRequestContainer.java b/src/main/java/org/gnunet/requests/MatchingRequestContainer.java
index b2c671e..04b17fd 100644
--- a/src/main/java/org/gnunet/requests/MatchingRequestContainer.java
+++ b/src/main/java/org/gnunet/requests/MatchingRequestContainer.java
@@ -1,3 +1,24 @@
+/*
+ This file is part of GNUnet.
+ (C) 2014 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.requests;
 
 import com.google.common.collect.Maps;
@@ -8,13 +29,13 @@ import java.util.Map;
 
 
 /**
- * Container for requests that are responded to with a matching request identification
+ * Container for requests that are responded to with a matching getRequestIdentifier identification
  */
-public class MatchingRequestContainer<K, T extends RequestContainer.Request> extends RequestContainer {
+public class MatchingRequestContainer<K, T extends Request> extends RequestContainer {
     /**
      * All queued requests.
      */
-    private Map<K,T> requests = Maps.newHashMap();
+    private Map<K,Identifier> requests = Maps.newHashMap();
 
     /**
      * Message queue to send to requests over.
@@ -25,48 +46,48 @@ public class MatchingRequestContainer<K, T extends RequestContainer.Request> ext
         this.mq = mq;
     }
 
+    private class Identifier extends SimpleRequestIdentifier<T> {
+        final K key;
+
+        public Identifier(T request, K key) {
+            super(request);
+            this.key = key;
+        }
+
+        @Override
+        public void retire() {
+            super.retire();
+            requests.remove(key);
+        }
+    }
+
     public Cancelable addRequest(K key, final T request) {
         if (requests.containsKey(key))
-            throw new AssertionError("key already present in request container");
-        requests.put(key, request);
-        mq.send(request.assembleRequest());
-        return new Cancelable() {
-            @Override
-            public void cancel() {
-                setRequestCancelled(request);
-                if (isRequestTransmitting(request)) {
-                    cancelRequestTransmission(request);
-                } else {
-                    request.cancel();
-                }
-            }
-        };
+            throw new AssertionError("key already present in getRequestIdentifier container");
+        Identifier identifier = new Identifier(request, key);
+        requests.put(key, identifier);
+        identifier.send(mq);
+        return identifier;
     }
 
     @Override
     public void restart() {
-        Map<K,T> requestsOld = requests;
+        Map<K, Identifier> requestsOld = requests;
         requests = Maps.newHashMap();
-        for (Map.Entry<K,T> e : requestsOld.entrySet()) {
-            if (!isRequestCancelled(e.getValue())) {
-                setRequestTransmitting(e.getValue(), false);
-                addRequest(e.getKey(), e.getValue());
-            }
+        for (Map.Entry<K,Identifier> e : requestsOld.entrySet()) {
+            addRequest(e.getKey(), e.getValue().getRequest());
         }
     }
 
-    public T getRequest(K key) {
+    public RequestIdentifier<T> getRequestIdentifier(K key) {
         return requests.get(key);
     }
 
-    /**
-     * Retrieve the request matching the given key, and remove it.
-     * Return null if there is no matching request.
-     *
-     * @param key key to look for
-     * @return request, or null
-     */
-    public T pollRequest(K key) {
-        return requests.remove(key);
+    public T getAndRetireRequest(K key) {
+        RequestIdentifier<T> i = getRequestIdentifier(key);
+        if (null == i)
+            return null;
+        i.retire();
+        return i.getRequest();
     }
 }
diff --git a/src/main/java/org/gnunet/requests/RequestContainer.java b/src/main/java/org/gnunet/requests/RequestContainer.java
index 8fc631e..9c7fb48 100644
--- a/src/main/java/org/gnunet/requests/RequestContainer.java
+++ b/src/main/java/org/gnunet/requests/RequestContainer.java
@@ -1,90 +1,31 @@
-package org.gnunet.requests;
-
+/*
+ This file is part of GNUnet.
+ (C) 2014 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.
+ */
 
-import org.gnunet.mq.Envelope;
-import org.gnunet.util.Cancelable;
+package org.gnunet.requests;
 
 /**
  * Container for requests to a service.
  */
 public abstract class RequestContainer {
     /**
-     * Do we allow sending pending requests while
-     * other requests are still active (true), or do we send
-     * requests one-by-one (false)?
-     */
-    protected boolean overlap = true;
-
-    /**
-     * A request that can be put in a request container.
-     */
-    public abstract static class Request {
-        private boolean transmitting;
-        private boolean canceled;
-        private Cancelable cancelRequest;
-
-        /**
-         * Create an envelope for executing the request.
-         *
-         * @return Envelope to be sent to the service with the request message.
-         */
-        public abstract Envelope assembleRequest();
-
-        /**
-         * Implement cancellation logic for requests.
-         * Only called when the request has already been sent to the service.
-         *
-         * By default, canceling a request throws an exception.
-         */
-        public void cancel() {
-            throw new AssertionError("request of type " + this.getClass() + " can not be canceled (not implemented)");
-        }
-    }
-
-    /**
-     * Re-send all requests in the queue that have not been canceled.
+     * Re-send all active and pending requests.
      */
     public abstract void restart();
-
-    /**
-     * Allow or disallow requests to be send while other requests in the queue have not been completed.
-     *
-     * @param overlap true to allow overlapped requests, false to disallow them
-     */
-    public void setOverlap(boolean overlap) {
-        this.overlap = overlap;
-    }
-
-    /**
-     * Check if the given request is transmitting, that is, the request
-     * is waiting to be sent to the service.
-     *
-     * @param r request
-     * @return whether the request is still being transmitted
-     */
-    protected boolean isRequestTransmitting(Request r) {
-        return r.transmitting;
-    }
-
-    protected void setRequestTransmitting(Request r, boolean transmitting) {
-        r.transmitting = transmitting;
-    }
-
-    protected void setRequestTransmissionCancel(Request request, Cancelable cancel) {
-        request.cancelRequest = cancel;
-    }
-
-    protected void cancelRequestTransmission(Request r) {
-        r.cancelRequest.cancel();
-        r.cancelRequest = null;
-    }
-
-    protected void setRequestCancelled(Request r) {
-        r.canceled = true;
-    }
-
-    protected boolean isRequestCancelled(Request r) {
-        return r.canceled;
-    }
-
 }
diff --git a/src/main/java/org/gnunet/requests/SequentialRequestContainer.java b/src/main/java/org/gnunet/requests/SequentialRequestContainer.java
index 037055f..c343812 100644
--- a/src/main/java/org/gnunet/requests/SequentialRequestContainer.java
+++ b/src/main/java/org/gnunet/requests/SequentialRequestContainer.java
@@ -1,86 +1,153 @@
+/*
+ This file is part of GNUnet.
+ (C) 2014 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.requests;
 
-import org.gnunet.mq.Envelope;
 import org.gnunet.mq.MessageQueue;
-import org.gnunet.mq.NotifySentHandler;
-import org.gnunet.util.Cancelable;
 
 import java.util.LinkedList;
 
 /**
  * Container for requests that are responded to in sequential order.
  */
-public class SequentialRequestContainer<T extends RequestContainer.Request> extends RequestContainer {
-    private LinkedList<T> requests = new LinkedList<T>();
-    private MessageQueue mq;
+public class SequentialRequestContainer<T extends Request> extends RequestContainer {
+    /**
+     * Allow overlapping requests.
+     */
+    private boolean overlap;
+    /**
+     * Request in our queue with information about them.
+     */
+    private LinkedList<Identifier> requests = new LinkedList<Identifier>();
+    /**
+     * Message queue that is used to send envelopes.
+     */
+    private final MessageQueue mq;
 
-    int requestsActive = 0;
+    /**
+     * Number of active requests.
+     */
+    private int requestsActive = 0;
 
-    public SequentialRequestContainer(MessageQueue mq) {
+    /**
+     * Create a sequential request container that sends messages with the given
+     * message queue.
+     *
+     * @param mq message queue to send messages with
+     * @param overlap allow sending requests while other request have not yet completed
+     */
+    public SequentialRequestContainer(MessageQueue mq, boolean overlap) {
         this.mq = mq;
+        this.overlap = overlap;
+    }
+    /**
+     * Create a sequential request container that sends messages with the given
+     * message queue.  Do not allow other requests to be send while the current request
+     * is still active.
+     *
+     * @param mq message queue to send messages with
+     */
+    public SequentialRequestContainer(MessageQueue mq) {
+       this(mq, false);
+    }
+    /**
+     * Get the current request's identifier.
+     *
+     * @return current request
+     */
+    public RequestIdentifier<T> getRequestIdentifier() {
+        return requests.peekFirst();
     }
 
+    /**
+     * Get the current request and retire it.
+     * If there is no current request, null will be returned.
+     *
+     * @return current request
+     */
+    public T getAndRetireRequest() {
+        RequestIdentifier<T> i = getRequestIdentifier();
+        if (null == i)
+            return null;
+        i.retire();
+        return i.getRequest();
+    }
+
+    public Iterable<RequestIdentifier<T>> iter() {
+        return (Iterable) requests;
+    }
+
+    /**
+     * Get the current request.
+     *
+     * @return the current request.
+     */
     public T getRequest() {
-        return requests.getFirst();
+        RequestIdentifier<T> i = getRequestIdentifier();
+        if (null == i)
+            return null;
+        return i.getRequest();
     }
 
-    public void next() {
-        if (requestsActive == 0 || requests.isEmpty())
-            throw new AssertionError();
-        requestsActive--;
-        requests.removeFirst();
-        if (requestsActive == 0 && !requests.isEmpty()) {
-            Request r = requests.getFirst();
-            setRequestTransmitting(r, true);
-            Envelope ev = r.assembleRequest();
-            setRequestTransmissionCancel(r, ev);
-            mq.send(r.assembleRequest());
-            requestsActive++;
+    /**
+     * A request identifier in a sequential request container.
+     */
+    private class Identifier extends SimpleRequestIdentifier<T> {
+        public Identifier(T request) {
+            super(request);
+        }
+
+        @Override
+        public void retire() {
+            super.retire();
+            if (requestsActive == 0 || requests.isEmpty())
+                throw new AssertionError();
+            boolean found = requests.remove(this);
+            if (!found)
+                throw new AssertionError("request not in queue");
+            requestsActive--;
+            Identifier next = requests.peekFirst();
+            if (null == next || next.queued)
+                return;
+            if (requestsActive == 0 || overlap)
+                next.send(mq);
         }
     }
 
-    public Cancelable addRequest(final T request) {
-        requests.addLast(request);
+    public RequestIdentifier<T> addRequest(final T request) {
+        final Identifier identifier = new Identifier(request);
+        requests.addLast(identifier);
+        // only send immediately if we are allowed to
         if (overlap || requestsActive == 0) {
-            requestsActive++;
-            setRequestTransmitting(request, true);
-            Envelope ev = request.assembleRequest();
-            ev.notifySent(new NotifySentHandler() {
-                @Override
-                public void onSent() {
-                    setRequestTransmitting(request, false);
-                }
-            });
-            setRequestTransmissionCancel(request, ev);
-            mq.send(request.assembleRequest());
+            identifier.send(mq);
+            requestsActive += 1;
         }
-        return new Cancelable() {
-            @Override
-            public void cancel() {
-                setRequestCancelled(request);
-                if (isRequestTransmitting(request)) {
-                    cancelRequestTransmission(request);
-                } else {
-                    request.cancel();
-                }
-            }
-        };
+        return identifier;
     }
 
     @Override
     public void restart() {
-        LinkedList<T> requestsOld = requests;
-        requests = new LinkedList<T>();
-        for (T r : requestsOld) {
-            if (!isRequestCancelled(r)) {
-                setRequestTransmitting(r, false);
-                addRequest(r);
-            }
+        LinkedList<Identifier> requestsOld = requests;
+        requests = new LinkedList<Identifier>();
+        for (Identifier r : requestsOld) {
+            addRequest(r.getRequest());
         }
     }
-
-
-    public Iterable<T> iter() {
-        return requests;
-    }
 }
diff --git a/src/main/java/org/gnunet/requests/package-info.java b/src/main/java/org/gnunet/requests/package-info.java
index 892a606..b4fd674 100644
--- a/src/main/java/org/gnunet/requests/package-info.java
+++ b/src/main/java/org/gnunet/requests/package-info.java
@@ -1,6 +1,6 @@
 /*
  This file is part of GNUnet.
- (C) 2011, 2012 Christian Grothoff (and other contributing authors)
+ (C) 2014 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
@@ -18,7 +18,6 @@
  Boston, MA 02111-1307, USA.
  */
 
-
 /**
  * General mechanism for queueing requests to a service.
  */