Implemented a Ray-proposed solution where we keep track of events that were
(pre-applied and) posted before we received our object but processed (and hence dispatched to us) after we received our object. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@4715 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
@@ -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. <em>Do not
|
||||
* construct these objects by hand.</em> Use {@link
|
||||
* Constructs a new attribute changed event on the specified target object with the supplied
|
||||
* attribute name and value. <em>Do not construct these objects by hand.</em> 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;
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -73,39 +73,19 @@ public class EntryAddedEvent<T extends DSet.Entry> 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;
|
||||
|
||||
@@ -72,13 +72,17 @@ public class EntryRemovedEvent<T extends DSet.Entry> 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<T> set = target.getSet(_name);
|
||||
// remove, fetch the previous value for interested callers
|
||||
_oldEntry = set.removeKey(_key);
|
||||
|
||||
@@ -73,14 +73,18 @@ public class EntryUpdatedEvent<T extends DSet.Entry> 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<T> set = target.getSet(_name);
|
||||
// fetch the previous value for interested callers
|
||||
_oldEntry = set.update(_entry);
|
||||
|
||||
@@ -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
|
||||
*/
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<DObject>(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<DObject>(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<DObject> _subscrips = new HashIntMap<DObject>();
|
||||
protected HashIntMap<ClientProxy> _subscrips = new HashIntMap<ClientProxy>();
|
||||
protected ClassLoader _loader;
|
||||
|
||||
/** The time at which this client started their session. */
|
||||
|
||||
@@ -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<String,UnitProfile> _profiles = new HashMap<String,UnitProfile>();
|
||||
|
||||
|
||||
Reference in New Issue
Block a user