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

Further wired up the client side of the distributed object system. Removed

Browse Source 
the facilities for fetching (without subscribing to) an object. This is
done extremely rarely and the user might as well just subscribe and
immediately unsubscribe because the dichotomy between fetching and
subscribing just served to overly complicate the internals for no good
reason.


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@30 542714f4-19e9-0310-aa3c-eee0fc999fb1
This commit is contained in:
Michael Bayne
2001-06-09 23:39:04 +00:00
parent 06f6fa26d5
commit 6a1de87f2a
20 changed files with 574 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/java/com/threerings/presents/client/Client.java
+54 -6
View File
@@ -1,5 +1,5 @@
//
// $Id: Client.java,v 1.4 2001/05/30 23:58:31 mdb Exp $
// $Id: Client.java,v 1.5 2001/06/09 23:39:03 mdb Exp $
package com.threerings.cocktail.cher.client;
@@ -7,24 +7,47 @@ import java.util.ArrayList;
import java.util.List;
import com.threerings.cocktail.cher.Log;
import com.threerings.cocktail.cher.dobj.DObjectManager;
import com.threerings.cocktail.cher.net.Credentials;
import com.threerings.cocktail.cher.net.Registry;
/**
* Through the client object, a connection to the system is established
* and maintained. The client object maintains two separate threads (a
* reader and a writer) through which all network traffic is managed.
* reader and a writer) by which all network traffic is managed.
*/
public class Client
{
/**
* Constructs a client object with the supplied credentials. These
* creds will be used to authenticate with any server to which this
* client attempts to connect.
* This is used by the client to allow dobj event dispatching to take
* place along side the activities of the rest of the application
* (usually this means running dobj events on the AWT thread).
*/
public Client (Credentials creds)
public static interface Invoker
{
/**
* Requests that the supplied runnable be queued up for invocation
* on the main event dispatching thread of the application.
*/
public void invokeLater (Runnable run);
}
/**
* Constructs a client object with the supplied credentials and
* invoker. The creds will be used to authenticate with any server to
* which this client attempts to connect. The invoker is used to
* operate the distributed object event dispatch mechanism. To allow
* the dobj event dispatch to coexist with threads like the AWT
* thread, the client will request that the invoker queue up a
* runnable whenever there are distributed object events that need to
* be processed. The invoker can then queue that runnable up on the
* AWT thread if it is so inclined to make life simpler for the rest
* of the application.
*/
public Client (Credentials creds, Invoker invoker)
{
_creds = creds;
_invoker = invoker;
}
/**
@@ -69,6 +92,15 @@ public class Client
_port = port;
}
/**
* Returns the invoker in use by this client. This can be used to
* queue up event dispatching stints.
*/
public Invoker getInvoker ()
{
return _invoker;
}
/**
* Returns the hostname of the server to which this client is
* currently configured to connect.
@@ -96,6 +128,20 @@ public class Client
return _creds;
}
/**
* Returns the distributed object manager associated with this
* session. This reference is only valid for the duration of the
* session and a new reference must be obtained if the client
* disconnects and reconnects to the server.
*
* @return the dobjmgr in effect or null if we have no established
* connection to the server.
*/
public DObjectManager getDObjectManager ()
{
return (_comm != null) ? _comm.getDObjectManager() : null;
}
/**
* Requests that this client connect and logon to the server with
* which it was previously configured.
@@ -180,6 +226,8 @@ public class Client
}
protected Credentials _creds;
protected Invoker _invoker;
protected String _hostname;
protected int _port;
src/java/com/threerings/presents/client/ClientDObjectMgr.java
+275
View File
@@ -0,0 +1,275 @@
//
// $Id: ClientDObjectMgr.java,v 1.1 2001/06/09 23:39:03 mdb Exp $
package com.threerings.cocktail.cher.client;
import java.util.ArrayList;
import com.samskivert.util.IntMap;
import com.samskivert.util.Queue;
import com.threerings.cocktail.cher.Log;
import com.threerings.cocktail.cher.dobj.*;
import com.threerings.cocktail.cher.net.*;
/**
* The client distributed object manager manages a set of proxy objects
* which mirror the distributed objects maintained on the server.
* Requests for modifications, etc. are forwarded to the server and events
* are dispatched from the server to this client for objects to which this
* client is subscribed.
*/
public class ClientDObjectMgr
implements DObjectManager, Runnable
{
/**
* Constructs a client distributed object manager.
*
* @param comm a communicator instance by which it can communicate
* with the server.
* @param client a reference to the client that is managing this whole
* communications and event dispatch business.
*/
public ClientDObjectMgr (Communicator comm, Client client)
{
_comm = comm;
_client = client;
}
// inherit documentation from the interface
public void createObject (Class dclass, Subscriber target,
boolean subscribe)
{
// not presently supported
throw new RuntimeException("createObject() not supported");
}
// inherit documentation from the interface
public void subscribeToObject (int oid, Subscriber target)
{
queueAction(oid, target, true);
}
// inherit documentation from the interface
public void unsubscribeFromObject (int oid, Subscriber target)
{
queueAction(oid, target, false);
// forward an unsubscribe request to the server
_comm.postMessage(new UnsubscribeRequest(oid));
}
protected void queueAction (int oid, Subscriber target, boolean subscribe)
{
// queue up an action
_actions.append(new ObjectAction(oid, target, subscribe));
// and queue up the omgr to get invoked on the invoker thread
_client.getInvoker().invokeLater(this);
}
// inherit documentation from the interface
public void postEvent (DEvent event)
{
// send a forward event request to the server
_comm.postMessage(new ForwardEventRequest(event));
}
/**
* Called by the communicator when a downstream message arrives from
* the network layer. We queue it up for processing and request some
* processing time on the main thread.
*/
public void processMessage (DownstreamMessage msg)
{
// append it to our queue
_actions.append(msg);
// and queue ourselves up to be run
_client.getInvoker().invokeLater(this);
}
/**
* Invoked on the AWT thread to process any newly arrived messages
* that we have waiting in our queue.
*/
public void run ()
{
// process all of the events on our queue
Object obj;
while ((obj = _actions.getNonBlocking()) != null) {
// do the proper thing depending on the object
if (obj instanceof EventNotification) {
DEvent evt = ((EventNotification)obj).getEvent();
dispatchEvent(evt);
} else if (obj instanceof ObjectResponse) {
registerObjectAndNotify(((ObjectResponse)obj).getObject());
} else if (obj instanceof FailureResponse) {
int oid = ((FailureResponse)obj).getOid();
notifyFailure(oid);
} else if (obj instanceof PongResponse) {
Log.info("Got pong.");
} else if (obj instanceof ObjectAction) {
ObjectAction act = (ObjectAction)obj;
if (act.subscribe) {
doSubscribe(act.oid, act.target);
} else {
doUnsubscribe(act.oid, act.target);
}
}
}
}
/**
* Called when a new event arrives from the server that should be
* dispatched to subscribers here on the client.
*/
protected void dispatchEvent (DEvent event)
{
Log.info("Dispatch event: " + event);
// look up the object on which we're dispatching this event
DObject target = (DObject)_ocache.get(event.getTargetOid());
if (target == null) {
Log.info("Unable to dispatch event on non-proxied " +
"object [event=" + event + "].");
return;
}
}
/**
* Registers this object in our proxy cache and notifies the
* subscribers that were waiting for subscription to this object.
*/
protected void registerObjectAndNotify (DObject obj)
{
// let the object know that we'll be managing it
obj.setManager(this);
// stick the object into the proxy object table
_ocache.put(obj.getOid(), obj);
// let the penders know that the object is available
PendingRequest req = (PendingRequest)_penders.remove(obj.getOid());
if (req == null) {
Log.warning("Got object, but no one cares?! " +
"[oid=" + obj.getOid() + ", obj=" + obj + "].");
return;
}
for (int i = 0; i < req.targets.size(); i++) {
Subscriber target = (Subscriber)req.targets.get(i);
target.objectAvailable(obj);
}
}
/**
* Notifies the subscribers that had requested this object (for
* subscription) that it is not available.
*/
protected void notifyFailure (int oid)
{
Log.info("Get failed: " + oid);
}
/**
* This is guaranteed to be invoked via the invoker and can safely do
* main thread type things like call back to the subscriber.
*/
protected void doSubscribe (int oid, Subscriber target)
{
Log.info("doSubscribe: " + oid + ": " + target);
// first see if we've already got the object in our table
DObject obj = (DObject)_ocache.get(oid);
if (obj != null) {
// add the subscriber and call them back straight away
obj.addSubscriber(target);
target.objectAvailable(obj);
return;
}
// see if we've already got an outstanding request for this object
PendingRequest req = (PendingRequest)_penders.get(oid);
if (req != null) {
// add this subscriber to the list of subscribers to be
// notified when the request is satisfied
req.addTarget(target);
return;
}
// otherwise we need to create a new request
req = new PendingRequest(oid);
req.addTarget(target);
_penders.put(oid, req);
Log.info("Registering pending request [oid=" + oid + "].");
// and issue a request to get things rolling
_comm.postMessage(new SubscribeRequest(oid));
}
/**
* This is guaranteed to be invoked via the invoker and can safely do
* main thread type things like call back to the subscriber.
*/
protected void doUnsubscribe (int oid, Subscriber target)
{
DObject dobj = (DObject)_ocache.get(oid);
if (dobj != null) {
dobj.removeSubscriber(target);
} else {
Log.info("Requested to remove subscriber from " +
"non-proxied object [oid=" + oid +
", sub=" + target + "].");
}
}
/**
* The object action is used to queue up a subscribe or unsubscribe
* request.
*/
protected class ObjectAction
{
public int oid;
public Subscriber target;
public boolean subscribe;
public ObjectAction (int oid, Subscriber target, boolean subscribe)
{
this.oid = oid;
this.target = target;
this.subscribe = subscribe;
}
}
protected static class PendingRequest
{
public int oid;
public ArrayList targets = new ArrayList();
public PendingRequest (int oid)
{
this.oid = oid;
}
public void addTarget (Subscriber target)
{
targets.add(target);
}
}
protected Communicator _comm;
protected Client _client;
protected Queue _actions = new Queue();
/**
* This table contains all of the distributed objects that are active
* on this client.
*/
protected IntMap _ocache = new IntMap();
/** This table contains pending subscriptions. */
protected IntMap _penders = new IntMap();
}
src/java/com/threerings/presents/client/Communicator.java
+41 -3
View File
@@ -1,5 +1,5 @@
//
// $Id: Communicator.java,v 1.6 2001/05/30 23:58:31 mdb Exp $
// $Id: Communicator.java,v 1.7 2001/06/09 23:39:03 mdb Exp $
package com.threerings.cocktail.cher.client;
@@ -11,6 +11,7 @@ import com.samskivert.util.LoopingThread;
import com.samskivert.util.Queue;
import com.threerings.cocktail.cher.Log;
import com.threerings.cocktail.cher.dobj.DObjectManager;
import com.threerings.cocktail.cher.io.*;
import com.threerings.cocktail.cher.io.ObjectStreamException;
import com.threerings.cocktail.cher.net.*;
@@ -50,6 +51,17 @@ public class Communicator
_client = client;
}
/**
* Returns the distributed object manager in effect for this session.
* This instance is only valid while the client is connected to the
* server. If we become disconnected and have to reconnect, a new omgr
* instance should be obtained.
*/
public DObjectManager getDObjectManager ()
{
return _omgr;
}
/**
* Logs on to the server and initiates our full-duplex message
* exchange.
@@ -144,6 +156,9 @@ public class Communicator
// extract bootstrap information
// create our distributed object manager
_omgr = new ClientDObjectMgr(this, _client);
// create a new writer thread and start it up
if (_writer != null) {
throw new RuntimeException("Writer already started!?");
@@ -176,6 +191,23 @@ public class Communicator
logoff();
}
/**
* Callback called by the reader if the server closes the other end of
* the connection.
*/
protected synchronized void connectionClosed ()
{
// make sure the socket isn't already closed down (meaning we've
// already dealt with the closed connection)
if (_socket == null) {
return;
}
Log.info("Connection closed.");
// now do the whole logoff thing
logoff();
}
/**
* Callback called by the reader thread when it goes away.
*/
@@ -239,6 +271,8 @@ public class Communicator
protected void processMessage (DownstreamMessage msg)
{
Log.info("Process msg: " + msg);
// post this message to the dobjmgr queue
_omgr.processMessage(msg);
}
/**
@@ -342,8 +376,10 @@ public class Communicator
Log.info("Reader thread woken up in time to die.");
} catch (EOFException eofe) {
Log.info("Connection closed by peer.");
// nothing left for us to do
// let the communicator know that our connection was
// closed
connectionClosed();
// and shut ourselves down
shutdown();
} catch (IOException ioe) {
@@ -443,4 +479,6 @@ public class Communicator
/** We use this to frame our downstream messages. */
protected FramedInputStream _fin;
protected DataInputStream _din;
protected ClientDObjectMgr _omgr;
}