Allow downstream and upstream messages to be compounded

A client can collect a bunch of messages into a CompoundUpstreamMessage, which are unpacked and
dispatched in order by the server. The server collects any messages sent while processing that
compound message into a CompoundDownstreamMessage, which is unpacked and dispatched by the client.

This allows the client to indicate that it's going to send enough messages that it'd blow out the
throttle, but that it's doing so intentionally.



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

src/main/as/com/threerings/presents/client/Client.as

@@ -20,7 +20,6 @@
 // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 
 package com.threerings.presents.client {
-
 import flash.events.EventDispatcher;
 import flash.events.TimerEvent;
 
@@ -280,6 +279,21 @@ public class Client extends EventDispatcher
         return true;
     }
 
+    /**
+     * Runs the given function while collecting any generated messages in a CompoundEvent, which
+     * will be sent when the function returns.
+     */
+    public function inCompoundMessage (run :Function) :void
+    {
+        _comm.startCompoundMessage();
+        try {
+            run();
+        } finally {
+            _comm.finishCompoundMessage();
+        }
+
+    }
+
     /**
      * Create the client property object to pass to {@link ObjectInputStream}. Subclasses may
      * add references useful during e.g. deserialization.

src/main/as/com/threerings/presents/client/ClientDObjectMgr.as

@@ -20,6 +20,7 @@
 // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 
 package com.threerings.presents.client {
+import com.threerings.presents.net.CompoundDownstreamMessage;
 
 import flash.events.Event;
 import flash.events.TimerEvent;
@@ -175,6 +176,10 @@ public class ClientDObjectMgr
 
         } else if (msg is UpdateThrottleMessage) {
             _client.setOutgoingMessageThrottle((msg as UpdateThrottleMessage).messagesPerSec);
+        } else if (msg is CompoundDownstreamMessage) {
+            for each (var submsg :DownstreamMessage in CompoundDownstreamMessage(msg).msgs) {
+                processMessage(submsg);
+            }
         }
     }
 

src/main/as/com/threerings/presents/client/Communicator.as

@@ -20,6 +20,7 @@
 // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 
 package com.threerings.presents.client {
+import com.threerings.presents.net.CompoundUpstreamMessage;
 
 import flash.events.Event;
 import flash.events.IOErrorEvent;
@@ -75,6 +76,10 @@ public class Communicator
 
     public function postMessage (msg :UpstreamMessage) :void
     {
+        if (_compound != null) {
+            _compound.msgs.push(msg);
+            return;
+        }
         _outq.push(msg);
         if (_writer != null) {
             sendPendingMessages(null);
@@ -109,6 +114,23 @@ public class Communicator
         return false;
     }
 
+    public function startCompoundMessage () :void
+    {
+        if (_compound == null) {
+            _compound = new CompoundUpstreamMessage();
+        }
+        _compoundDepth++;
+    }
+
+    public function finishCompoundMessage () :void
+    {
+        if (--_compoundDepth == 0) {
+            var toSend :CompoundUpstreamMessage = _compound;
+            _compound = null;
+            postMessage(toSend);
+        }
+    }
+
     /**
      * Attempts to logon on using the port at the specified index.
      */
@@ -383,6 +405,13 @@ public class Communicator
     protected var _writer :Timer;
     protected var _notedThrottle :Boolean = false;
 
+    protected var _compound :CompoundUpstreamMessage;
+
+    /**
+     * The count of startCompoundMessage calls that have occurred without a finishCompoundMessage
+     */
+    protected var _compoundDepth :int;
+
     protected const log :Log = Log.getLog(this);
 
     /** The current port we'll try to connect to. */

src/main/as/com/threerings/presents/net/CompoundDownstreamMessage.as

@@ -0,0 +1,38 @@
+//
+// $Id$
+//
+// Narya library - tools for developing networked games
+// Copyright (C) 2002-2011 Three Rings Design, Inc., All Rights Reserved
+// http://code.google.com/p/narya/
+//
+// This library is free software; you can redistribute it and/or modify it
+// under the terms of the GNU Lesser General Public License as published
+// by the Free Software Foundation; either version 2.1 of the License, or
+// (at your option) any later version.
+//
+// This library 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
+// Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public
+// License along with this library; if not, write to the Free Software
+// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+package com.threerings.presents.net {
+import com.threerings.io.streamers.ArrayStreamer;
+import com.threerings.io.TypedArray;
+import com.threerings.io.ObjectInputStream;
+
+public class CompoundDownstreamMessage extends DownstreamMessage
+{
+    public var msgs :TypedArray = TypedArray.create(DownstreamMessage);
+
+    override public function readObject (ins :ObjectInputStream) :void
+    {
+        super.readObject(ins);
+
+        msgs = ins.readField(ArrayStreamer.INSTANCE);
+    }
+}
+}

