DObjectManager: added unsubscribeFromObject() for subscribers that wish to

unsubscribe in the absence of a reference to the object in question and
potentially from an unsavory thread.

Client, ClientManager, etc.: Further work on the server-side of the
distributed part of the distributed object functionality.


git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@27 542714f4-19e9-0310-aa3c-eee0fc999fb1

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

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
     {

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
     {

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
     {

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
     {

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();
 }

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) {
+        if (oldconn != null) {
-            Log.info("Closing stale connection [old=" + _conn +
+            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
+        // tell the connection to pass messages on to us (if we're setting
-        _conn.setMessageHandler(this);
+        // 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());
+    }
 }

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