diff --git a/src/main/as/com/threerings/presents/client/Client.as b/src/main/as/com/threerings/presents/client/Client.as index 18e48fd0b..9e342bb54 100644 --- a/src/main/as/com/threerings/presents/client/Client.as +++ b/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. diff --git a/src/main/as/com/threerings/presents/client/ClientDObjectMgr.as b/src/main/as/com/threerings/presents/client/ClientDObjectMgr.as index 674889911..5d863e63d 100644 --- a/src/main/as/com/threerings/presents/client/ClientDObjectMgr.as +++ b/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); + } } } diff --git a/src/main/as/com/threerings/presents/client/Communicator.as b/src/main/as/com/threerings/presents/client/Communicator.as index e0636339b..88b968b55 100644 --- a/src/main/as/com/threerings/presents/client/Communicator.as +++ b/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. */ diff --git a/src/main/as/com/threerings/presents/net/CompoundDownstreamMessage.as b/src/main/as/com/threerings/presents/net/CompoundDownstreamMessage.as new file mode 100644 index 000000000..10695fae2 --- /dev/null +++ b/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); + } +} +} diff --git a/src/main/as/com/threerings/presents/net/CompoundUpstreamMessage.as b/src/main/as/com/threerings/presents/net/CompoundUpstreamMessage.as new file mode 100644 index 000000000..e083d33bb --- /dev/null +++ b/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); + } +} +} diff --git a/src/main/as/com/threerings/presents/net/PongResponse.as b/src/main/as/com/threerings/presents/net/PongResponse.as index f6e760cd5..5d7f14af1 100644 --- a/src/main/as/com/threerings/presents/net/PongResponse.as +++ b/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; diff --git a/src/main/java/com/threerings/presents/client/ClientDObjectMgr.java b/src/main/java/com/threerings/presents/client/ClientDObjectMgr.java index f342f2cf6..1aa9a5e5f 100644 --- a/src/main/java/com/threerings/presents/client/ClientDObjectMgr.java +++ b/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 - if (obj instanceof EventNotification) { - dispatchEvent(((EventNotification)obj).getEvent()); + dispatchAction(obj); + } + } - } else if (obj instanceof BootstrapNotification) { - BootstrapData data = ((BootstrapNotification)obj).getData(); - _client.gotBootstrap(data, this); + protected void dispatchAction (Object obj) + { + if (obj instanceof EventNotification) { + dispatchEvent(((EventNotification)obj).getEvent()); - } else if (obj instanceof ObjectResponse) { - registerObjectAndNotify((ObjectResponse)obj); + } else if (obj instanceof BootstrapNotification) { + BootstrapData data = ((BootstrapNotification)obj).getData(); + _client.gotBootstrap(data, this); - } else if (obj instanceof UnsubscribeResponse) { - int oid = ((UnsubscribeResponse)obj).getOid(); - if (_dead.remove(oid) == null) { - log.warning("Received unsub ACK from unknown object", "oid", oid); - } + } else if (obj instanceof ObjectResponse) { + registerObjectAndNotify((ObjectResponse)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 { - log.warning("Unknown action", "action", obj); + } else if (obj instanceof UnsubscribeResponse) { + int oid = ((UnsubscribeResponse)obj).getOid(); + if (_dead.remove(oid) == null) { + log.warning("Received unsub ACK from unknown object", "oid", oid); } + + } 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); } } diff --git a/src/main/java/com/threerings/presents/net/CompoundDownstreamMessage.java b/src/main/java/com/threerings/presents/net/CompoundDownstreamMessage.java new file mode 100644 index 000000000..036d3f6d6 --- /dev/null +++ b/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 msgs = Lists.newArrayList(); + + @Override + public String toString () + { + return "[type=COMPOUND, msgid=" + messageId + "]"; + } +} diff --git a/src/main/java/com/threerings/presents/net/CompoundUpstreamMessage.java b/src/main/java/com/threerings/presents/net/CompoundUpstreamMessage.java new file mode 100644 index 000000000..81bf3dc08 --- /dev/null +++ b/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 msgs = Lists.newArrayList(); + + @Override + public String toString () + { + return "[type=COMPOUND, msgid=" + messageId + "]"; + } +} diff --git a/src/main/java/com/threerings/presents/server/PresentsSession.java b/src/main/java/com/threerings/presents/server/PresentsSession.java index af081a6a8..324b5f3b2 100644 --- a/src/main/java/com/threerings/presents/server/PresentsSession.java +++ b/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 ( - ClientObject oldClobj, ClientObject newClobj) + protected void clientObjectWillChange (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 _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 _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()); } }