Buffering past requests so that the client can request them again in case of...

Buffering past requests so that the client can request them again in case of network failure. Needed for full compliance of section 12 of the XEP, all that remains for this section is to make sure requests are executed in the correct order.

git-svn-id: http://svn.igniterealtime.org/svn/repos/wildfire/trunk@5956 b35dd754-fafc-0310-a699-88a17e54d16e

src/java/org/jivesoftware/wildfire/SessionManager.java

@@ -542,13 +542,13 @@ public class SessionManager extends BasicModule {
         return session;
     }
 
-    public HttpSession createClientHttpSession(InetAddress address, StreamID id)
+    public HttpSession createClientHttpSession(long rid, InetAddress address, StreamID id)
             throws UnauthorizedException
     {
         if (serverName == null) {
             throw new UnauthorizedException("Server not initialized");
         }
-        HttpSession session = new HttpSession(serverName, address, id);
+        HttpSession session = new HttpSession(serverName, address, id, rid);
         Connection conn = session.getConnection();
         conn.init(session);
         conn.registerCloseListener(clientSessionListener, session);

src/java/org/jivesoftware/wildfire/http/HttpSession.java

@@ -38,10 +38,11 @@ import java.net.InetAddress;
  */
 public class HttpSession extends ClientSession {
     private int wait;
-    private int hold = -1000;
+    private int hold = 0;
     private String language;
     private final Queue<HttpConnection> connectionQueue = new LinkedList<HttpConnection>();
     private final List<Deliverable> pendingElements = new ArrayList<Deliverable>();
+    private final List<Deliverable> sentElements = new ArrayList<Deliverable>();
     private boolean isSecure;
     private int maxPollingInterval;
     private long lastPoll = -1;
@@ -49,14 +50,42 @@ public class HttpSession extends ClientSession {
     private boolean isClosed;
     private int inactivityTimeout;
     private long lastActivity;
+    private long lastRequestID;
 
-    public HttpSession(String serverName, InetAddress address, StreamID streamID) {
+    public HttpSession(String serverName, InetAddress address, StreamID streamID, long rid) {
         super(serverName, null, streamID);
         conn = new HttpVirtualConnection(address);
         this.lastActivity = System.currentTimeMillis();
+        this.lastRequestID = rid;
     }
 
-    void addConnection(HttpConnection connection, boolean isPoll) throws HttpBindException,
+    HttpConnection createConnection(long rid, boolean isPoll, boolean isSecure)
+            throws HttpConnectionClosedException, HttpBindException
+    {
+        HttpConnection connection = new HttpConnection(rid, isSecure);
+        if(rid <= lastRequestID) {
+            Deliverable deliverable = retrieveDeliverable(rid);
+            connection.deliverBody(deliverable.getDeliverable());
+            return connection;
+        }
+        else if (rid > lastRequestID + hold + 1) {
+            throw new HttpBindException("Unexpected RID Error", true, 404);
+        }
+
+        addConnection(connection, isPoll);
+        return connection;
+    }
+
+    private Deliverable retrieveDeliverable(long rid) throws HttpBindException {
+        for(Deliverable delivered : sentElements) {
+            if(delivered.getRequestID() == rid) {
+                return delivered;
+            }
+        }
+        throw new HttpBindException("Unexpected RID Error", true, 404);
+    }
+
+    private void addConnection(HttpConnection connection, boolean isPoll) throws HttpBindException,
             HttpConnectionClosedException
     {
         if(connection == null) {
@@ -77,7 +106,7 @@ public class HttpSession extends ClientSession {
             String deliverable = createDeliverable(pendingElements);
             pendingElements.clear();
             fireConnectionOpened(connection);
-            connection.deliverBody(deliverable);
+            deliver(connection, deliverable);
             fireConnectionClosed(connection);
         }
         else {
@@ -91,6 +120,21 @@ public class HttpSession extends ClientSession {
             connectionQueue.offer(connection);
             fireConnectionOpened(connection);
         }
+        lastRequestID = connection.getRequestId();
+    }
+
+    private void deliver(HttpConnection connection, String deliverable)
+            throws HttpConnectionClosedException
+    {
+        connection.deliverBody(deliverable);
+
+        Deliverable delivered = new Deliverable(deliverable);
+        delivered.setRequestID(connection.getRequestId());
+        while(sentElements.size() > hold) {
+            sentElements.remove(0);
+        }
+
+        sentElements.add(delivered);
     }
 
     private void fireConnectionOpened(HttpConnection connection) {
@@ -203,7 +247,7 @@ public class HttpSession extends ClientSession {
         while(!delivered && connectionQueue.size() > 0) {
             HttpConnection connection = connectionQueue.remove();
             try {
-                connection.deliverBody(deliverable);
+                deliver(connection, deliverable);
                 delivered = true;
                 fireConnectionClosed(connection);
             }
@@ -381,9 +425,10 @@ public class HttpSession extends ClientSession {
         }
     }
 
-    private class Deliverable {
+    private class Deliverable implements Comparable<Deliverable> {
         private final String text;
         private final Packet packet;
+        private long requestID;
 
         public Deliverable(String text) {
             this.text = text;
@@ -403,5 +448,17 @@ public class HttpSession extends ClientSession {
                 return text;
             }
         }
+
+        public void setRequestID(long requestID) {
+            this.requestID = requestID;
+        }
+
+        public long getRequestID() {
+            return requestID;
+        }
+
+        public int compareTo(Deliverable o) {
+            return (int) (o.getRequestID() - requestID);
+        }
     }
 }

src/java/org/jivesoftware/wildfire/http/HttpSessionManager.java

@@ -128,7 +128,7 @@ public class HttpSessionManager {
         int wait = getIntAttribute(rootNode.attributeValue("wait"), 60);
         int hold = getIntAttribute(rootNode.attributeValue("hold"), 1);
 
-        HttpSession session = createSession(address);
+        HttpSession session = createSession(connection.getRequestId(), address);
         session.setWait(wait);
         session.setHold(hold);
         session.setSecure(connection.isSecure());
@@ -149,11 +149,11 @@ public class HttpSessionManager {
         return session;
     }
 
-    private HttpSession createSession(InetAddress address) throws UnauthorizedException {
+    private HttpSession createSession(long rid, InetAddress address) throws UnauthorizedException {
         // Create a ClientSession for this user.
         StreamID streamID = SessionManager.getInstance().nextStreamID();
         // Send to the server that a new client session has been created
-        HttpSession session = sessionManager.createClientHttpSession(address, streamID);
+        HttpSession session = sessionManager.createClientHttpSession(rid, address, streamID);
         // Register that the new session is associated with the specified stream ID
         sessionMap.put(streamID.getID(), session);
         session.addSessionCloseListener(new SessionListener() {
@@ -226,8 +226,7 @@ public class HttpSessionManager {
         //noinspection unchecked
         List<Element> elements = rootNode.elements();
         boolean isPoll = elements.size() <= 0;
-        HttpConnection connection = new HttpConnection(rid, isSecure);
-        session.addConnection(connection, isPoll);
+        HttpConnection connection = session.createConnection(rid, isPoll, isSecure);
         SessionPacketRouter router = new SessionPacketRouter(session);
 
         for (Element packet : elements) {