AfflictionNetworks/narya
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Widening.

Browse Source 
git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@4527 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2007-02-02 23:49:24 +00:00
parent 175eb754f4
commit 12410ee32c
1 changed files with 68 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/presents/client/ClientDObjectMgr.java
+68 -98
View File
@@ -40,10 +40,9 @@ import com.threerings.presents.dobj.*;
import com.threerings.presents.net.*; import com.threerings.presents.net.*;
/** /**
* The client distributed object manager manages a set of proxy objects * The client distributed object manager manages a set of proxy objects which mirror the
* which mirror the distributed objects maintained on the server. * distributed objects maintained on the server. Requests for modifications, etc. are forwarded to
* Requests for modifications, etc. are forwarded to the server and events * the server and events are dispatched from the server to this client for objects to which this
* are dispatched from the server to this client for objects to which this
* client is subscribed. * client is subscribed.
*/ */
public class ClientDObjectMgr public class ClientDObjectMgr
@@ -52,20 +51,17 @@ public class ClientDObjectMgr
/** /**
* Constructs a client distributed object manager. * Constructs a client distributed object manager.
* *
* @param comm a communicator instance by which it can communicate * @param comm a communicator instance by which it can communicate with the server.
* with the server. * @param client a reference to the client that is managing this whole communications and event
* @param client a reference to the client that is managing this whole * dispatch business.
* communications and event dispatch business.
*/ */
public ClientDObjectMgr (Communicator comm, Client client) public ClientDObjectMgr (Communicator comm, Client client)
{ {
_comm = comm; _comm = comm;
_client = client; _client = client;
// register a debug hook for dumping all objects in the // register a debug hook for dumping all objects in the distributed object table
// distributed object table DebugChords.registerHook(DUMP_OTABLE_MODMASK, DUMP_OTABLE_KEYCODE, DUMP_OTABLE_HOOK);
DebugChords.registerHook(
DUMP_OTABLE_MODMASK, DUMP_OTABLE_KEYCODE, DUMP_OTABLE_HOOK);
// register a flush interval // register a flush interval
new Interval(client.getRunQueue()) { new Interval(client.getRunQueue()) {
@@ -83,33 +79,21 @@ public class ClientDObjectMgr
} }
// inherit documentation from the interface // inherit documentation from the interface
public <T extends DObject> void subscribeToObject ( public <T extends DObject> void subscribeToObject (int oid, Subscriber<T> target)
int oid, Subscriber<T> target)
{ {
if (oid <= 0) { if (oid <= 0) {
target.requestFailed( target.requestFailed(oid, new ObjectAccessException("Invalid oid " + oid + "."));
oid, new ObjectAccessException("Invalid oid " + oid + "."));
} else { } else {
queueAction(oid, target, true); queueAction(oid, target, true);
} }
} }
// inherit documentation from the interface // inherit documentation from the interface
public <T extends DObject> void unsubscribeFromObject ( public <T extends DObject> void unsubscribeFromObject (int oid, Subscriber<T> target)
int oid, Subscriber<T> target)
{ {
queueAction(oid, target, false); queueAction(oid, target, false);
} }
protected <T extends DObject> void queueAction (
int oid, Subscriber<T> target, boolean subscribe)
{
// queue up an action
_actions.append(new ObjectAction<T>(oid, target, subscribe));
// and queue up the omgr to get invoked on the invoker thread
_client.getRunQueue().postRunnable(this);
}
// inherit documentation from the interface // inherit documentation from the interface
public void postEvent (DEvent event) public void postEvent (DEvent event)
{ {
@@ -120,16 +104,14 @@ public class ClientDObjectMgr
// inherit documentation from the interface // inherit documentation from the interface
public void removedLastSubscriber (DObject obj, boolean deathWish) public void removedLastSubscriber (DObject obj, boolean deathWish)
{ {
// if this object has a registered flush delay, don't can it just yet, // if this object has a registered flush delay, don't can it just yet, just slip it onto
// just slip it onto the flush queue // the flush queue
Class<?> oclass = obj.getClass(); Class<?> oclass = obj.getClass();
for (Class<?> dclass : _delays.keySet()) { for (Class<?> dclass : _delays.keySet()) {
if (dclass.isAssignableFrom(oclass)) { if (dclass.isAssignableFrom(oclass)) {
long expire = System.currentTimeMillis() + long expire = System.currentTimeMillis() + _delays.get(dclass).longValue();
_delays.get(dclass).longValue();
_flushes.put(obj.getOid(), new FlushRecord(obj, expire)); _flushes.put(obj.getOid(), new FlushRecord(obj, expire));
// Log.info("Flushing " + obj.getOid() + " at " + // Log.info("Flushing " + obj.getOid() + " at " + new java.util.Date(expire));
// new java.util.Date(expire));
return; return;
} }
} }
@@ -149,9 +131,8 @@ public class ClientDObjectMgr
} }
/** /**
* Called by the communicator when a downstream message arrives from * Called by the communicator when a downstream message arrives from the network layer. We
* the network layer. We queue it up for processing and request some * queue it up for processing and request some processing time on the main thread.
* processing time on the main thread.
*/ */
public void processMessage (DownstreamMessage msg) public void processMessage (DownstreamMessage msg)
{ {
@@ -162,8 +143,8 @@ public class ClientDObjectMgr
} }
/** /**
* Invoked on the main client thread to process any newly arrived * Invoked on the main client thread to process any newly arrived messages that we have waiting
* messages that we have waiting in our queue. * in our queue.
*/ */
public void run () public void run ()
{ {
@@ -186,8 +167,7 @@ public class ClientDObjectMgr
} else if (obj instanceof UnsubscribeResponse) { } else if (obj instanceof UnsubscribeResponse) {
int oid = ((UnsubscribeResponse)obj).getOid(); int oid = ((UnsubscribeResponse)obj).getOid();
if (_dead.remove(oid) == null) { if (_dead.remove(oid) == null) {
Log.warning("Received unsub ACK from unknown object " + Log.warning("Received unsub ACK from unknown object [oid=" + oid + "].");
"[oid=" + oid + "].");
} }
} else if (obj instanceof FailureResponse) { } else if (obj instanceof FailureResponse) {
@@ -208,14 +188,22 @@ public class ClientDObjectMgr
} }
} }
protected <T extends DObject> void queueAction (
int oid, Subscriber<T> target, boolean subscribe)
{
// queue up an action
_actions.append(new ObjectAction<T>(oid, target, subscribe));
// and queue up the omgr to get invoked on the invoker thread
_client.getRunQueue().postRunnable(this);
}
/** /**
* Called when a new event arrives from the server that should be * Called when a new event arrives from the server that should be dispatched to subscribers
* dispatched to subscribers here on the client. * here on the client.
*/ */
protected void dispatchEvent (DEvent event) protected void dispatchEvent (DEvent 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) {
List<DEvent> events = ((CompoundEvent)event).getEvents(); List<DEvent> events = ((CompoundEvent)event).getEvents();
int ecount = events.size(); int ecount = events.size();
@@ -230,8 +218,7 @@ public class ClientDObjectMgr
DObject target = _ocache.get(toid); DObject target = _ocache.get(toid);
if (target == null) { if (target == null) {
if (!_dead.containsKey(toid)) { if (!_dead.containsKey(toid)) {
Log.warning("Unable to dispatch event on non-proxied " + Log.warning("Unable to dispatch event on non-proxied object " + event + ".");
"object [event=" + event + "].");
} }
return; return;
} }
@@ -240,12 +227,10 @@ public class ClientDObjectMgr
// apply the event to the object // apply the event to the object
boolean notify = event.applyToObject(target); boolean notify = event.applyToObject(target);
// if this is an object destroyed event, we need to remove the // if this is an object destroyed event, we need to remove the object from our table
// object from our object table
if (event instanceof ObjectDestroyedEvent) { if (event instanceof ObjectDestroyedEvent) {
// Log.info("Pitching destroyed object " + // Log.info("Pitching destroyed object [oid=" + toid +
// "[oid=" + toid + ", class=" + // ", class=" + StringUtil.shortClassName(target) + "].");
// StringUtil.shortClassName(target) + "].");
_ocache.remove(toid); _ocache.remove(toid);
} }
@@ -255,18 +240,16 @@ public class ClientDObjectMgr
} }
} catch (Exception e) { } catch (Exception e) {
Log.warning("Failure processing event [event=" + event + Log.warning("Failure processing event [event=" + event + ", target=" + target + "].");
", target=" + target + "].");
Log.logStackTrace(e); Log.logStackTrace(e);
} }
} }
/** /**
* Registers this object in our proxy cache and notifies the * Registers this object in our proxy cache and notifies the subscribers that were waiting for
* subscribers that were waiting for subscription to this object. * subscription to this object.
*/ */
protected <T extends DObject> void registerObjectAndNotify ( protected <T extends DObject> void registerObjectAndNotify (ObjectResponse<T> orsp)
ObjectResponse<T> orsp)
{ {
// let the object know that we'll be managing it // let the object know that we'll be managing it
T obj = orsp.getObject(); T obj = orsp.getObject();
@@ -278,14 +261,13 @@ public class ClientDObjectMgr
// let the penders know that the object is available // let the penders know that the object is available
PendingRequest<?> req = _penders.remove(obj.getOid()); PendingRequest<?> req = _penders.remove(obj.getOid());
if (req == null) { if (req == null) {
Log.warning("Got object, but no one cares?! " + Log.warning("Got object, but no one cares?! [oid=" + obj.getOid() +
"[oid=" + obj.getOid() + ", obj=" + obj + "]."); ", obj=" + obj + "].");
return; return;
} }
for (int i = 0; i < req.targets.size(); i++) { for (int i = 0; i < req.targets.size(); i++) {
@SuppressWarnings("unchecked") Subscriber<T> target = @SuppressWarnings("unchecked") Subscriber<T> target = (Subscriber<T>)req.targets.get(i);
(Subscriber<T>)req.targets.get(i);
// add them as a subscriber // add them as a subscriber
obj.addSubscriber(target); obj.addSubscriber(target);
// and let them know that the object is in // and let them know that the object is in
@@ -294,16 +276,15 @@ public class ClientDObjectMgr
} }
/** /**
* Notifies the subscribers that had requested this object (for * Notifies the subscribers that had requested this object (for subscription) that it is not
* subscription) that it is not available. * available.
*/ */
protected void notifyFailure (int oid) protected void notifyFailure (int oid)
{ {
// let the penders know that the object is not available // let the penders know that the object is not available
PendingRequest<?> req = _penders.remove(oid); PendingRequest<?> req = _penders.remove(oid);
if (req == null) { if (req == null) {
Log.warning("Failed to get object, but no one cares?! " + Log.warning("Failed to get object, but no one cares?! [oid=" + oid + "].");
"[oid=" + oid + "].");
return; return;
} }
@@ -314,8 +295,8 @@ public class ClientDObjectMgr
} }
/** /**
* This is guaranteed to be invoked via the invoker and can safely do * This is guaranteed to be invoked via the invoker and can safely do main thread type things
* main thread type things like call back to the subscriber. * like call back to the subscriber.
*/ */
protected <T extends DObject> void doSubscribe (ObjectAction<T> action) protected <T extends DObject> void doSubscribe (ObjectAction<T> action)
{ {
@@ -338,11 +319,9 @@ public class ClientDObjectMgr
} }
// see if we've already got an outstanding request for this object // see if we've already got an outstanding request for this object
@SuppressWarnings("unchecked") PendingRequest<T> req = @SuppressWarnings("unchecked") PendingRequest<T> req = (PendingRequest<T>)_penders.get(oid);
(PendingRequest<T>)_penders.get(oid);
if (req != null) { if (req != null) {
// add this subscriber to the list of subscribers to be // add this subscriber to the list to be notified when the request is satisfied
// notified when the request is satisfied
req.addTarget(target); req.addTarget(target);
return; return;
} }
@@ -358,8 +337,8 @@ public class ClientDObjectMgr
} }
/** /**
* This is guaranteed to be invoked via the invoker and can safely do * This is guaranteed to be invoked via the invoker and can safely do main thread type things
* main thread type things like call back to the subscriber. * like call back to the subscriber.
*/ */
protected void doUnsubscribe (int oid, Subscriber target) protected void doUnsubscribe (int oid, Subscriber target)
{ {
@@ -368,39 +347,36 @@ public class ClientDObjectMgr
dobj.removeSubscriber(target); dobj.removeSubscriber(target);
} else { } else {
Log.info("Requested to remove subscriber from " + Log.info("Requested to remove subscriber from non-proxied object [oid=" + oid +
"non-proxied object [oid=" + oid +
", sub=" + target + "]."); ", sub=" + target + "].");
} }
} }
/** /**
* Flushes a distributed object subscription, issuing an unsubscribe * Flushes a distributed object subscription, issuing an unsubscribe request to the server.
* request to the server.
*/ */
protected void flushObject (DObject obj) protected void flushObject (DObject obj)
{ {
// move this object into the dead pool so that we don't claim to // move this object into the dead pool so that we don't claim to have it around anymore;
// have it around anymore; once our unsubscribe message is // once our unsubscribe message is processed, it'll be 86ed
// processed, it'll be 86ed
int ooid = obj.getOid(); int ooid = obj.getOid();
_ocache.remove(ooid); _ocache.remove(ooid);
_dead.put(ooid, obj); _dead.put(ooid, obj);
// ship off an unsubscribe message to the server; we'll remove the // ship off an unsubscribe message to the server; we'll remove the object from our table
// object from our table when we get the unsub ack // when we get the unsub ack
_comm.postMessage(new UnsubscribeRequest(ooid)); _comm.postMessage(new UnsubscribeRequest(ooid));
} }
/** /**
* Called periodically to flush any objects that have been lingering * Called periodically to flush any objects that have been lingering due to a previously
* due to a previously enacted flush delay. * enacted flush delay.
*/ */
protected void flushObjects () protected void flushObjects ()
{ {
long now = System.currentTimeMillis(); long now = System.currentTimeMillis();
for (Iterator<IntMap.IntEntry<FlushRecord>> iter = for (Iterator<IntMap.IntEntry<FlushRecord>> iter = _flushes.intEntrySet().iterator();
_flushes.intEntrySet().iterator(); iter.hasNext(); ) { iter.hasNext(); ) {
IntMap.IntEntry<FlushRecord> entry = iter.next(); IntMap.IntEntry<FlushRecord> entry = iter.next();
int oid = entry.getIntKey(); int oid = entry.getIntKey();
FlushRecord rec = entry.getValue(); FlushRecord rec = entry.getValue();
@@ -413,8 +389,7 @@ public class ClientDObjectMgr
} }
/** /**
* The object action is used to queue up a subscribe or unsubscribe * The object action is used to queue up a subscribe or unsubscribe request.
* request.
*/ */
protected static final class ObjectAction<T extends DObject> protected static final class ObjectAction<T extends DObject>
{ {
@@ -438,8 +413,7 @@ public class ClientDObjectMgr
protected static final class PendingRequest<T extends DObject> protected static final class PendingRequest<T extends DObject>
{ {
public int oid; public int oid;
public ArrayList<Subscriber<T>> targets = public ArrayList<Subscriber<T>> targets = new ArrayList<Subscriber<T>>();
new ArrayList<Subscriber<T>>();
public PendingRequest (int oid) public PendingRequest (int oid)
{ {
@@ -468,8 +442,7 @@ public class ClientDObjectMgr
} }
} }
/** A reference to the communicator that sends and receives messages /** A reference to the communicator that sends and receives messages for this client. */
* for this client. */
protected Communicator _comm; protected Communicator _comm;
/** A reference to our client instance. */ /** A reference to our client instance. */
@@ -485,8 +458,7 @@ public class ClientDObjectMgr
protected HashIntMap<DObject> _dead = new HashIntMap<DObject>(); protected HashIntMap<DObject> _dead = new HashIntMap<DObject>();
/** Pending object subscriptions. */ /** Pending object subscriptions. */
protected HashIntMap<PendingRequest<?>> _penders = protected HashIntMap<PendingRequest<?>> _penders = new HashIntMap<PendingRequest<?>>();
new HashIntMap<PendingRequest<?>>();
/** A mapping from distributed object class to flush delay. */ /** A mapping from distributed object class to flush delay. */
protected HashMap<Class<?>,Long> _delays = new HashMap<Class<?>,Long>(); protected HashMap<Class<?>,Long> _delays = new HashMap<Class<?>,Long>();
@@ -494,8 +466,7 @@ public class ClientDObjectMgr
/** A set of objects waiting to be flushed. */ /** A set of objects waiting to be flushed. */
protected HashIntMap<FlushRecord> _flushes = new HashIntMap<FlushRecord>(); protected HashIntMap<FlushRecord> _flushes = new HashIntMap<FlushRecord>();
/** A debug hook that allows the dumping of all objects in the object /** A debug hook that allows the dumping of all objects in the object table out to the log. */
* table out to the log. */
protected DebugChords.Hook DUMP_OTABLE_HOOK = new DebugChords.Hook() { protected DebugChords.Hook DUMP_OTABLE_HOOK = new DebugChords.Hook() {
public void invoke () { public void invoke () {
Log.info("Dumping " + _ocache.size() + " objects:"); Log.info("Dumping " + _ocache.size() + " objects:");
@@ -506,8 +477,7 @@ public class ClientDObjectMgr
}; };
/** The modifiers for our dump table debug hook (Alt+Shift). */ /** The modifiers for our dump table debug hook (Alt+Shift). */
protected static int DUMP_OTABLE_MODMASK = protected static int DUMP_OTABLE_MODMASK = KeyEvent.ALT_DOWN_MASK|KeyEvent.SHIFT_DOWN_MASK;
KeyEvent.ALT_DOWN_MASK|KeyEvent.SHIFT_DOWN_MASK;
/** The key code for our dump table debug hook (o). */ /** The key code for our dump table debug hook (o). */
protected static int DUMP_OTABLE_KEYCODE = KeyEvent.VK_O; protected static int DUMP_OTABLE_KEYCODE = KeyEvent.VK_O;