src/main/as/com/threerings/presents/net/CompoundUpstreamMessage.as

@@ -0,0 +1,41 @@
+//
+// $Id$
+//
+// Narya library - tools for developing networked games
+// Copyright (C) 2002-2011 Three Rings Design, Inc., All Rights Reserved
+// http://code.google.com/p/narya/
+//
+// This library is free software; you can redistribute it and/or modify it
+// under the terms of the GNU Lesser General Public License as published
+// by the Free Software Foundation; either version 2.1 of the License, or
+// (at your option) any later version.
+//
+// This library 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
+// Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public
+// License along with this library; if not, write to the Free Software
+// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+package com.threerings.presents.net {
+
+import com.threerings.io.ObjectInputStream;
+import com.threerings.io.ObjectOutputStream;
+import com.threerings.io.TypedArray;
+import com.threerings.io.streamers.ArrayStreamer;
+
+import com.threerings.presents.net.UpstreamMessage;
+
+public class CompoundUpstreamMessage extends UpstreamMessage
+{
+    public var msgs :TypedArray = TypedArray.create(UpstreamMessage);
+
+    override public function writeObject (out :ObjectOutputStream) :void
+    {
+        super.writeObject(out);
+        out.writeField(msgs, ArrayStreamer.INSTANCE);
+    }
+}
+}

src/main/as/com/threerings/presents/net/PongResponse.as

