Finally got around to making compound events actually stick together until

they arrive at the client. Mmm... network efficiency++.


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@2396 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2003-04-10 17:48:42 +00:00
parent 2a554f1569
commit 36624d96ac
6 changed files with 177 additions and 144 deletions
@@ -1,5 +1,5 @@
// //
// $Id: ClientDObjectMgr.java,v 1.22 2003/03/11 04:43:14 mdb Exp $ // $Id: ClientDObjectMgr.java,v 1.23 2003/04/10 17:48:42 mdb Exp $
package com.threerings.presents.client; package com.threerings.presents.client;
@@ -166,6 +166,7 @@ public class ClientDObjectMgr
} else if (obj instanceof EventNotification) { } else if (obj instanceof EventNotification) {
DEvent evt = ((EventNotification)obj).getEvent(); DEvent evt = ((EventNotification)obj).getEvent();
// Log.info("Dispatch event: " + evt);
dispatchEvent(evt); dispatchEvent(evt);
} else if (obj instanceof ObjectResponse) { } else if (obj instanceof ObjectResponse) {
@@ -202,8 +203,6 @@ public class ClientDObjectMgr
*/ */
protected void dispatchEvent (DEvent event) protected void dispatchEvent (DEvent event)
{ {
// Log.info("Dispatch event: " + event);
// if this is a compound event, we need to process its contained // if this is a compound event, we need to process its contained
// events in order // events in order
if (event instanceof CompoundEvent) { if (event instanceof CompoundEvent) {
@@ -1,5 +1,5 @@
// //
// $Id: CompoundEvent.java,v 1.8 2002/12/20 23:41:26 mdb Exp $ // $Id: CompoundEvent.java,v 1.9 2003/04/10 17:48:42 mdb Exp $
package com.threerings.presents.dobj; package com.threerings.presents.dobj;
@@ -27,9 +27,9 @@ public class CompoundEvent extends DEvent
/** /**
* Constructs a compound event and prepares it for operation. * Constructs a compound event and prepares it for operation.
*/ */
public CompoundEvent (DObjectManager omgr) public CompoundEvent (DObject target, DObjectManager omgr)
{ {
super(0); // we don't have a single target object oid super(target.getOid());
// sanity check // sanity check
if (omgr == null) { if (omgr == null) {
@@ -38,22 +38,10 @@ public class CompoundEvent extends DEvent
} }
_omgr = omgr; _omgr = omgr;
_target = target;
_events = new StreamableArrayList(); _events = new StreamableArrayList();
} }
/**
* Lets the event know that this dobject is participating in their
* transaction. The supplied dobject will have their transaction
* cleared when this event is committed or cancelled.
*/
public void addObject (DObject object)
{
if (_participants == null) {
_participants = new ArrayList();
}
_participants.add(object);
}
/** /**
* Posts an event to this transaction. The event will be delivered as * Posts an event to this transaction. The event will be delivered as
* part of the entire transaction if it is committed or discarded if * part of the entire transaction if it is committed or discarded if
@@ -66,6 +54,7 @@ public class CompoundEvent extends DEvent
/** /**
* Returns the list of events contained within this compound event. * Returns the list of events contained within this compound event.
* Don't mess with it.
*/ */
public List getEvents () public List getEvents ()
{ {
@@ -80,13 +69,20 @@ public class CompoundEvent extends DEvent
*/ */
public void commit () public void commit ()
{ {
// first clear our participants // first clear our target
clearParticipants(); clearTarget();
// then post this event onto the queue (but only if we actually // then post this event onto the queue (but only if we actually
// accumulated some events) // accumulated some events)
if (_events.size() > 0) { switch (_events.size()) {
case 0: // nothing doing
break;
case 1: // no point in being compound
_omgr.postEvent((DEvent)_events.get(0));
break;
default: // now we're talking
_omgr.postEvent(this); _omgr.postEvent(this);
break;
} }
} }
@@ -96,8 +92,8 @@ public class CompoundEvent extends DEvent
*/ */
public void cancel () public void cancel ()
{ {
// clear our participants // clear our target
clearParticipants(); clearTarget();
// clear our event queue in case someone holds onto us // clear our event queue in case someone holds onto us
_events.clear(); _events.clear();
} }
@@ -125,18 +121,13 @@ public class CompoundEvent extends DEvent
} }
/** /**
* Calls out to all of the participating dobjects, clearing their * Calls out to our target object, clearing its transaction reference.
* transaction reference.
*/ */
protected void clearParticipants () protected void clearTarget ()
{ {
if (_participants != null) { if (_target != null) {
int psize = _participants.size(); _target.clearTransaction();
for (int i = 0; i < psize; i++) { _target = null;
DObject obj = (DObject)_participants.get(i);
obj.clearTransaction();
}
_participants = null;
} }
} }
@@ -154,10 +145,8 @@ public class CompoundEvent extends DEvent
* committed. */ * committed. */
protected transient DObjectManager _omgr; protected transient DObjectManager _omgr;
/** A list of the dobject participants in this transaction. They will /** The object for which we're managing a transaction. */
* be notified when we are committed or cancelled so that they can protected transient DObject _target;
* stop posting their events to us. */
protected transient ArrayList _participants;
/** A list of the events associated with this compound event. */ /** A list of the events associated with this compound event. */
protected StreamableArrayList _events; protected StreamableArrayList _events;
@@ -1,5 +1,5 @@
// //
// $Id: DObject.java,v 1.61 2003/03/30 19:38:56 mdb Exp $ // $Id: DObject.java,v 1.62 2003/04/10 17:48:42 mdb Exp $
package com.threerings.presents.dobj; package com.threerings.presents.dobj;
@@ -16,18 +16,26 @@ import com.threerings.presents.Log;
/** /**
* The distributed object forms the foundation of the Presents system. All * The distributed object forms the foundation of the Presents system. All
* information shared among users of the system is done via distributed * information shared among users of the system is done via distributed
* objects. A distributed object has a set of subscribers. These * objects. A distributed object has a set of listeners. These listeners
* subscribers have access to the object or a proxy of the object and * have access to the object or a proxy of the object and therefore have
* therefore have access to the data stored in the object's members at all * access to the data stored in the object's members at all times.
* times.
* *
* <p> When there is any change to that data, initiated by one of the * <p> Additionally, an object as a set of subscribers. Subscribers manage
* subscribers, an event is generated which is dispatched to all * the lifespan of the object; while a subscriber is subscribed, the
* subscribers of the object, notifying them of that change and affecting * listeners registered with an object will be notified of events. When
* that change to the copy of the object maintained at each client. In * the subscriber unsubscribes, the object becomes non-live and the
* this way, both a respository of shared information and a mechanism for * listeners are no longer notified. <em>Note:</em> on the server, object
* asynchronous notification are made available as a fundamental * subscription is merely a formality as all objects remain live all the
* application building blocks. * time, so <em>do not</em> depend on event notifications ceasing when a
* subscriber has relinquished its subscription. Always unregister all
* listeners when they no longer need to hear from an object.
*
* <p> When there is any change to the the object's fields data, an event
* is generated which is dispatched to all listeners of the object,
* notifying them of that change and effecting that change to the copy of
* the object maintained at each client. In this way, both a respository
* of shared information and a mechanism for asynchronous notification are
* made available as a fundamental application building blocks.
* *
* <p> To define what information is shared, an application creates a * <p> To define what information is shared, an application creates a
* distributed object declaration which is much like a class declaration * distributed object declaration which is much like a class declaration
@@ -78,14 +86,16 @@ import com.threerings.presents.Log;
* These method calls on the actual distributed object will result in the * These method calls on the actual distributed object will result in the
* proper attribute change events being generated and dispatched. * proper attribute change events being generated and dispatched.
* *
* <p> Note that distributed object fields can only be of a limited set of * <p> Note that distributed object fields can be any of the following set
* supported types. These types are: * of primitive types:
* *
* <code><pre> * <code><pre>
* byte, short, int, long, float, double * boolean, byte, short, int, long, float, double
* Byte, Short, Integer, Long, Float, Double, String * Boolean, Byte, Short, Integer, Long, Float, Double, String
* byte[], short[], int[], long[], float[], double[], String[] * boolean[], byte[], short[], int[], long[], float[], double[], String[]
* </pre></code> * </pre></code>
*
* Fields of type {@link Streamable} can also be used.
*/ */
public class DObject implements Streamable public class DObject implements Streamable
{ {
@@ -173,11 +183,9 @@ public class DObject implements Streamable
Object[] els = ListUtil.testAndAdd(_listeners, listener); Object[] els = ListUtil.testAndAdd(_listeners, listener);
if (els != null) { if (els != null) {
_listeners = els; _listeners = els;
} else { } else {
// complain if an object requests to add itself more than once Log.warning("Refusing repeat listener registration " +
Log.warning("Refusing listener that's already in the list " + "[dobj=" + which() + ", list=" + listener + "]");
"[dobj=" + which() + ", listener=" + listener + "]");
Thread.dumpStack(); Thread.dumpStack();
} }
} }
@@ -341,24 +349,19 @@ public class DObject implements Streamable
/** /**
* Called by the distributed object manager after it has applied an * Called by the distributed object manager after it has applied an
* event to this object. This dispatches an event notification to all * event to this object. This dispatches an event notification to all
* of the subscribers of this object. * of the listeners registered with this object.
* *
* @param event the event that was just applied. * @param event the event that was just applied.
*/ */
public void notifyListeners (DEvent event) public void notifyListeners (DEvent event)
{ {
// if we have no listeners, we're home free
if (_listeners == null) { if (_listeners == null) {
return; return;
} }
// iterate over the listener list, performing the necessary
// notifications
int llength = _listeners.length; int llength = _listeners.length;
for (int i = 0; i < llength; i++) { for (int i = 0; i < llength; i++) {
Object listener = _listeners[i]; Object listener = _listeners[i];
// skip empty slots
if (listener == null) { if (listener == null) {
continue; continue;
} }
@@ -374,8 +377,34 @@ public class DObject implements Streamable
} catch (Exception e) { } catch (Exception e) {
Log.warning("Listener choked during notification " + Log.warning("Listener choked during notification " +
"[listener=" + listener + "[list=" + listener + ", event=" + event + "].");
", event=" + event + "]."); Log.logStackTrace(e);
}
}
}
/**
* Called by the distributed object manager after it has applied an
* event to this object. This dispatches an event notification to all
* of the proxy listeners registered with this object.
*
* @param event the event that was just applied.
*/
public void notifyProxies (DEvent event)
{
if (_subs == null) {
return;
}
for (int ii = 0, ll = _subs.length; ii < ll; ii++) {
Object sub = _subs[ii];
try {
if (sub != null && sub instanceof ProxySubscriber) {
((ProxySubscriber)sub).eventReceived(event);
}
} catch (Exception e) {
Log.warning("Proxy choked during notification " +
"[sub=" + sub + ", event=" + event + "].");
Log.logStackTrace(e); Log.logStackTrace(e);
} }
} }
@@ -565,19 +594,18 @@ public class DObject implements Streamable
* *
* <p> When the transaction is complete, the caller must call {@link * <p> When the transaction is complete, the caller must call {@link
* #commitTransaction} or {@link CompoundEvent#commit} to commit the * #commitTransaction} or {@link CompoundEvent#commit} to commit the
* transaction and release all involved objects back to their normal * transaction and release the object back to its normal
* non-transacting state. If the caller decides not to commit their * non-transacting state. If the caller decides not to commit their
* transaction, they must call {@link #cancelTransaction} or {@link * transaction, they must call {@link #cancelTransaction} or {@link
* CompoundEvent#cancel} to cancel the transaction and release all * CompoundEvent#cancel} to cancel the transaction. Failure to do so
* involved objects. Failure to do so will cause the pooch to be * will cause the pooch to be totally screwed.
* totally screwed.
* *
* <p> Note: like all other distributed object operations, * <p> Note: like all other distributed object operations,
* transactions are not thread safe. It is expected that a single * transactions are not thread safe. It is expected that a single
* thread will handle all distributed object operations and that * thread will handle all distributed object operations and that
* thread will begin and complete a transaction before giving up * thread will begin and complete a transaction before giving up
* control to unknown code which might try to operate on the * control to unknown code which might try to operate on the
* transacting distributed object (or objects). * transacting distributed object.
* *
* <p> Note also: if the object is already engaged in a transaction, a * <p> Note also: if the object is already engaged in a transaction, a
* transaction participant count will be incremented to note that an * transaction participant count will be incremented to note that an
@@ -589,19 +617,8 @@ public class DObject implements Streamable
* cancels the transaction, the entire transaction is cancelled for * cancels the transaction, the entire transaction is cancelled for
* all participants, regardless of whether the other participants * all participants, regardless of whether the other participants
* attempted to commit the transaction. * attempted to commit the transaction.
*
* <p> Note also: nested transactions are not allowed when an object
* is joined to another object's transaction via {@link
* #joinTransaction}.
*
* @return the compound event that encapsulates the transaction. This
* can be ignored if the transaction will be limited to this object,
* but if it is desired that other objects be involved in the
* transaction, the caller will need to pass the {@link CompoundEvent}
* returned by this method to a call to {@link #joinTransaction} on
* the other objects that are involved.
*/ */
public CompoundEvent startTransaction () public void startTransaction ()
{ {
// sanity check // sanity check
if (!isActive()) { if (!isActive()) {
@@ -611,40 +628,10 @@ public class DObject implements Streamable
} }
if (_tevent != null) { if (_tevent != null) {
// a transaction count of -1 indicates that we're joined to _tcount++;
// another objects transaction that should not allow
// transaction nesting
if (_tcount == -1) {
String errmsg = "Object involved in shared transaction, " +
"nesting not allowed [dobj=" + this + "]";
throw new IllegalStateException(errmsg);
} else {
_tcount++;
}
} else { } else {
_tevent = new CompoundEvent(_omgr); _tevent = new CompoundEvent(this, _omgr);
_tevent.addObject(this);
} }
return _tevent;
}
/**
* Causes this object to join the supplied transaction. See {@link
* #startTransaction} for more information on transactions. The
* transaction can be committed by a call to {@link
* #commitTransaction} on any participanting object.
*/
public void joinTransaction (CompoundEvent event)
{
if (_tevent != null) {
String errmsg = "Cannot join transaction while already " +
"transacting [dobj=" + this + "]";
throw new IllegalStateException(errmsg);
}
_tevent = event;
_tevent.addObject(this);
_tcount = -1; // make a note that nesting is not allowed
} }
/** /**
@@ -0,0 +1,22 @@
//
// $Id: ProxySubscriber.java,v 1.1 2003/04/10 17:48:42 mdb Exp $
package com.threerings.presents.dobj;
/**
* Defines a special kind of subscriber that proxies events for a
* subordinate distributed object manager. All events dispatched on
* objects with which this subscriber is registered are passed along to
* the subscriber for delivery to its subordinate manager.
*
* @see DObject#addListener
*/
public interface ProxySubscriber extends Subscriber
{
/**
* Called when any event has been dispatched on an object.
*
* @param event The event that was dispatched on the object.
*/
public void eventReceived (DEvent event);
}
@@ -1,5 +1,5 @@
// //
// $Id: PresentsClient.java,v 1.53 2003/03/30 21:04:18 mdb Exp $ // $Id: PresentsClient.java,v 1.54 2003/04/10 17:48:42 mdb Exp $
package com.threerings.presents.server; package com.threerings.presents.server;
@@ -15,9 +15,8 @@ import com.threerings.presents.data.ClientObject;
import com.threerings.presents.dobj.DEvent; import com.threerings.presents.dobj.DEvent;
import com.threerings.presents.dobj.DObject; import com.threerings.presents.dobj.DObject;
import com.threerings.presents.dobj.EventListener;
import com.threerings.presents.dobj.ObjectAccessException; import com.threerings.presents.dobj.ObjectAccessException;
import com.threerings.presents.dobj.Subscriber; import com.threerings.presents.dobj.ProxySubscriber;
import com.threerings.presents.net.BootstrapData; import com.threerings.presents.net.BootstrapData;
import com.threerings.presents.net.BootstrapNotification; import com.threerings.presents.net.BootstrapNotification;
@@ -54,8 +53,7 @@ import com.threerings.presents.server.net.MessageHandler;
* conmgr thread and therefore also need not be synchronized. * conmgr thread and therefore also need not be synchronized.
*/ */
public class PresentsClient public class PresentsClient
implements Subscriber, EventListener, MessageHandler, implements ProxySubscriber, MessageHandler, ClientResolutionListener
ClientResolutionListener
{ {
/** Used by {@link #setUsername} to report success or failure. */ /** Used by {@link #setUsername} to report success or failure. */
public static interface UserChangeListener public static interface UserChangeListener
@@ -416,7 +414,6 @@ public class PresentsClient
{ {
DObject object = (DObject)_subscrips.remove(oid); DObject object = (DObject)_subscrips.remove(oid);
if (object != null) { if (object != null) {
object.removeListener(this);
object.removeSubscriber(this); object.removeSubscriber(this);
} else { } else {
Log.warning("Requested to unmap non-existent subscription " + Log.warning("Requested to unmap non-existent subscription " +
@@ -435,7 +432,6 @@ public class PresentsClient
DObject object = (DObject)enum.next(); DObject object = (DObject)enum.next();
// Log.info("Clearing subscription [client=" + this + // Log.info("Clearing subscription [client=" + this +
// ", obj=" + object.getOid() + "]."); // ", obj=" + object.getOid() + "].");
object.removeListener(this);
object.removeSubscriber(this); object.removeSubscriber(this);
} }
} }
@@ -649,8 +645,6 @@ public class PresentsClient
// queue up an object response // queue up an object response
Connection conn = getConnection(); Connection conn = getConnection();
if (conn != null) { if (conn != null) {
// add ourselves as an event listener
object.addListener(this);
// pass the successful subscrip on to the client // pass the successful subscrip on to the client
conn.postMessage(new ObjectResponse(object)); conn.postMessage(new ObjectResponse(object));
// make a note of this new subscription // make a note of this new subscription
@@ -1,5 +1,5 @@
// //
// $Id: PresentsDObjectMgr.java,v 1.30 2003/03/31 04:11:24 mdb Exp $ // $Id: PresentsDObjectMgr.java,v 1.31 2003/04/10 17:48:42 mdb Exp $
package com.threerings.presents.server; package com.threerings.presents.server;
@@ -23,7 +23,7 @@ import com.threerings.presents.dobj.*;
* server. By virtue of running on the server, it manages its objects * server. By virtue of running on the server, it manages its objects
* directly rather than managing proxies of objects which is what is done * directly rather than managing proxies of objects which is what is done
* on the client. Thus it simply queues up events and dispatches them to * on the client. Thus it simply queues up events and dispatches them to
* subscribers. * listeners.
* *
* <p> The server object manager is meant to run on the main thread of the * <p> The server object manager is meant to run on the main thread of the
* server application and thus provides a method to be invoked by the * server application and thus provides a method to be invoked by the
@@ -180,18 +180,9 @@ public class PresentsDObjectMgr
} }
} else if (unit instanceof CompoundEvent) { } else if (unit instanceof CompoundEvent) {
// if this is a compound event, we need to apply each processCompoundEvent((CompoundEvent)unit);
// event indivdually
CompoundEvent event = (CompoundEvent)unit;
List events = event.getEvents();
int ecount = events.size();
for (int i = 0; i < ecount; i++) {
processEvent((DEvent)events.get(i));
}
} else { } else {
// otherwise it's a regular event, so do the standard
// processing
processEvent((DEvent)unit); processEvent((DEvent)unit);
} }
@@ -217,9 +208,45 @@ public class PresentsDObjectMgr
Log.info("DOMGR exited."); Log.info("DOMGR exited.");
} }
/**
* Performs the processing associated with a compound event, notifying
* listeners and the like.
*/
protected void processCompoundEvent (CompoundEvent event)
{
List events = event.getEvents();
int ecount = events.size();
// look up the target object
DObject target = (DObject)_objects.get(event.getTargetOid());
if (target == null) {
Log.debug("Compound event target no longer exists " +
"[event=" + event + "].");
return;
}
// check the permissions on all of the events
for (int ii = 0; ii < ecount; ii++) {
DEvent sevent = (DEvent)events.get(ii);
if (!target.checkPermissions(sevent)) {
Log.warning("Event failed permissions check " +
"[event=" + sevent + ", target=" + target + "].");
return;
}
}
// dispatch the events
for (int ii = 0; ii < ecount; ii++) {
dispatchEvent((DEvent)events.get(ii), target);
}
// always notify proxies of compound events
target.notifyProxies(event);
}
/** /**
* Performs the processing associated with an event, notifying * Performs the processing associated with an event, notifying
* subscribers and the like. * listeners and the like.
*/ */
protected void processEvent (DEvent event) protected void processEvent (DEvent event)
{ {
@@ -244,22 +271,36 @@ public class PresentsDObjectMgr
return; return;
} }
if (dispatchEvent(event, target)) {
// unless requested not to, notify any proxies
target.notifyProxies(event);
}
}
/**
* Dispatches an event after the target object has been resolved and
* the permissions have been checked. This is used by {@link
* #processEvent} and {@link #processCompoundEvent}.
*
* @return the value returned by {@link DEvent#applyToObject}.
*/
protected boolean dispatchEvent (DEvent event, DObject target)
{
boolean notify = true; // assume always notify
try { try {
// do any internal management necessary based on this // do any internal management necessary based on this event
// event
Method helper = (Method)_helpers.get(event.getClass()); Method helper = (Method)_helpers.get(event.getClass());
if (helper != null) { if (helper != null) {
// invoke the helper method // invoke the helper method
Object rv = Object rv = helper.invoke(this, new Object[] { event, target });
helper.invoke(this, new Object[] { event, target });
// if helper returns false, we abort event processing // if helper returns false, we abort event processing
if (!((Boolean)rv).booleanValue()) { if (!((Boolean)rv).booleanValue()) {
return; return false;
} }
} }
// everything's good so far, apply the event to the object // everything's good so far, apply the event to the object
boolean notify = event.applyToObject(target); notify = event.applyToObject(target);
// if the event returns false from applyToObject, this // if the event returns false from applyToObject, this
// means it's a silent event and we shouldn't notify the // means it's a silent event and we shouldn't notify the
@@ -276,6 +317,7 @@ public class PresentsDObjectMgr
// track the number of events dispatched // track the number of events dispatched
++_eventCount; ++_eventCount;
return true;
} }
/** /**