Let's make it easy for anyone to indicate their expected connection, and use it

to suppress subscription failure events in addition to subscription success, as
well as event forwards (though those shouldn't, in theory, go out over a new
connection because subscribers are cleared when a connection is dropped).
Nothing else looked like it might result in a spurious message being sent if a
connection was dropped and reestablished, so none of the other calls were made
to supply an expected connection.


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6376 542714f4-19e9-0310-aa3c-eee0fc999fb1

src/main/java/com/threerings/presents/peer/server/PeerSession.java

@@ -32,6 +32,7 @@ import com.threerings.presents.net.Message;
 import com.threerings.presents.peer.data.NodeObject;
 import com.threerings.presents.peer.net.PeerBootstrapData;
 import com.threerings.presents.server.PresentsSession;
+import com.threerings.presents.server.net.PresentsConnection;
 
 import static com.threerings.presents.Log.log;
 
@@ -115,9 +116,9 @@ public class PeerSession extends PresentsSession
     }
 
     @Override // from PresentsSession
-    protected final boolean postMessage (DownstreamMessage msg)
+    protected final boolean postMessage (DownstreamMessage msg, PresentsConnection expect)
     {
-        if (!super.postMessage(msg)) {
+        if (!super.postMessage(msg, expect)) {
             return false;
         }
         _stats.peerMessagesOut++;

src/main/java/com/threerings/presents/server/PresentsSession.java

@@ -556,7 +556,7 @@ public class PresentsSession
         synchronized(_pendingThrottles) {
             _pendingThrottles.add(_messagesPerSec);
         }
-        postMessage(new UpdateThrottleMessage(_messagesPerSec));
+        postMessage(new UpdateThrottleMessage(_messagesPerSec), null);
         // when we get a ThrottleUpdatedMessage from the client, we'll apply the new throttle
     }
 
@@ -706,7 +706,7 @@ public class PresentsSession
         populateBootstrapData(data);
 
         // create a send bootstrap notification
-        postMessage(new BootstrapNotification(data));
+        postMessage(new BootstrapNotification(data), null);
     }
 
     /**
@@ -856,15 +856,25 @@ public class PresentsSession
     {
         _omgr.postRunnable(new Runnable() {
             public void run () {
-                postMessage(msg);
+                postMessage(msg, null);
             }
         });
     }
 
     /** Queues a message for delivery to the client. */
-    protected boolean postMessage (DownstreamMessage msg)
+    protected boolean postMessage (DownstreamMessage msg, PresentsConnection expect)
     {
         PresentsConnection conn = getConnection();
+
+        // make sure that the connection they expect us to be using is the one we're using; there
+        // are circumstances were sufficient delay between request and response gives the client
+        // time to drop their original connection and establish a new one, opening the door to
+        // major confusion
+        if (expect != null && conn != expect) {
+            return false;
+        }
+
+        // make sure we have a connection at all
         if (conn != null) {
             conn.postMessage(msg);
             _messagesOut++; // count 'em up!
@@ -958,11 +968,6 @@ public class PresentsSession
     {
         public DObject object;
 
-        public ClientProxy ()
-        {
-            _oconn = getConnection();
-        }
-
         public void unsubscribe ()
         {
             object.removeSubscriber(this);
@@ -972,8 +977,7 @@ public class PresentsSession
         // from interface ProxySubscriber
         public void objectAvailable (DObject dobj)
         {
-            // ensure that the connection is the same one that requested the subscription
-            if (getConnection() == _oconn && postMessage(new ObjectResponse<DObject>(dobj))) {
+            if (postMessage(new ObjectResponse<DObject>(dobj), _oconn)) {
                 _firstEventId = _omgr.getNextEventId(false);
                 object = dobj;
                 ClientProxy orec;
@@ -997,7 +1001,7 @@ public class PresentsSession
         // from interface ProxySubscriber
         public void requestFailed (int oid, ObjectAccessException cause)
         {
-            postMessage(new FailureResponse(oid, cause.getMessage()));
+            postMessage(new FailureResponse(oid, cause.getMessage()), _oconn);
         }
 
         // from interface ProxySubscriber
@@ -1017,7 +1021,7 @@ public class PresentsSession
                 return;
             }
 
-            postMessage(new EventNotification(event));
+            postMessage(new EventNotification(event), _oconn);
         }
 
         // from interface ProxySubscriber
@@ -1027,7 +1031,8 @@ public class PresentsSession
         }
 
         protected long _firstEventId;
-        protected PresentsConnection _oconn;
+        // the connection that was active at the time we were constructed
+        protected PresentsConnection _oconn = getConnection();
     }
 
     /**