@@ -31,11 +31,6 @@ import com.threerings.io.Streamable;
 
 public class PongResponse extends DownstreamMessage
 {
-    public function PongResponse ()
-    {
-        super();
-    }
-
     public function getPackStamp () :Long
     {
         return _packStamp;

src/main/java/com/threerings/presents/client/ClientDObjectMgr.java

@@ -47,6 +47,8 @@ import com.threerings.presents.dobj.ObjectDestroyedEvent;
 import com.threerings.presents.dobj.Subscriber;
 import com.threerings.presents.net.BootstrapData;
 import com.threerings.presents.net.BootstrapNotification;
+import com.threerings.presents.net.CompoundDownstreamMessage;
+import com.threerings.presents.net.DownstreamMessage;
 import com.threerings.presents.net.EventNotification;
 import com.threerings.presents.net.FailureResponse;
 import com.threerings.presents.net.ForwardEventRequest;
@@ -184,44 +186,52 @@ public class ClientDObjectMgr
         // process the next event on our queue
         Object obj;
         if ((obj = _actions.getNonBlocking()) != null) {
-            // do the proper thing depending on the object
+            dispatchAction(obj);
-            if (obj instanceof EventNotification) {
+        }
-                dispatchEvent(((EventNotification)obj).getEvent());
+    }
 
-            } else if (obj instanceof BootstrapNotification) {
+    protected void dispatchAction (Object obj)
-                BootstrapData data = ((BootstrapNotification)obj).getData();
+    {
-                _client.gotBootstrap(data, this);
+        if (obj instanceof EventNotification) {
+            dispatchEvent(((EventNotification)obj).getEvent());
 
-            } else if (obj instanceof ObjectResponse<?>) {
+        } else if (obj instanceof BootstrapNotification) {
-                registerObjectAndNotify((ObjectResponse<?>)obj);
+            BootstrapData data = ((BootstrapNotification)obj).getData();
+            _client.gotBootstrap(data, this);
 
-            } else if (obj instanceof UnsubscribeResponse) {
+        } else if (obj instanceof ObjectResponse<?>) {
-                int oid = ((UnsubscribeResponse)obj).getOid();
+            registerObjectAndNotify((ObjectResponse<?>)obj);
-                if (_dead.remove(oid) == null) {
-                    log.warning("Received unsub ACK from unknown object", "oid", oid);
-                }
 
-            } else if (obj instanceof FailureResponse) {
+        } else if (obj instanceof UnsubscribeResponse) {
-                notifyFailure(((FailureResponse)obj).getOid(), ((FailureResponse)obj).getMessage());
+            int oid = ((UnsubscribeResponse)obj).getOid();
-
+            if (_dead.remove(oid) == null) {
-            } else if (obj instanceof PongResponse) {
+                log.warning("Received unsub ACK from unknown object", "oid", oid);
-                _client.gotPong((PongResponse)obj);
-
-            } else if (obj instanceof UpdateThrottleMessage) {
-                UpdateThrottleMessage upmsg = (UpdateThrottleMessage)obj;
-                _client.setOutgoingMessageThrottle(upmsg.messagesPerSec);
-
-            } else if (obj instanceof ObjectAction<?>) {
-                ObjectAction<?> act = (ObjectAction<?>)obj;
-                if (act.subscribe) {
-                    doSubscribe(act);
-                } else {
-                    doUnsubscribe(act.oid, act.target);
-                }
-
-            } else {
-                log.warning("Unknown action", "action", obj);
             }
+
+        } else if (obj instanceof FailureResponse) {
+            notifyFailure(((FailureResponse)obj).getOid(), ((FailureResponse)obj).getMessage());
+
+        } else if (obj instanceof PongResponse) {
+            _client.gotPong((PongResponse)obj);
+
+        } else if (obj instanceof UpdateThrottleMessage) {
+            UpdateThrottleMessage upmsg = (UpdateThrottleMessage)obj;
+            _client.setOutgoingMessageThrottle(upmsg.messagesPerSec);
+
+        } else if (obj instanceof ObjectAction<?>) {
+            ObjectAction<?> act = (ObjectAction<?>)obj;
+            if (act.subscribe) {
+                doSubscribe(act);
+            } else {
+                doUnsubscribe(act.oid, act.target);
+            }
+
+        } else if (obj instanceof CompoundDownstreamMessage) {
+            for (DownstreamMessage submsg : ((CompoundDownstreamMessage)obj).msgs) {
+                dispatchAction(submsg);
+            }
+        } else {
+            log.warning("Unknown action", "action", obj);
         }
     }
 

src/main/java/com/threerings/presents/net/CompoundDownstreamMessage.java

@@ -0,0 +1,40 @@
+//
+// $Id$
+//
+// Narya library - tools for developing networked games
+// Copyright (C) 2002-2011 Three Rings Design, Inc., All Rights Reserved
+// http://code.google.com/p/narya/
+//
+// This library is free software; you can redistribute it and/or modify it
+// under the terms of the GNU Lesser General Public License as published
+// by the Free Software Foundation; either version 2.1 of the License, or
+// (at your option) any later version.
+//
+// This library 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
+// Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public
+// License along with this library; if not, write to the Free Software
+// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+package com.threerings.presents.net;
+
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Groups messages to be dispatched without triggering the message throttle.
+ */
+public class CompoundDownstreamMessage extends DownstreamMessage
+{
+    public List<DownstreamMessage> msgs = Lists.newArrayList();
+
+    @Override
+    public String toString ()
+    {
+        return "[type=COMPOUND, msgid=" + messageId + "]";
+    }
+}

src/main/java/com/threerings/presents/net/CompoundUpstreamMessage.java

@@ -0,0 +1,42 @@
+//
+// $Id$
+//
+// Narya library - tools for developing networked games
+// Copyright (C) 2002-2011 Three Rings Design, Inc., All Rights Reserved
+// http://code.google.com/p/narya/
+//
+// This library is free software; you can redistribute it and/or modify it
+// under the terms of the GNU Lesser General Public License as published
+// by the Free Software Foundation; either version 2.1 of the License, or
+// (at your option) any later version.
+//
+// This library 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
+// Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public
+// License along with this library; if not, write to the Free Software
+// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+package com.threerings.presents.net;
+
+import java.util.List;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Groups messages to be dispatched without triggering the message throttle. Any messages sent to
+ * a client while processing an upstream message will be sent back en masse in a
+ * CompoundDownstreamMessage.
+ */
+public class CompoundUpstreamMessage extends UpstreamMessage
+{
+    public List<UpstreamMessage> msgs = Lists.newArrayList();
+
+    @Override
+    public String toString ()
+    {
+        return "[type=COMPOUND, msgid=" + messageId + "]";
+    }
+}

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

@@ -54,6 +54,8 @@ import com.threerings.presents.dobj.ProxySubscriber;
 import com.threerings.presents.net.AuthRequest;
 import com.threerings.presents.net.BootstrapData;
 import com.threerings.presents.net.BootstrapNotification;
+import com.threerings.presents.net.CompoundDownstreamMessage;
+import com.threerings.presents.net.CompoundUpstreamMessage;
 import com.threerings.presents.net.Credentials;
 import com.threerings.presents.net.DownstreamMessage;
 import com.threerings.presents.net.EventNotification;
@@ -70,6 +72,7 @@ import com.threerings.presents.net.TransmitDatagramsRequest;
 import com.threerings.presents.net.UnsubscribeRequest;
 import com.threerings.presents.net.UnsubscribeResponse;
 import com.threerings.presents.net.UpdateThrottleMessage;
+import com.threerings.presents.net.UpstreamMessage;
 import com.threerings.presents.server.net.PresentsConnection;
 
 import com.threerings.nio.conman.Connection;
@@ -433,8 +436,6 @@ public class PresentsSession
     // from interface Connection.MessageHandler
     public void handleMessage (Message message)
     {
-        _messagesIn++; // count 'em up!
-
         // if the client has been getting crazy with the cheeze whiz, stick a fork in them; the
         // first time through we end our session, subsequently _throttle is null and we just drop
         // any messages that come in until we've fully shutdown
@@ -447,6 +448,16 @@ public class PresentsSession
             handleThrottleExceeded();
         }
 
+        dispatchMessage(message);
+    }
+
+    /**
+     * Processes a message without throttling.
+     */
+    protected void dispatchMessage (Message message)
+    {
+        _messagesIn++; // count 'em up!
+
         // we dispatch to a message dispatcher that is specialized for the particular class of
         // message that we received
         MessageDispatcher disp = _disps.get(message.getClass());
@@ -464,8 +475,7 @@ public class PresentsSession
      * applied to this client. The old client object will not yet have been destroyed, so any final
      * events can be sent along prior to the new object being put into effect.
      */
-    protected void clientObjectWillChange (
+    protected void clientObjectWillChange (ClientObject oldClobj, ClientObject newClobj)
-        ClientObject oldClobj, ClientObject newClobj)
     {
     }
 
@@ -901,6 +911,31 @@ public class PresentsSession
         });
     }
 
