diff --git a/src/java/com/threerings/presents/dobj/AttributeChangedEvent.java b/src/java/com/threerings/presents/dobj/AttributeChangedEvent.java
index d0b81a23b..2cd3524d7 100644
--- a/src/java/com/threerings/presents/dobj/AttributeChangedEvent.java
+++ b/src/java/com/threerings/presents/dobj/AttributeChangedEvent.java
@@ -28,17 +28,17 @@ import java.util.HashMap;
import com.samskivert.util.StringUtil;
/**
- * An attribute changed event is dispatched when a single attribute of a
- * distributed object has changed. It can also be constructed to request
- * an attribute change on an object and posted to the dobjmgr.
+ * An attribute changed event is dispatched when a single attribute of a distributed object has
+ * changed. It can also be constructed to request an attribute change on an object and posted to
+ * the dobjmgr.
*
* @see DObjectManager#postEvent
*/
public class AttributeChangedEvent extends NamedEvent
{
/**
- * Constructs a blank instance of this event in preparation for
- * unserialization from the network.
+ * Constructs a blank instance of this event in preparation for unserialization from the
+ * network.
*/
public AttributeChangedEvent ()
{
@@ -53,8 +53,7 @@ public class AttributeChangedEvent extends NamedEvent
}
/**
- * Returns the value of the attribute prior to the application of this
- * event.
+ * Returns the value of the attribute prior to the application of this event.
*/
public Object getOldValue ()
{
@@ -62,8 +61,8 @@ public class AttributeChangedEvent extends NamedEvent
}
/**
- * Returns the new value of the attribute as a byte. This will fail
- * if the attribute in question is not a byte.
+ * Returns the new value of the attribute as a byte. This will fail if the attribute in
+ * question is not a byte.
*/
public byte getByteValue ()
{
@@ -71,8 +70,8 @@ public class AttributeChangedEvent extends NamedEvent
}
/**
- * Returns the new value of the attribute as a short. This will fail
- * if the attribute in question is not a short.
+ * Returns the new value of the attribute as a short. This will fail if the attribute in
+ * question is not a short.
*/
public short getShortValue ()
{
@@ -80,8 +79,8 @@ public class AttributeChangedEvent extends NamedEvent
}
/**
- * Returns the new value of the attribute as an int. This will fail if
- * the attribute in question is not an int.
+ * Returns the new value of the attribute as an int. This will fail if the attribute in
+ * question is not an int.
*/
public int getIntValue ()
{
@@ -89,8 +88,8 @@ public class AttributeChangedEvent extends NamedEvent
}
/**
- * Returns the new value of the attribute as a long. This will fail if
- * the attribute in question is not a long.
+ * Returns the new value of the attribute as a long. This will fail if the attribute in
+ * question is not a long.
*/
public long getLongValue ()
{
@@ -98,8 +97,8 @@ public class AttributeChangedEvent extends NamedEvent
}
/**
- * Returns the new value of the attribute as a float. This will fail
- * if the attribute in question is not a float.
+ * Returns the new value of the attribute as a float. This will fail if the attribute in
+ * question is not a float.
*/
public float getFloatValue ()
{
@@ -107,37 +106,38 @@ public class AttributeChangedEvent extends NamedEvent
}
/**
- * Returns the new value of the attribute as a double. This will fail
- * if the attribute in question is not a double.
+ * Returns the new value of the attribute as a double. This will fail if the attribute in
+ * question is not a double.
*/
public double getDoubleValue ()
{
return ((Double)_value).doubleValue();
}
- /**
- * Applies this attribute change to the object.
- */
+ @Override // from DEvent
+ public boolean alreadyApplied ()
+ {
+ // if we have an old value, that means we're running on the master server and we have
+ // already applied this attribute change to the object
+ return (_oldValue != UNSET_OLD_VALUE);
+ }
+
+ @Override // from DEvent
public boolean applyToObject (DObject target)
throws ObjectAccessException
{
- // if we have no old value, that means we're not running on the
- // master server and we have not already applied this attribute
- // change to the object, so we must grab the previous value and
- // actually apply the attribute change
- if (_oldValue == UNSET_OLD_VALUE) {
+ // if we're not already applied, grab the previous value and apply the attribute change
+ if (!alreadyApplied()) {
_oldValue = target.getAttribute(_name);
Object value = _value;
if (value != null) {
Class vclass = value.getClass();
if (vclass.isPrimitive()) {
- // do nothing; we check this to avoid the more
- // expensive isAssignableFrom check on primitives
- // which are far and away the most common case
+ // do nothing; we check this to avoid the more expensive isAssignableFrom check
+ // on primitives which are far and away the most common case
} else if (vclass.isArray()) {
int length = Array.getLength(value);
- Object clone = Array.newInstance(
- vclass.getComponentType(), length);
+ Object clone = Array.newInstance(vclass.getComponentType(), length);
System.arraycopy(value, 0, clone, 0, length);
value = clone;
} else if (DSet.class.isAssignableFrom(vclass)) {
@@ -151,21 +151,16 @@ public class AttributeChangedEvent extends NamedEvent
}
/**
- * Constructs a new attribute changed event on the specified target
- * object with the supplied attribute name and value. Do not
- * construct these objects by hand. Use {@link
+ * Constructs a new attribute changed event on the specified target object with the supplied
+ * attribute name and value. Do not construct these objects by hand. Use {@link
* DObject#changeAttribute} instead.
*
- * @param targetOid the object id of the object whose attribute has
- * changed.
- * @param name the name of the attribute (data member) that has
- * changed.
- * @param value the new value of the attribute (in the case of
- * primitive types, the reflection-defined object-alternative is
- * used).
+ * @param targetOid the object id of the object whose attribute has changed.
+ * @param name the name of the attribute (data member) that has changed.
+ * @param value the new value of the attribute (in the case of primitive types, the
+ * reflection-defined object-alternative is used).
*/
- protected AttributeChangedEvent (
- int targetOid, String name, Object value, Object oldValue)
+ protected AttributeChangedEvent (int targetOid, String name, Object value, Object oldValue)
{
super(targetOid, name);
_value = value;
diff --git a/src/java/com/threerings/presents/dobj/DEvent.java b/src/java/com/threerings/presents/dobj/DEvent.java
index 3e8ec2544..f75b169e4 100644
--- a/src/java/com/threerings/presents/dobj/DEvent.java
+++ b/src/java/com/threerings/presents/dobj/DEvent.java
@@ -30,6 +30,10 @@ import com.threerings.io.Streamable;
*/
public abstract class DEvent implements Streamable
{
+ /** This event's "number". Every event dispatched by the server is numbered in monotonically
+ * increasing fashion when the event is posted to the event dispatch queue. */
+ public transient long eventId;
+
/**
* A zero argument constructor for unserialization from yon network.
*/
@@ -63,6 +67,16 @@ public abstract class DEvent implements Streamable
return false;
}
+ /**
+ * If this event applies itself immediately to the distributed object on the server and then
+ * NOOPs later when {@link #applyToObject} is called, it should return true from this method.
+ * If it will modify the object during its {@link #applyToObject} call, it should return false.
+ */
+ public boolean alreadyApplied ()
+ {
+ return false;
+ }
+
/**
* Applies the attribute modifications represented by this event to the specified target
* object. This is called by the distributed object manager in the course of dispatching events
diff --git a/src/java/com/threerings/presents/dobj/ElementUpdatedEvent.java b/src/java/com/threerings/presents/dobj/ElementUpdatedEvent.java
index cc6787bc4..3144115bd 100644
--- a/src/java/com/threerings/presents/dobj/ElementUpdatedEvent.java
+++ b/src/java/com/threerings/presents/dobj/ElementUpdatedEvent.java
@@ -27,41 +27,37 @@ import java.lang.reflect.Field;
import com.samskivert.util.StringUtil;
/**
- * An element updated event is dispatched when an element of an array
- * field in a distributed object is updated. It can also be constructed to
- * request the update of an entry and posted to the dobjmgr.
+ * An element updated event is dispatched when an element of an array field in a distributed object
+ * is updated. It can also be constructed to request the update of an entry and posted to the
+ * dobjmgr.
*
* @see DObjectManager#postEvent
*/
public class ElementUpdatedEvent extends NamedEvent
{
/**
- * Constructs a new element updated event on the specified target
- * object with the supplied attribute name, element and index.
+ * Constructs a new element updated event on the specified target object with the supplied
+ * attribute name, element and index.
*
- * @param targetOid the object id of the object whose attribute has
- * changed.
- * @param name the name of the attribute (data member) for which an
- * element has changed.
- * @param value the new value of the element (in the case of primitive
- * types, the reflection-defined object-alternative is used).
- * @param oldValue the previous value of the element (in the case of
- * primitive types, the reflection-defined object-alternative is
- * used).
+ * @param targetOid the object id of the object whose attribute has changed.
+ * @param name the name of the attribute (data member) for which an element has changed.
+ * @param value the new value of the element (in the case of primitive types, the
+ * reflection-defined object-alternative is used).
+ * @param oldValue the previous value of the element (in the case of primitive types, the
+ * reflection-defined object-alternative is used).
* @param index the index in the array of the updated element.
*/
- public ElementUpdatedEvent (
- int targetOid, String name, Object value, Object oldValue, int index)
+ public ElementUpdatedEvent (int targetOid, String name, Object value, Object ovalue, int index)
{
super(targetOid, name);
_value = value;
- _oldValue = oldValue;
+ _oldValue = ovalue;
_index = index;
}
/**
- * Constructs a blank instance of this event in preparation for
- * unserialization from the network.
+ * Constructs a blank instance of this event in preparation for unserialization from the
+ * network.
*/
public ElementUpdatedEvent ()
{
@@ -76,8 +72,7 @@ public class ElementUpdatedEvent extends NamedEvent
}
/**
- * Returns the value of the element prior to the application of this
- * event.
+ * Returns the value of the element prior to the application of this event.
*/
public Object getOldValue ()
{
@@ -93,8 +88,8 @@ public class ElementUpdatedEvent extends NamedEvent
}
/**
- * Returns the new value of the element as a short. This will fail if
- * the element in question is not a short.
+ * Returns the new value of the element as a short. This will fail if the element in question
+ * is not a short.
*/
public short getShortValue ()
{
@@ -102,8 +97,8 @@ public class ElementUpdatedEvent extends NamedEvent
}
/**
- * Returns the new value of the element as an int. This will fail if
- * the element in question is not an int.
+ * Returns the new value of the element as an int. This will fail if the element in question is
+ * not an int.
*/
public int getIntValue ()
{
@@ -111,8 +106,8 @@ public class ElementUpdatedEvent extends NamedEvent
}
/**
- * Returns the new value of the element as a long. This will fail if
- * the element in question is not a long.
+ * Returns the new value of the element as a long. This will fail if the element in question is
+ * not a long.
*/
public long getLongValue ()
{
@@ -120,8 +115,8 @@ public class ElementUpdatedEvent extends NamedEvent
}
/**
- * Returns the new value of the element as a float. This will fail if
- * the element in question is not a float.
+ * Returns the new value of the element as a float. This will fail if the element in question
+ * is not a float.
*/
public float getFloatValue ()
{
@@ -129,47 +124,50 @@ public class ElementUpdatedEvent extends NamedEvent
}
/**
- * Returns the new value of the element as a double. This will fail if
- * the element in question is not a double.
+ * Returns the new value of the element as a double. This will fail if the element in question
+ * is not a double.
*/
public double getDoubleValue ()
{
return ((Double)_value).doubleValue();
}
- /**
- * Applies this element update to the object.
- */
+ @Override // from DEvent
+ public boolean alreadyApplied ()
+ {
+ return (_oldValue == UNSET_OLD_VALUE);
+ }
+
+ @Override // from DEvent
public boolean applyToObject (DObject target)
throws ObjectAccessException
{
- try {
- // fetch the array field from the object
- Field field = target.getClass().getField(_name);
- Class ftype = field.getType();
+ if (!alreadyApplied()) {
+ try {
+ // fetch the array field from the object
+ Field field = target.getClass().getField(_name);
+ Class ftype = field.getType();
- // sanity check
- if (!ftype.isArray()) {
- String msg = "Requested to set element on non-array field.";
- throw new Exception(msg);
- }
+ // sanity check
+ if (!ftype.isArray()) {
+ String msg = "Requested to set element on non-array field.";
+ throw new Exception(msg);
+ }
- // grab the previous value to provide to interested parties
- if (_oldValue == UNSET_OLD_VALUE) {
+ // grab the previous value to provide to interested parties
_oldValue = Array.get(field.get(target), _index);
+
+ // we don't do any magical expansion or any funny business; the array should be big
+ // enough to contain the value being updated or we'll throw an
+ // ArrayIndexOutOfBoundsException
+ Array.set(field.get(target), _index, _value);
+
+ } catch (Exception e) {
+ String msg = "Error updating element [field=" + _name + ", index=" + _index + "]";
+ throw new ObjectAccessException(msg, e);
}
-
- // we don't do any magical expansion or any funny business;
- // the array should be big enough to contain the value being
- // updated or we'll throw an ArrayIndexOutOfBoundsException
- Array.set(field.get(target), _index, _value);
- return true;
-
- } catch (Exception e) {
- String msg = "Error updating element [field=" + _name +
- ", index=" + _index + "]";
- throw new ObjectAccessException(msg, e);
}
+ return true;
}
// documentation inherited
diff --git a/src/java/com/threerings/presents/dobj/EntryAddedEvent.java b/src/java/com/threerings/presents/dobj/EntryAddedEvent.java
index 6377fe7ac..8c069c97e 100644
--- a/src/java/com/threerings/presents/dobj/EntryAddedEvent.java
+++ b/src/java/com/threerings/presents/dobj/EntryAddedEvent.java
@@ -73,39 +73,19 @@ public class EntryAddedEvent extends NamedEvent
return _entry;
}
- /**
- * Applies this event to the object.
- */
+ @Override // from DEvent
+ public boolean alreadyApplied ()
+ {
+ return _alreadyApplied;
+ }
+
+ @Override // from DEvent
public boolean applyToObject (DObject target)
throws ObjectAccessException
{
if (!_alreadyApplied) {
if (!target.getSet(_name).add(_entry)) {
- // If this entry is already in the set, then don't notify our listeners of the
- // event add, because nothing was actually added; the DSet will have already
- // complained; this happens occasionally due to a race condition between client
- // object subscription and set addition, for example, the follow sequence of events
- // can take place:
- //
- // 1. SUBSCRIBE arrives from client, AccessObjectEvent posted;
- //
- // 2. addToSet() called on server, value immediately added to set and
- // EntryAddedEvent is posted;
- //
- // 3. AccessObjectEvent processed; client is added to proxy subscriber list,
- // DObject serialized and sent to client; client receives object which already has
- // the entry in the set;
- //
- // 4. EntryAddedEvent processed, client is a subscriber so it is forwarded along;
- // when it arrives at the client, we find ourselves in this situation.
- //
- // Fixing this would require either a) giving up immediate adding to the DSet when
- // an event is posted, but we add/update immediately because it makes our lives
- // *vastly* simpler in a thousand other cases, or b) doing some magic in the DSet
- // where the server "sees" the new entry added, but if the DSet is serialized
- // before the EntryAddedEvent comes through to "commit" that entry it is not
- // included. This is probably a good idea and maybe we'll do it sometime.
- return false;
+ return false; // DSet will have already complained
}
}
return true;
diff --git a/src/java/com/threerings/presents/dobj/EntryRemovedEvent.java b/src/java/com/threerings/presents/dobj/EntryRemovedEvent.java
index 26d6008d3..0e38ce0a3 100644
--- a/src/java/com/threerings/presents/dobj/EntryRemovedEvent.java
+++ b/src/java/com/threerings/presents/dobj/EntryRemovedEvent.java
@@ -72,13 +72,17 @@ public class EntryRemovedEvent extends NamedEvent
return _oldEntry;
}
- /**
- * Applies this event to the object.
- */
+ @Override // from DEvent
+ public boolean alreadyApplied ()
+ {
+ return (_oldEntry != UNSET_OLD_ENTRY);
+ }
+
+ @Override // from DEvent
public boolean applyToObject (DObject target)
throws ObjectAccessException
{
- if (_oldEntry == UNSET_OLD_ENTRY) {
+ if (!alreadyApplied()) {
DSet set = target.getSet(_name);
// remove, fetch the previous value for interested callers
_oldEntry = set.removeKey(_key);
diff --git a/src/java/com/threerings/presents/dobj/EntryUpdatedEvent.java b/src/java/com/threerings/presents/dobj/EntryUpdatedEvent.java
index 55053ea46..261d98121 100644
--- a/src/java/com/threerings/presents/dobj/EntryUpdatedEvent.java
+++ b/src/java/com/threerings/presents/dobj/EntryUpdatedEvent.java
@@ -73,14 +73,18 @@ public class EntryUpdatedEvent extends NamedEvent
return _oldEntry;
}
- /**
- * Applies this event to the object.
- */
+ @Override // from DEvent
+ public boolean alreadyApplied ()
+ {
+ return (_oldEntry != UNSET_OLD_ENTRY);
+ }
+
+ @Override // from DEvent
public boolean applyToObject (DObject target)
throws ObjectAccessException
{
// only apply the change if we haven't already
- if (_oldEntry == UNSET_OLD_ENTRY) {
+ if (!alreadyApplied()) {
DSet set = target.getSet(_name);
// fetch the previous value for interested callers
_oldEntry = set.update(_entry);
diff --git a/src/java/com/threerings/presents/dobj/ProxySubscriber.java b/src/java/com/threerings/presents/dobj/ProxySubscriber.java
index 752b7a422..ec90f9378 100644
--- a/src/java/com/threerings/presents/dobj/ProxySubscriber.java
+++ b/src/java/com/threerings/presents/dobj/ProxySubscriber.java
@@ -22,10 +22,9 @@
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.
+ * 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
*/
diff --git a/src/java/com/threerings/presents/peer/server/PeerClient.java b/src/java/com/threerings/presents/peer/server/PeerClient.java
index e559d80eb..23086ef78 100644
--- a/src/java/com/threerings/presents/peer/server/PeerClient.java
+++ b/src/java/com/threerings/presents/peer/server/PeerClient.java
@@ -33,24 +33,6 @@ import com.threerings.presents.peer.net.PeerBootstrapData;
*/
public class PeerClient extends PresentsClient
{
- @Override // documentation inherited
- public synchronized void mapSubscrip (DObject object)
- {
- super.mapSubscrip(object);
- if (object instanceof NodeObject) {
- _peermgr.clientSubscribedToNode(_cloid);
- }
- }
-
- @Override // documentation inherited
- public synchronized void unmapSubscrip (int oid)
- {
- if (_subscrips.get(oid) instanceof NodeObject) {
- _peermgr.clientUnsubscribedFromNode(_cloid);
- }
- super.unmapSubscrip(oid);
- }
-
/**
* Creates a peer client and provides it with a reference to the peer
* manager. This is only done by the {@link PeerClientFactory}.
@@ -98,16 +80,22 @@ public class PeerClient extends PresentsClient
_cloid = _clobj.getOid();
}
- @Override // documentation inherited
- protected void clearSubscrips (boolean verbose)
+ @Override // from PresentsClient
+ protected void subscribedToObject (DObject object)
{
- for (DObject object : _subscrips.values()) {
- if (object instanceof NodeObject) {
- _peermgr.clientUnsubscribedFromNode(_cloid);
- break;
- }
+ super.subscribedToObject(object);
+ if (object instanceof NodeObject) {
+ _peermgr.clientSubscribedToNode(_cloid);
+ }
+ }
+
+ @Override // from PresentsClient
+ protected void unsubscribedFromObject (DObject object)
+ {
+ super.unsubscribedFromObject(object);
+ if (object instanceof NodeObject) {
+ _peermgr.clientUnsubscribedFromNode(_cloid);
}
- super.clearSubscrips(verbose);
}
protected PeerManager _peermgr;
diff --git a/src/java/com/threerings/presents/server/PresentsClient.java b/src/java/com/threerings/presents/server/PresentsClient.java
index 809c00db2..e0485e5c8 100644
--- a/src/java/com/threerings/presents/server/PresentsClient.java
+++ b/src/java/com/threerings/presents/server/PresentsClient.java
@@ -76,7 +76,7 @@ import com.threerings.presents.server.net.MessageHandler;
* from the conmgr thread and therefore also need not be synchronized.
*/
public class PresentsClient
- implements ProxySubscriber, MessageHandler, ClientResolutionListener
+ implements MessageHandler, ClientResolutionListener
{
/** Used by {@link #setUsername} to report success or failure. */
public static interface UserChangeListener
@@ -321,31 +321,6 @@ public class PresentsClient
}
}
- /**
- * Makes a note that this client is subscribed to this object so that we can clean up after
- * ourselves if and when the client goes away. This is called by the client internals and
- * needn't be called by code outside the client.
- */
- public synchronized void mapSubscrip (DObject object)
- {
- _subscrips.put(object.getOid(), object);
- }
-
- /**
- * Makes a note that this client is no longer subscribed to this object. The subscription map
- * is used to clean up after the client when it goes away. This is called by the client
- * internals and needn't be called by code outside the client.
- */
- public synchronized void unmapSubscrip (int oid)
- {
- DObject object = _subscrips.remove(oid);
- if (object != null) {
- object.removeSubscriber(this);
- } else {
- Log.warning("Requested to unmap non-existent subscription [oid=" + oid + "].");
- }
- }
-
// from interface ClientResolutionListener
public void clientResolved (Name username, ClientObject clobj)
{
@@ -403,35 +378,6 @@ public class PresentsClient
disp.dispatch(this, message);
}
- // from interface ProxySubscriber
- public void objectAvailable (DObject object)
- {
- if (postMessage(new ObjectResponse(object))) {
- // make a note of this new subscription
- mapSubscrip(object);
- } else {
- // if we failed to send the object response, unsubscribe
- object.removeSubscriber(this);
- }
- }
-
- // from interface ProxySubscriber
- public void requestFailed (int oid, ObjectAccessException cause)
- {
- postMessage(new FailureResponse(oid));
- }
-
- // from interface ProxySubscriber
- public void eventReceived (DEvent event)
- {
- if (event instanceof PresentsDObjectMgr.AccessObjectEvent) {
- Log.warning("Ignoring event that shouldn't be forwarded " + event + ".");
- Thread.dumpStack();
- } else {
- postMessage(new EventNotification(event));
- }
- }
-
/**
* Called when {@link #setUsername} has been called and the new client object is about to be
* applied to this client. The old client object will not yet have been destroyed, so any final
@@ -556,18 +502,35 @@ public class PresentsClient
});
}
+ /**
+ * Makes a note that this client is no longer subscribed to this object. The subscription map
+ * is used to clean up after the client when it goes away.
+ */
+ protected synchronized void unmapSubscrip (int oid)
+ {
+ ClientProxy rec;
+ synchronized (_subscrips) {
+ rec = _subscrips.remove(oid);
+ }
+ if (rec != null) {
+ rec.unsubscribe();
+ } else {
+ Log.warning("Requested to unmap non-existent subscription [oid=" + oid + "].");
+ }
+ }
+
/**
* Clears out the tracked client subscriptions. Called when the client goes away and shouldn't
* be called otherwise.
*/
protected void clearSubscrips (boolean verbose)
{
- for (DObject object : _subscrips.values()) {
+ for (ClientProxy rec : _subscrips.values()) {
if (verbose) {
Log.info("Clearing subscription [client=" + this +
- ", obj=" + object.getOid() + "].");
+ ", obj=" + rec.object.getOid() + "].");
}
- object.removeSubscriber(this);
+ rec.unsubscribe();
}
_subscrips.clear();
}
@@ -615,6 +578,20 @@ public class PresentsClient
clearSubscrips(false);
}
+ /**
+ * Called to inform derived classes when the client has subscribed to a distributed object.
+ */
+ protected void subscribedToObject (DObject object)
+ {
+ }
+
+ /**
+ * Called to inform derived classes when the client has unsubscribed from a distributed object.
+ */
+ protected void unsubscribedFromObject (DObject object)
+ {
+ }
+
/**
* This is called once we have a handle on the client distributed object. It sends a bootstrap
* notification to the client with all the information it will need to interact with the
@@ -803,6 +780,72 @@ public class PresentsClient
buf.append(", out=").append(_messagesOut);
}
+ /**
+ * Creates a properly initialized inner-class proxy subscriber.
+ */
+ protected ClientProxy createProxySubscriber ()
+ {
+ return new ClientProxy();
+ }
+
+ /** Used to track information about an object subscription. */
+ protected class ClientProxy implements ProxySubscriber
+ {
+ public DObject object;
+
+ public void unsubscribe ()
+ {
+ object.removeSubscriber(this);
+ unsubscribedFromObject(object);
+ }
+
+ // from interface ProxySubscriber
+ public void objectAvailable (DObject object)
+ {
+ if (postMessage(new ObjectResponse(object))) {
+ _firstEventId = PresentsServer.omgr.getNextEventId(false);
+ this.object = object;
+ synchronized (_subscrips) {
+ // make a note of this new subscription
+ _subscrips.put(object.getOid(), this);
+ }
+ subscribedToObject(object);
+
+ } else {
+ // if we failed to send the object response, unsubscribe
+ object.removeSubscriber(this);
+ }
+ }
+
+ // from interface ProxySubscriber
+ public void requestFailed (int oid, ObjectAccessException cause)
+ {
+ postMessage(new FailureResponse(oid));
+ }
+
+ // from interface ProxySubscriber
+ public void eventReceived (DEvent event)
+ {
+ if (event instanceof PresentsDObjectMgr.AccessObjectEvent) {
+ Log.warning("Ignoring event that shouldn't be forwarded " + event + ".");
+ Thread.dumpStack();
+ return;
+ }
+
+ // if this message was posted before we received this object and has already been
+ // applied, then this event has already been applied to this object and we should not
+ // forward it, it is equivalent to all the events applied to the object before we
+ // became a subscriber
+ if (event.eventId < _firstEventId) {
+ return;
+ }
+
+ postMessage(new EventNotification(event));
+ }
+
+ protected long _firstEventId;
+ }
+
/**
* Message dispatchers are used to dispatch each different type of upstream message. We can
* look the dispatcher up in a table and invoke it through an overloaded member which is faster
@@ -827,7 +870,7 @@ public class PresentsClient
// Log.info("Subscribing [client=" + client + ", oid=" + req.getOid() + "].");
// forward the subscribe request to the omgr for processing
- PresentsServer.omgr.subscribeToObject(req.getOid(), client);
+ PresentsServer.omgr.subscribeToObject(req.getOid(), client.createProxySubscriber());
}
}
@@ -842,9 +885,7 @@ public class PresentsClient
int oid = req.getOid();
// Log.info("Unsubscribing " + client + " [oid=" + oid + "].");
- // forward the unsubscribe request to the omgr for processing
- PresentsServer.omgr.unsubscribeFromObject(oid, client);
- // update our subscription tracking table
+ // unsubscribe from the object and clear out our proxy
client.unmapSubscrip(oid);
// post a response to the client letting them know that we will no longer send them
@@ -904,7 +945,7 @@ public class PresentsClient
protected Name _username;
protected Connection _conn;
protected ClientObject _clobj;
- protected HashIntMap _subscrips = new HashIntMap();
+ protected HashIntMap _subscrips = new HashIntMap();
protected ClassLoader _loader;
/** The time at which this client started their session. */
diff --git a/src/java/com/threerings/presents/server/PresentsDObjectMgr.java b/src/java/com/threerings/presents/server/PresentsDObjectMgr.java
index 9980729d8..3438e5822 100644
--- a/src/java/com/threerings/presents/server/PresentsDObjectMgr.java
+++ b/src/java/com/threerings/presents/server/PresentsDObjectMgr.java
@@ -93,6 +93,17 @@ public class PresentsDObjectMgr
PresentsServer.registerReporter(this);
}
+ /**
+ * Returns the id to be assigned to the next event posted to the event queue.
+ *
+ * @param increment if true, the event id will be incremented so that the caller can "claim"
+ * the returned event id.
+ */
+ public synchronized long getNextEventId (boolean increment)
+ {
+ return increment ? _nextEventId++ : _nextEventId;
+ }
+
/**
* Sets up an access controller that will be provided to any distributed objects created on the
* server. The controllers can subsequently be overridden if desired, but a default controller
@@ -169,7 +180,8 @@ public class PresentsDObjectMgr
// from interface DObjectManager
public void postEvent (DEvent event)
{
- // just append it to the queue
+ // assign the event's id and append it to the queue
+ event.eventId = getNextEventId(true);
_evqueue.append(event);
}
@@ -973,6 +985,9 @@ public class PresentsDObjectMgr
/** Used during unit profiling for timing values. */
protected Perf _timer = Perf.getPerf();
+ /** A monotonically increasing counter used to assign an id to all dispatched events. */
+ protected long _nextEventId = 1;
+
/** Used to profile our events and runnable units. */
protected HashMap _profiles = new HashMap();