diff --git a/src/java/com/threerings/presents/dobj/DObjectManager.java b/src/java/com/threerings/presents/dobj/DObjectManager.java index 43cbc73dc..38b66282f 100644 --- a/src/java/com/threerings/presents/dobj/DObjectManager.java +++ b/src/java/com/threerings/presents/dobj/DObjectManager.java @@ -1,5 +1,5 @@ // -// $Id: DObjectManager.java,v 1.3 2001/06/01 07:12:13 mdb Exp $ +// $Id: DObjectManager.java,v 1.4 2001/06/05 22:44:31 mdb Exp $ package com.threerings.cocktail.cher.dobj; @@ -68,6 +68,16 @@ public interface DObjectManager */ public void fetchObject (int oid, Subscriber target); + /** + * Requests that the specified subscriber be unsubscribed from the + * object identified by the supplied object id. + * + * @param oid The object id of the distributed object from which + * unsubscription is desired. + * @param target The subscriber to be unsubscribed. + */ + public void unsubscribeFromObject (int oid, Subscriber target); + /** * Posts a distributed object event into the system. Instead of * requesting the modification of a distributed object attribute by diff --git a/src/java/com/threerings/presents/net/FetchRequest.java b/src/java/com/threerings/presents/net/FetchRequest.java index b0d611caa..f3ec531f8 100644 --- a/src/java/com/threerings/presents/net/FetchRequest.java +++ b/src/java/com/threerings/presents/net/FetchRequest.java @@ -1,5 +1,5 @@ // -// $Id: FetchRequest.java,v 1.2 2001/05/30 23:58:31 mdb Exp $ +// $Id: FetchRequest.java,v 1.3 2001/06/05 22:44:31 mdb Exp $ package com.threerings.cocktail.cher.net; @@ -34,6 +34,14 @@ public class FetchRequest extends UpstreamMessage return TYPE; } + /** + * Returns the oid of the object we desire to fetch. + */ + public int getOid () + { + return _oid; + } + public void writeTo (DataOutputStream out) throws IOException { diff --git a/src/java/com/threerings/presents/net/ForwardEventRequest.java b/src/java/com/threerings/presents/net/ForwardEventRequest.java index 3b92a42e8..0fc480bec 100644 --- a/src/java/com/threerings/presents/net/ForwardEventRequest.java +++ b/src/java/com/threerings/presents/net/ForwardEventRequest.java @@ -1,5 +1,5 @@ // -// $Id: ForwardEventRequest.java,v 1.5 2001/06/05 21:29:51 mdb Exp $ +// $Id: ForwardEventRequest.java,v 1.6 2001/06/05 22:44:31 mdb Exp $ package com.threerings.cocktail.cher.net; @@ -35,6 +35,14 @@ public class ForwardEventRequest extends UpstreamMessage return TYPE; } + /** + * Returns the event that we wish to have forwarded. + */ + public DEvent getEvent () + { + return _event; + } + public void writeTo (DataOutputStream out) throws IOException { diff --git a/src/java/com/threerings/presents/net/SubscribeRequest.java b/src/java/com/threerings/presents/net/SubscribeRequest.java index 659014b2f..9897db0cc 100644 --- a/src/java/com/threerings/presents/net/SubscribeRequest.java +++ b/src/java/com/threerings/presents/net/SubscribeRequest.java @@ -1,5 +1,5 @@ // -// $Id: SubscribeRequest.java,v 1.2 2001/05/30 23:58:31 mdb Exp $ +// $Id: SubscribeRequest.java,v 1.3 2001/06/05 22:44:31 mdb Exp $ package com.threerings.cocktail.cher.net; @@ -34,6 +34,14 @@ public class SubscribeRequest extends UpstreamMessage return TYPE; } + /** + * Returns the oid of the object to which we desire subscription. + */ + public int getOid () + { + return _oid; + } + public void writeTo (DataOutputStream out) throws IOException { diff --git a/src/java/com/threerings/presents/net/UnsubscribeRequest.java b/src/java/com/threerings/presents/net/UnsubscribeRequest.java index 30e2c62c2..95c770cca 100644 --- a/src/java/com/threerings/presents/net/UnsubscribeRequest.java +++ b/src/java/com/threerings/presents/net/UnsubscribeRequest.java @@ -1,5 +1,5 @@ // -// $Id: UnsubscribeRequest.java,v 1.3 2001/06/05 21:29:51 mdb Exp $ +// $Id: UnsubscribeRequest.java,v 1.4 2001/06/05 22:44:31 mdb Exp $ package com.threerings.cocktail.cher.net; @@ -34,6 +34,14 @@ public class UnsubscribeRequest extends UpstreamMessage return TYPE; } + /** + * Returns the oid of the object from which we are unsubscribing. + */ + public int getOid () + { + return _oid; + } + public void writeTo (DataOutputStream out) throws IOException { diff --git a/src/java/com/threerings/presents/server/ClientManager.java b/src/java/com/threerings/presents/server/ClientManager.java index 4e89922ef..17bf2a2b1 100644 --- a/src/java/com/threerings/presents/server/ClientManager.java +++ b/src/java/com/threerings/presents/server/ClientManager.java @@ -1,5 +1,5 @@ // -// $Id: ClientManager.java,v 1.2 2001/06/02 01:30:37 mdb Exp $ +// $Id: ClientManager.java,v 1.3 2001/06/05 22:44:31 mdb Exp $ package com.threerings.cocktail.cher.server; @@ -42,15 +42,19 @@ public class ClientManager implements ConnectionObserver String username = creds.getUsername(); // see if there's a client already registered with this username - Client client = (Client)_clients.get(username); + Client client = (Client)_usermap.get(username); if (client != null) { Log.info("Session resumed [username=" + username + ", conn=" + conn + "]."); + client.resumeSession(conn); } else { Log.info("Session initiated [username=" + username + ", conn=" + conn + "]."); + // create a new client and stick'em in the table + client = new Client(this, username, conn); + _usermap.put(username, conn); } } @@ -66,7 +70,18 @@ public class ClientManager implements ConnectionObserver public synchronized void connectionFailed (Connection conn, IOException fault) { - Log.info("Connection failed: " + conn + ": " + fault); + // remove the client from the connection map + Client client = (Client)_connmap.remove(conn); + if (client != null) { + Log.info("Unmapped failed client [client=" + client + + ", conn=" + conn + ", fault=" + fault + "]."); + // let the client know things went haywire + client.connectionFailed(fault); + + } else { + Log.info("Unmapped connection failed? [conn=" + conn + + ", fault=" + fault + "]."); + } } /** @@ -76,8 +91,24 @@ public class ClientManager implements ConnectionObserver */ public synchronized void connectionClosed (Connection conn) { - Log.info("Connection closed: " + conn); + // remove the client from the connection map + Client client = (Client)_connmap.remove(conn); + if (client != null) { + Log.info("Unmapped client [client=" + client + + ", conn=" + conn + "]."); + } else { + Log.info("Closed unmapped connection? [conn=" + conn + "]."); + } } - protected HashMap _clients = new HashMap(); + /** + * Called by the client instance when the client requests a logoff. + * This is called from the conmgr thread. + */ + synchronized void clientDidEndSession (Client client) + { + } + + protected HashMap _usermap = new HashMap(); + protected HashMap _connmap = new HashMap(); } diff --git a/src/java/com/threerings/presents/server/PresentsClient.java b/src/java/com/threerings/presents/server/PresentsClient.java index 94021a428..f51107f4e 100644 --- a/src/java/com/threerings/presents/server/PresentsClient.java +++ b/src/java/com/threerings/presents/server/PresentsClient.java @@ -1,8 +1,11 @@ // -// $Id: PresentsClient.java,v 1.1 2001/06/02 01:30:37 mdb Exp $ +// $Id: PresentsClient.java,v 1.2 2001/06/05 22:44:31 mdb Exp $ package com.threerings.cocktail.cher.server; +import java.io.IOException; +import java.util.HashMap; + import com.threerings.cocktail.cher.Log; import com.threerings.cocktail.cher.dobj.*; import com.threerings.cocktail.cher.net.*; @@ -28,12 +31,21 @@ public class Client implements Subscriber, MessageHandler * Constructs a new client instance bound to the specified username * and initially associated with the specified connection instance. */ - public Client (String username, Connection conn) + public Client (ClientManager cmgr, String username, Connection conn) { + _cmgr = cmgr; _username = username; setConnection(conn); } + /** + * Returns the username with which this client instance is associated. + */ + public String getUsername () + { + return _username; + } + /** * Called by the client manager when a new connection arrives that * authenticates as this already established client. This must only be @@ -41,14 +53,16 @@ public class Client implements Subscriber, MessageHandler */ public void resumeSession (Connection conn) { + Connection oldconn = getConnection(); + // check to see if we've already got a connection object, in which // case it's probably stale - if (_conn != null) { - Log.info("Closing stale connection [old=" + _conn + + if (oldconn != null) { + Log.info("Closing stale connection [old=" + oldconn + ", new=" + conn + "]."); // close the old connection (which results in everything being // properly unregistered) - _conn.close(); + oldconn.close(); } // start using the new connection @@ -57,41 +71,226 @@ public class Client implements Subscriber, MessageHandler Log.info("Session resumed [client=" + this + "]."); } - protected void setConnection (Connection conn) + /** + * Called by the connection manager when this client's connection + * fails. This is invoked on the conmgr thread, and should behave + * accordingly. + */ + public void connectionFailed (IOException fault) + { + // clear out our connection reference + setConnection(null); + } + + protected synchronized void setConnection (Connection conn) { // keep a handle to the new connection _conn = conn; - // tell the connection to pass messages on to us - _conn.setMessageHandler(this); + // tell the connection to pass messages on to us (if we're setting + // a connection rather than clearing one out) + if (_conn != null) { + _conn.setMessageHandler(this); + } + } + + /** + * The connection instance must be accessed via this member function + * because it is read from both the dobjmgr and conmgr threads and is + * modified by the conmgr thread. + * + * @return The connection instance associated with this client or null + * if the client is not currently connected. + */ + protected synchronized Connection getConnection () + { + return _conn; } // documentation inherited from interface public void handleMessage (UpstreamMessage message) { + // we dispatch to a message dispatcher that is specialized for the + // particular class of message that we received + MessageDispatcher disp = (MessageDispatcher) + _disps.get(message.getClass()); + if (disp == null) { + Log.warning("No dispacther for message [msg=" + message + "]."); + return; + } + + // otherwise pass the message on to the dispatcher + disp.dispatch(this, message); } // documentation inherited from interface public void objectAvailable (DObject object) { // queue up an object response - _conn.postMessage(new ObjectResponse(object)); + Connection conn = getConnection(); + if (conn != null) { + conn.postMessage(new ObjectResponse(object)); + } else { + Log.info("Dropped object available notification " + + "[client=" + this + ", oid=" + object.getOid() + "]."); + } } // documentation inherited from interface public void requestFailed (int oid, ObjectAccessException cause) { - _conn.postMessage(new FailureResponse(oid)); + Connection conn = getConnection(); + if (conn != null) { + conn.postMessage(new FailureResponse(oid)); + } else { + Log.info("Dropped failure notification " + + "[client=" + this + ", oid=" + oid + + ", cause=" + cause + "]."); + } } // documentation inherited from interface public boolean handleEvent (DEvent event, DObject target) { // forward the event to the client - _conn.postMessage(new EventNotification(event)); + Connection conn = getConnection(); + if (conn != null) { + conn.postMessage(new EventNotification(event)); + } else { + Log.info("Dropped event forward notification " + + "[client=" + this + ", event=" + event + "]."); + } return true; } + /** + * 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 (so we + * think) than doing a bunch of instanceofs. + */ + protected static interface MessageDispatcher + { + /** + * Dispatch the supplied message for the specified client. + */ + public void dispatch (Client client, UpstreamMessage mge); + } + + /** + * Processes subscribe requests. + */ + protected static class SubscribeDispatcher implements MessageDispatcher + { + public void dispatch (Client client, UpstreamMessage msg) + { + SubscribeRequest req = (SubscribeRequest)msg; + Log.info("Subscribing [client=" + client + + ", oid=" + req.getOid() + "]."); + // forward the subscribe request to the omgr for processing + CherServer.omgr.subscribeToObject(req.getOid(), client); + } + } + + /** + * Processes fetch requests. + */ + protected static class FetchDispatcher implements MessageDispatcher + { + public void dispatch (Client client, UpstreamMessage msg) + { + FetchRequest req = (FetchRequest)msg; + Log.info("Fetching [client=" + client + + ", oid=" + req.getOid() + "]."); + // forward the fetch request to the omgr for processing + CherServer.omgr.fetchObject(req.getOid(), client); + } + } + + /** + * Processes unsubscribe requests. + */ + protected static class UnsubscribeDispatcher implements MessageDispatcher + { + public void dispatch (Client client, UpstreamMessage msg) + { + UnsubscribeRequest req = (UnsubscribeRequest)msg; + Log.info("Unsubscribing [client=" + client + + ", oid=" + req.getOid() + "]."); + // forward the unsubscribe request to the omgr for processing + CherServer.omgr.unsubscribeFromObject(req.getOid(), client); + } + } + + /** + * Processes forward event requests. + */ + protected static class ForwardEventDispatcher implements MessageDispatcher + { + public void dispatch (Client client, UpstreamMessage msg) + { + ForwardEventRequest req = (ForwardEventRequest)msg; + Log.info("Forwarding event [client=" + client + + ", event=" + req.getEvent() + "]."); + // forward the event to the omgr for processing + CherServer.omgr.postEvent(req.getEvent()); + } + } + + /** + * Processes ping requests. + */ + protected static class PingDispatcher implements MessageDispatcher + { + public void dispatch (Client client, UpstreamMessage msg) + { + Log.info("Received client ping [client=" + client + "]."); + // send a pong response + Connection conn = client.getConnection(); + if (conn != null) { + conn.postMessage(new PongResponse()); + } else { + Log.info("Dropped pong response [client=" + client + "]."); + } + } + } + + /** + * Processes logoff requests. + */ + protected static class LogoffDispatcher implements MessageDispatcher + { + public void dispatch (Client client, UpstreamMessage msg) + { + Log.info("Client requested logoff " + + "[client=" + client + "]."); + // close our connection (which results in everything being + // properly unregistered) + Connection conn = client.getConnection(); + if (conn != null) { + conn.close(); + } else { + Log.info("Unable to close connection for logoff request " + + "[client=" + client + "]."); + } + // then let the client manager know what's up + client._cmgr.clientDidEndSession(client); + } + } + + protected ClientManager _cmgr; protected String _username; - protected transient Connection _conn; + protected Connection _conn; + + protected static HashMap _disps = new HashMap(); + + // register our message dispatchers + static { + _disps.put(SubscribeRequest.class, new SubscribeDispatcher()); + _disps.put(FetchRequest.class, new FetchDispatcher()); + _disps.put(UnsubscribeRequest.class, new UnsubscribeDispatcher()); + _disps.put(ForwardEventRequest.class, new ForwardEventDispatcher()); + _disps.put(PingRequest.class, new PingDispatcher()); + _disps.put(LogoffRequest.class, new LogoffDispatcher()); + } } diff --git a/src/java/com/threerings/presents/server/PresentsDObjectMgr.java b/src/java/com/threerings/presents/server/PresentsDObjectMgr.java index c6c0d9a4e..b7507fd14 100644 --- a/src/java/com/threerings/presents/server/PresentsDObjectMgr.java +++ b/src/java/com/threerings/presents/server/PresentsDObjectMgr.java @@ -1,5 +1,5 @@ // -// $Id: PresentsDObjectMgr.java,v 1.3 2001/06/02 01:30:37 mdb Exp $ +// $Id: PresentsDObjectMgr.java,v 1.4 2001/06/05 22:44:31 mdb Exp $ package com.threerings.cocktail.cher.server; @@ -48,14 +48,24 @@ public class CherDObjectMgr implements DObjectManager public void subscribeToObject (int oid, Subscriber target) { // queue up an access object event - postEvent(new AccessObjectEvent(oid, target, true)); + postEvent(new AccessObjectEvent(oid, target, + AccessObjectEvent.SUBSCRIBE)); } // inherit documentation from the interface public void fetchObject (int oid, Subscriber target) { // queue up an access object event - postEvent(new AccessObjectEvent(oid, target, false)); + postEvent(new AccessObjectEvent(oid, target, + AccessObjectEvent.FETCH)); + } + + // inherit documentation from the interface + public void unsubscribeFromObject (int oid, Subscriber target) + { + // queue up an access object event + postEvent(new AccessObjectEvent(oid, target, + AccessObjectEvent.UNSUBSCRIBE)); } // inherit documentation from the interface @@ -228,13 +238,17 @@ public class CherDObjectMgr implements DObjectManager */ protected class AccessObjectEvent extends DEvent { + public static final int SUBSCRIBE = 0; + public static final int FETCH = 1; + public static final int UNSUBSCRIBE = 2; + public AccessObjectEvent (int oid, Subscriber target, - boolean subscribe) + int action) { super(0); // target the bogus object _oid = oid; _target = target; - _subscribe = subscribe; + _action = action; } public boolean applyToObject (DObject target) @@ -243,6 +257,14 @@ public class CherDObjectMgr implements DObjectManager // look up the target object DObject obj = (DObject)_objects.get(_oid); + // if we're unsubscribing, take care of that and get on out + if (_action == UNSUBSCRIBE) { + if (obj != null) { + obj.removeSubscriber(_target); + } + return false; + } + // if it don't exist, let them know if (obj == null) { _target.requestFailed(_oid, new NoSuchObjectException(_oid)); @@ -257,7 +279,7 @@ public class CherDObjectMgr implements DObjectManager } // if they wanted to subscribe, do so - if (_subscribe) { + if (_action == SUBSCRIBE) { obj.addSubscriber(_target); } @@ -271,7 +293,7 @@ public class CherDObjectMgr implements DObjectManager protected int _oid; protected Subscriber _target; - protected boolean _subscribe; + protected int _action; } protected class ShutdownEvent extends DEvent