+    /**
+     * Collects downstream messages in a compound message until finishCompoundMessage is called.
+     */
+    protected void startCompoundMessage ()
+    {
+        if (_compound == null) {
+            _compound = new CompoundDownstreamMessage();
+        }
+        _compoundDepth++;
+    }
+
+    /**
+     * Sends the compound message created in startCompoundMessage.
+     */
+    protected void finishCompoundMessage ()
+    {
+        if (--_compoundDepth == 0) {
+            CompoundDownstreamMessage downstream = _compound;
+            _compound = null;
+            if (!downstream.msgs.isEmpty()) {
+                postMessage(downstream, null);
+            }
+        }
+    }
+
     /** Queues a message for delivery to the client. */
     protected boolean postMessage (DownstreamMessage msg, PresentsConnection expect)
     {
@@ -914,6 +949,11 @@ public class PresentsSession
             return false;
         }
 
+        if (_compound != null) {
+            _compound.msgs.add(msg);
+            return true;
+        }
+
         // make sure we have a connection at all
         if (conn != null) {
             conn.postMessage(msg);
@@ -1124,6 +1164,29 @@ public class PresentsSession
         }
     }
 
+    /**
+     * Processes compound messages.
+     */
+    protected static class CompoundDispatcher implements MessageDispatcher
+    {
+        public void dispatch (final PresentsSession client, Message msg)
+        {
+            // Compound downstream messages sent while dispatching the upstream compound message
+            client._omgr.postRunnable(new Runnable() {
+                public void run () {
+                    client.startCompoundMessage();
+                }});
+            for (UpstreamMessage submsg : ((CompoundUpstreamMessage)msg).msgs) {
+                client.dispatchMessage(submsg);
+            }
+            // Send any messages produced en masse now that we've finished dispatching
+            client._omgr.postRunnable(new Runnable() {
+                public void run () {
+                    client.finishCompoundMessage();
+                }});
+        }
+    }
+
     /**
      * Processes unsubscribe requests.
      */
@@ -1236,6 +1299,18 @@ public class PresentsSession
     protected ClientObject _clobj;
     protected IntMap<ClientProxy> _subscrips = IntMaps.newHashIntMap();
 
+    /**
+     * Message in which we're currently compounding messages to send, or null if we're sending them
+     * straight on.
+     */
+    protected CompoundDownstreamMessage _compound;
+
+    /**
+     * The count of startCompoundMessage calls that have occurred without a finishCompoundMessage.
+     * _compound won't be set until this reaches 0 again.
+     */
+    protected int _compoundDepth;
+
     /** The Oids of objects that have been destroyed while we were subscribed. */
     protected HashSet<Integer> _destroyedSubs = Sets.newHashSet();
     protected ClassLoader _loader;
@@ -1282,5 +1357,6 @@ public class PresentsSession
         _disps.put(TransmitDatagramsRequest.class, new TransmitDatagramsDispatcher());
         _disps.put(ThrottleUpdatedMessage.class, new ThrottleUpdatedDispatcher());
         _disps.put(LogoffRequest.class, new LogoffDispatcher());
+        _disps.put(CompoundUpstreamMessage.class, new CompoundDispatcher());
     }
 }