Created ServerCommunicator which does all I/O using the ConnectionManager's

polled socket system. This much more cleanly and efficiently integrates peer
(and other server to server) connections into the normal server I/O framework
and should also eliminate for good the annoying server hangs that result when
our old blocking client I/O threads got their pants wedged.


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

src/java/com/threerings/presents/client/BlockingCommunicator.java

@@ -183,17 +183,10 @@ public class BlockingCommunicator extends Communicator
         return _lastWrite;
     }
 
-    /**
+    @Override // from Communicator
-     * Callback called by the reader when the authentication process completes successfully. Here
-     * we extract the bootstrap information for the client and start up the writer thread to manage
-     * the other half of our bi-directional message stream.
-     */
     protected synchronized void logonSucceeded (AuthResponseData data)
     {
-        log.debug("Logon succeeded: " + data);
+        super.logonSucceeded(data);
-
-        // create our distributed object manager
-        _omgr = new ClientDObjectMgr(this, _client);
 
         // create a new writer thread and start it up
         if (_writer != null) {
@@ -201,19 +194,13 @@ public class BlockingCommunicator extends Communicator
         }
         _writer = new Writer();
         _writer.start();
-
-        // fill the auth data into the client's local field so that it can be requested by external
-        // entities
-        _client._authData = data;
-
-        // wait for the bootstrap notification before we claim that we're actually logged on
     }
 
     /**
      * Callback called by the reader or writer thread when something goes awry with our socket
      * connection to the server.
      */
-    protected synchronized void connectionFailed (IOException ioe)
+    protected synchronized void connectionFailed (final IOException ioe)
     {
         // make sure the socket isn't already closed down (meaning we've already dealt with the
         // failed connection)
@@ -224,7 +211,11 @@ public class BlockingCommunicator extends Communicator
         log.info("Connection failed", ioe);
 
         // let the client know that things went south
-        _client.notifyObservers(Client.CLIENT_CONNECTION_FAILED, ioe);
+        notifyClientObservers(new ObserverOps.Client() {
+            protected void notify (ClientObserver obs) {
+                obs.clientConnectionFailed(_client, ioe);
+            }
+        });
 
         // and request that we go through the motions of logging off
         logoff();
@@ -260,7 +251,7 @@ public class BlockingCommunicator extends Communicator
             closeChannel();
 
             // let the client know when we finally go away
-            _client.cleanup(_logonError);
+            clientCleanup(_logonError);
         }
 
         log.debug("Reader thread exited.");
@@ -276,7 +267,11 @@ public class BlockingCommunicator extends Communicator
         log.debug("Writer thread exited.");
 
         // let the client observers know that we're logged off
-        _client.notifyObservers(Client.CLIENT_DID_LOGOFF, null);
+        notifyClientObservers(new ObserverOps.Session() {
+            protected void notify (SessionObserver obs) {
+                obs.clientDidLogoff(_client);
+            }
+        });
 
         // now that the writer thread has gone away, we can safely close our socket and let the
         // client know that the logoff process has completed
@@ -284,7 +279,7 @@ public class BlockingCommunicator extends Communicator
 
         // let the client know when we finally go away
         if (_reader == null) {
-            _client.cleanup(_logonError);
+            clientCleanup(_logonError);
         }
     }
 
@@ -439,14 +434,6 @@ public class BlockingCommunicator extends Communicator
         _datagramChannel.write(buf);
     }
 
-    /**
-     * Makes a note of the time at which we last communicated with the server.
-     */
-    protected synchronized void updateWriteStamp ()
-    {
-        _lastWrite = System.currentTimeMillis();
-    }
-
     /**
      * Reads a new message from the socket (blocking until a message has arrived).
      */
@@ -504,16 +491,6 @@ public class BlockingCommunicator extends Communicator
         }
     }
 
-    /**
-     * Callback called by the reader thread when it has parsed a new message from the socket and
-     * wishes to have it processed.
-     */
-    protected void processMessage (DownstreamMessage msg)
-    {
-        // post this message to the dobjmgr queue
-        _omgr.processMessage(msg);
-    }
-
     protected void openChannel (InetAddress host)
         throws IOException
     {
@@ -545,13 +522,17 @@ public class BlockingCommunicator extends Communicator
         @Override
         protected void willStart ()
         {
-            // first we connect and authenticate with the server
             try {
                 // connect to the server
                 connect();
 
-                // then authenticate
+                // construct an auth request and send it
-                logon();
+                sendMessage(new AuthRequest(_client.getCredentials(), _client.getVersion(),
+                                            _client.getBootGroups()));
+
+                // now wait for the auth response
+                log.debug("Waiting for auth response.");
+                gotAuthResponse((AuthResponse)receiveMessage());
 
             } catch (Exception e) {
                 log.debug("Logon failed: " + e);
@@ -586,30 +567,6 @@ public class BlockingCommunicator extends Communicator
             _oout = new ObjectOutputStream(_fout);
         }
 
-        protected void logon ()
-            throws IOException, LogonException
-        {
-            // construct an auth request and send it
-            AuthRequest req = new AuthRequest(
-                _client.getCredentials(), _client.getVersion(), _client.getBootGroups());
-            sendMessage(req);
-
-            // now wait for the auth response
-            log.debug("Waiting for auth response.");
-            AuthResponse rsp = (AuthResponse)receiveMessage();
-            AuthResponseData data = rsp.getData();
-            log.debug("Got auth response: " + data);
-
-            // if the auth request failed, we want to let the communicator know by throwing a logon
-            // exception
-            if (!data.code.equals(AuthResponseData.SUCCESS)) {
-                throw new LogonException(data.code);
-            }
-
-            // we're all clear. let the communicator know that we're in
-            logonSucceeded(data);
-        }
-
         // now that we're authenticated, we manage the reading half of things by continuously
         // reading messages from the socket and processing them
         @Override
@@ -961,7 +918,6 @@ public class BlockingCommunicator extends Communicator
     protected DatagramChannel _datagramChannel;
     protected Queue<UpstreamMessage> _dataq = new Queue<UpstreamMessage>();
 
-    protected long _lastWrite;
     protected Exception _logonError;
 
     /** We use this to frame our upstream messages. */
@@ -983,7 +939,6 @@ public class BlockingCommunicator extends Communicator
 
     protected DatagramSequencer _sequencer;
 
-    protected ClientDObjectMgr _omgr;
     protected ClassLoader _loader;
 
     /** The number of times per port to try to establish a datagram "connection". */

src/java/com/threerings/presents/client/Client.java

@@ -418,8 +418,12 @@ public class Client
             return false;
         }
 
-        // notify our observers
+        // notify our observers immediately
-        notifyObservers(CLIENT_WILL_LOGON, null);
+        _observers.apply(new ObserverOps.Session() {
+            protected void notify (SessionObserver obs) {
+                obs.clientWillLogon(Client.this);
+            }
+        });
 
         // otherwise create a new communicator instance and start it up.  this will initiate the
         // logon process
@@ -504,7 +508,15 @@ public class Client
         }
 
         // if the request is abortable, let's run it past the observers before we act upon it
-        if (abortable && notifyObservers(CLIENT_WILL_LOGOFF, null)) {
+        final boolean[] rejected = new boolean[] { false };
+        _observers.apply(new ObserverOps.Client() {
+            protected void notify (ClientObserver obs) {
+                if (!obs.clientWillLogoff(Client.this)) {
+                    rejected[0] = true;
+                }
+            }
+        });
+        if (abortable && rejected[0]) {
             return false;
         }
 
@@ -527,7 +539,12 @@ public class Client
     public String[] prepareStandaloneLogon ()
     {
         _standalone = true;
-        notifyObservers(CLIENT_WILL_LOGON, null);
+        // notify our observers immediately
+        _observers.apply(new ObserverOps.Session() {
+            protected void notify (SessionObserver obs) {
+                obs.clientWillLogon(Client.this);
+            }
+        });
         return getBootGroups();
     }
 
@@ -548,7 +565,11 @@ public class Client
      */
     public void standaloneLogoff ()
     {
-        notifyObservers(CLIENT_DID_LOGOFF, null);
+        notifyObservers(new ObserverOps.Session() {
+            protected void notify (SessionObserver obs) {
+                obs.clientDidLogoff(Client.this);
+            }
+        });
         cleanup(null); // this will set _standalone to false
     }
 
@@ -715,9 +736,9 @@ public class Client
      */
     protected void reportLogonTribulations (final LogonException cause)
     {
-        _runQueue.postRunnable(new Runnable() {
+        notifyObservers(new ObserverOps.Client() {
-            public void run () {
+            protected void notify (ClientObserver obs) {
-                notifyObservers(CLIENT_FAILED_TO_LOGON, cause);
+                obs.clientFailedToLogon(Client.this, cause);
             }
         });
     }
@@ -732,16 +753,24 @@ public class Client
         _clobj = clobj;
 
         // let the client know that logon has now fully succeeded
-        notifyObservers(CLIENT_DID_LOGON, null);
+        notifyObservers(new ObserverOps.Session() {
+            protected void notify (SessionObserver obs) {
+                obs.clientDidLogon(Client.this);
+            }
+        });
     }
 
     /**
      * Called by the invocation director if it fails to subscribe to the client object after logon.
      */
-    protected void getClientObjectFailed (Exception cause)
+    protected void getClientObjectFailed (final Exception cause)
     {
         // pass the buck onto the listeners
-        notifyObservers(CLIENT_FAILED_TO_LOGON, cause);
+        notifyObservers(new ObserverOps.Client() {
+            protected void notify (ClientObserver obs) {
+                obs.clientFailedToLogon(Client.this, cause);
+            }
+        });
     }
 
     /**
@@ -753,32 +782,23 @@ public class Client
         _cloid = _clobj.getOid();
 
         // report to our observers
-        notifyObservers(CLIENT_OBJECT_CHANGED, null);
+        notifyObservers(new ObserverOps.Session() {
+            protected void notify (SessionObserver obs) {
+                obs.clientObjectDidChange(Client.this);
+            }
+        });
     }
 
-    protected boolean notifyObservers (int code, Exception cause)
+    protected void notifyObservers (final ObserverOps.Session op)
     {
-        final Notifier noty = new Notifier(code, cause);
+        if (_runQueue == null) {
-        Runnable unit = new Runnable() {
+            _observers.apply(op);
-            public void run () {
-                synchronized (_observers) {
-                    _observers.apply(noty);
-                }
-            }
-        };
-
-        // we need to run immediately if this is WILL_LOGON, WILL_LOGOFF or if we have no RunQueue
-        // (which currently only happens in some really obscure circumstances where we're using a
-        // Client instance on the server so that we can sort of pretend to be a real client)
-        if (code == CLIENT_WILL_LOGON || code == CLIENT_WILL_LOGOFF || _runQueue == null) {
-            unit.run();
-            return noty.getRejected();
-
         } else {
-            // otherwise we can queue this notification up with our RunQueue and ensure that it's
+            _runQueue.postRunnable(new Runnable() {
-            // run on the proper thread
+                public void run () {
-            _runQueue.postRunnable(unit);
+                    _observers.apply(op);
-            return false;
+                }
+            });
         }
     }
 
@@ -809,11 +829,15 @@ public class Client
                 // now that the communicator is cleaned up; this allows a logon failure listener to
                 // immediately try another logon (hopefully with something changed like the server
                 // or port)
-                if (logonError != null) {
+                notifyObservers(new ObserverOps.Client() {
-                    notifyObservers(CLIENT_FAILED_TO_LOGON, logonError);
+                    protected void notify (ClientObserver obs) {
-                } else {
+                        if (logonError != null) {
-                    notifyObservers(CLIENT_DID_CLEAR, null);
+                            obs.clientFailedToLogon(Client.this, logonError);
-                }
+                        } else {
+                            obs.clientDidClear(Client.this);
+                        }
+                    }
+                });
             }
         });
     }
@@ -844,81 +868,6 @@ public class Client
         return false;
     }
 
-    /**
-     * Used to notify client observers of events.
-     */
-    protected class Notifier
-        implements ObserverList.ObserverOp<SessionObserver>
-    {
-        public Notifier (int code, Exception cause)
-        {
-            _code = code;
-            _cause = cause;
-            _rejected = false;
-        }
-
-        public boolean getRejected ()
-        {
-            return _rejected;
-        }
-
-        public boolean apply (SessionObserver obs)
-        {
-            switch (_code) {
-            case CLIENT_WILL_LOGON:
-                obs.clientWillLogon(Client.this);
-                break;
-
-            case CLIENT_DID_LOGON:
-                obs.clientDidLogon(Client.this);
-                break;
-
-            case CLIENT_FAILED_TO_LOGON:
-                if (obs instanceof ClientObserver) {
-                    ((ClientObserver)obs).clientFailedToLogon(Client.this, _cause);
-                }
-                break;
-
-            case CLIENT_OBJECT_CHANGED:
-                obs.clientObjectDidChange(Client.this);
-                break;
-
-            case CLIENT_CONNECTION_FAILED:
-                if (obs instanceof ClientObserver) {
-                    ((ClientObserver)obs).clientConnectionFailed(Client.this, _cause);
-                }
-                break;
-
-            case CLIENT_WILL_LOGOFF:
-                if (obs instanceof ClientObserver) {
-                    if (!((ClientObserver)obs).clientWillLogoff(Client.this)) {
-                        _rejected = true;
-                    }
-                }
-                break;
-
-            case CLIENT_DID_LOGOFF:
-                obs.clientDidLogoff(Client.this);
-                break;
-
-            case CLIENT_DID_CLEAR:
-                if (obs instanceof ClientObserver) {
-                    ((ClientObserver)obs).clientDidClear(Client.this);
-                }
-                break;
-
-            default:
-                throw new RuntimeException("Invalid code supplied to notifyObservers: " + _code);
-            }
-
-            return true;
-        }
-
-        protected int _code;
-        protected Exception _cause;
-        protected boolean _rejected;
-    }
-
     /** Handles the process of switching between servers.
      * See {@link Client#moveToServer(String,int[],int[],ConfirmListener)}. */
     protected class ServerSwitcher extends ClientAdapter
@@ -1037,8 +986,7 @@ public class Client
     protected int[] _datagramPorts;
 
     /** Our list of client observers. */
-    protected ObserverList<SessionObserver> _observers =
+    protected ObserverList<SessionObserver> _observers = ObserverList.newSafeInOrder();
-        new ObserverList<SessionObserver>(ObserverList.SAFE_IN_ORDER_NOTIFY);
 
     /** The entity that manages our network communications. */
     protected Communicator _comm;
@@ -1075,14 +1023,4 @@ public class Client
 
     /** How often we recompute our time offset from the server. */
     protected static final long CLOCK_SYNC_INTERVAL = 600 * 1000L;
-
-    // client observer codes
-    static final int CLIENT_WILL_LOGON = 0;
-    static final int CLIENT_DID_LOGON = 1;
-    static final int CLIENT_FAILED_TO_LOGON = 2;
-    static final int CLIENT_OBJECT_CHANGED = 3;
-    static final int CLIENT_CONNECTION_FAILED = 4;
-    static final int CLIENT_WILL_LOGOFF = 5;
-    static final int CLIENT_DID_LOGOFF = 6;
-    static final int CLIENT_DID_CLEAR = 7;
 }

src/java/com/threerings/presents/client/ClientDObjectMgr.java

@@ -44,10 +44,10 @@ import com.threerings.presents.dobj.ObjectDestroyedEvent;
 import com.threerings.presents.dobj.Subscriber;
 import com.threerings.presents.net.BootstrapData;
 import com.threerings.presents.net.BootstrapNotification;
-import com.threerings.presents.net.DownstreamMessage;
 import com.threerings.presents.net.EventNotification;
 import com.threerings.presents.net.FailureResponse;
 import com.threerings.presents.net.ForwardEventRequest;
+import com.threerings.presents.net.Message;
 import com.threerings.presents.net.ObjectResponse;
 import com.threerings.presents.net.PongResponse;
 import com.threerings.presents.net.SubscribeRequest;
@@ -157,10 +157,10 @@ public class ClientDObjectMgr
     }
 
     /**
-     * Called by the communicator when a downstream message arrives from the network layer. We
+     * Called by the communicator when a message arrives from the network layer. We queue it up for
-     * queue it up for processing and request some processing time on the main thread.
+     * processing and request some processing time on the main thread.
      */
-    public void processMessage (DownstreamMessage msg)
+    public void processMessage (Message msg)
     {
         // append it to our queue
         _actions.append(msg);
@@ -211,6 +211,9 @@ public class ClientDObjectMgr
                 } else {
                     doUnsubscribe(act.oid, act.target);
                 }
+
+            } else {
+                log.warning("Unknown action", "action", obj);
             }
         }
     }

src/java/com/threerings/presents/client/Communicator.java

@@ -21,6 +21,9 @@
 
 package com.threerings.presents.client;
 
+import com.threerings.presents.net.AuthResponse;
+import com.threerings.presents.net.AuthResponseData;
+import com.threerings.presents.net.Message;
 import com.threerings.presents.net.UpstreamMessage;
 
 /**
@@ -66,7 +69,70 @@ public abstract class Communicator
     /**
      * Returns the time at which we last sent a packet to the server.
      */
-    public abstract long getLastWrite ();
+    public long getLastWrite ()
+    {
+        return _lastWrite;
+    }
+
+    /**
+     * Makes a note of the time at which we last communicated with the server.
+     */
+    protected synchronized void updateWriteStamp ()
+    {
+        _lastWrite = System.currentTimeMillis();
+    }
+
+    /**
+     * Subclasses must call this method when they receive the authentication response.
+     */
+    protected void gotAuthResponse (AuthResponse rsp)
+        throws LogonException
+    {
+        AuthResponseData data = rsp.getData();
+        if (!data.code.equals(AuthResponseData.SUCCESS)) {
+            throw new LogonException(data.code);
+        }
+        logonSucceeded(data);
+    }
+
+    /**
+     * Called when the authentication process completes successfully. Derived classes can override
+     * this method and complete any initialization that need wait for authentication success.
+     */
+    protected synchronized void logonSucceeded (AuthResponseData data)
+    {
+        // create our distributed object manager
+        _omgr = new ClientDObjectMgr(this, _client);
+
+        // fill the auth data into the client's local field so that it can be requested by external
+        // entities
+        _client._authData = data;
+
+        // wait for the bootstrap notification before we claim that we're actually logged on
+    }
+
+    /**
+     * Callback called by the reader thread when it has parsed a new message from the socket and
+     * wishes to have it processed.
+     */
+    protected void processMessage (Message msg)
+    {
+        // post this message to the dobjmgr queue
+        _omgr.processMessage(msg);
+    }
+
+    protected void notifyClientObservers (ObserverOps.Session op)
+    {
+        _client.notifyObservers(op);
+    }
+
+    protected void clientCleanup (Exception logonError)
+    {
+        _client.cleanup(logonError);
+        _client = null; // prevent any post-cleanup tomfoolery
+    }
 
     protected Client _client;
+    protected ClientDObjectMgr _omgr;
+    protected long _lastWrite;
 }

src/java/com/threerings/presents/client/ObserverOps.java

@@ -2,7 +2,7 @@
 // $Id$
 //
 // Narya library - tools for developing networked games
-// Copyright (C) 2002-2007 Three Rings Design, Inc., All Rights Reserved
+// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved
 // http://www.threerings.net/code/narya/
 //
 // This library is free software; you can redistribute it and/or modify it
@@ -19,19 +19,31 @@
 // License along with this library; if not, write to the Free Software
 // Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 
-package com.threerings.presents.server.net;
+package com.threerings.presents.client;
 
-import com.threerings.presents.net.UpstreamMessage;
+import com.samskivert.util.ObserverList;
 
 /**
- * After the connection object has parsed an entire upstream message, it
+ * Used to notify session and client observers.
- * passes it on to its message handler.
  */
-public interface MessageHandler
+public class ObserverOps
 {
-    /**
+    public abstract static class Session implements ObserverList.ObserverOp<SessionObserver>
-     * Called when a complete message has been parsed from incoming
+    {
-     * network data.
+        public boolean apply (SessionObserver obs) {
-     */
+            notify(obs);
-    void handleMessage (UpstreamMessage message);
+            return true;
+        }
+        protected abstract void notify (SessionObserver obs);
+    }
+
+    public abstract static class Client extends Session
+    {
+        @Override public void notify (SessionObserver obs) {
+            if (obs instanceof ClientObserver) {
+                notify((ClientObserver)obs);
+            }
+        }
+        protected abstract void notify (ClientObserver obs);
+    }
 }

src/java/com/threerings/presents/net/DownstreamMessage.java

@@ -22,16 +22,15 @@
 package com.threerings.presents.net;
 
 /**
- * This class encapsulates a message in the distributed object protocol
+ * This class encapsulates a message in the distributed object protocol that flows from the server
- * that flows from the server to the client. Downstream messages include
+ * to the client. Downstream messages include object subscription, event forwarding and session
- * object subscription, event forwarding and session management.
+ * management.
  */
 public abstract class DownstreamMessage extends Message
 {
     /**
-     * The message id of the upstream message with which this downstream
+     * The message id of the upstream message with which this downstream message is associated (or
-     * message is associated (or -1 if it is not associated with any
+     * -1 if it is not associated with any upstream message).
-     * upstream message).
      */
     public short messageId = -1;
 

src/java/com/threerings/presents/net/Message.java

@@ -28,6 +28,9 @@ import com.threerings.io.SimpleStreamableObject;
  */
 public abstract class Message extends SimpleStreamableObject
 {
+    /** A timestamp indicating when this message was received from the network. */
+    public transient long received;
+
     /**
      * Sets the message transport parameters.  For messages received over the network, these
      * describe the mode of transport over which the message was received.  When sending messages,

src/java/com/threerings/presents/net/UpstreamMessage.java

@@ -22,27 +22,22 @@
 package com.threerings.presents.net;
 
 /**
- * This class encapsulates a message in the distributed object protocol
+ * This class encapsulates a message in the distributed object protocol that flows from the client
- * that flows from the client to the server.  Upstream messages include
+ * to the server.  Upstream messages include object subscription, event forwarding and session
- * object subscription, event forwarding and session management.
+ * management.
  */
 public abstract class UpstreamMessage extends Message
 {
     /**
-     * This is a unique (within the context of a reasonable period of
+     * This is a unique (within the context of a reasonable period of time) identifier assigned to
-     * time) identifier assigned to each upstream message. The message ids
+     * each upstream message. The message ids are used to correlate a downstream response message
-     * are used to correlate a downstream response message to the
+     * to the appropriate upstream request message.
-     * appropriate upstream request message.
      */
     public short messageId;
 
-    /** A timestamp indicating when this upstream message was received. */
-    public transient long received;
-
     /**
-     * Each upstream message derived class must provide a zero argument
+     * Each upstream message derived class must provide a zero argument constructor so that it can
-     * constructor so that it can be unserialized when read from the
+     * be unserialized when read from the network.
-     * network.
      */
     public UpstreamMessage ()
     {
@@ -59,8 +54,7 @@ public abstract class UpstreamMessage extends Message
     }
 
     /**
-     * Returns the next message id suitable for use by an upstream
+     * Returns the next message id suitable for use by an upstream message.
-     * message.
      */
     protected static synchronized short nextMessageId ()
     {
@@ -69,8 +63,8 @@ public abstract class UpstreamMessage extends Message
     }
 
     /**
-     * This is used to generate monotonically increasing message ids on
+     * This is used to generate monotonically increasing message ids on the client as new messages
-     * the client as new messages are generated.
+     * are generated.
      */
     protected static short _nextMessageId;
 }

src/java/com/threerings/presents/peer/server/PeerManager.java

@@ -861,7 +861,7 @@ public abstract class PeerManager
         PeerNode peer = _peers.get(record.nodeName);
         if (peer == null) {
             _peers.put(record.nodeName, peer = createPeerNode());
-            peer.init(this, _omgr, record);
+            peer.init(this, _omgr, _conmgr, record);
         }
         peer.refresh(record);
     }

src/java/com/threerings/presents/peer/server/PeerNode.java

@@ -27,7 +27,6 @@ import java.util.Date;
 
 import com.samskivert.util.ResultListenerList;
 
-import com.threerings.presents.client.BlockingCommunicator;
 import com.threerings.presents.client.Client;
 import com.threerings.presents.client.ClientObserver;
 import com.threerings.presents.client.Communicator;
@@ -41,6 +40,8 @@ import com.threerings.presents.peer.data.NodeObject;
 import com.threerings.presents.peer.net.PeerBootstrapData;
 import com.threerings.presents.peer.server.persist.NodeRecord;
 import com.threerings.presents.server.PresentsDObjectMgr;
+import com.threerings.presents.server.net.ConnectionManager;
+import com.threerings.presents.server.net.ServerCommunicator;
 
 import static com.threerings.presents.Log.log;
 
@@ -53,14 +54,17 @@ public class PeerNode
     /** This peer's node object. */
     public NodeObject nodeobj;
 
-    public void init (PeerManager peermgr, PresentsDObjectMgr omgr, NodeRecord record)
+    /**
+     * Initializes this peer node and creates its internal client.
+     */
+    public void init (PeerManager peermgr, final PresentsDObjectMgr omgr,
+                      final ConnectionManager conmgr, NodeRecord record)
     {
         _peermgr = peermgr;
         _omgr = omgr;
         _record = record;
         _client = new Client(null, _omgr) {
-            @Override
+            @Override protected void convertFromRemote (DObject target, DEvent event) {
-            protected void convertFromRemote (DObject target, DEvent event) {
                 super.convertFromRemote(target, event);
                 // rewrite the event's target oid using the oid currently configured on the
                 // distributed object (we will have it mapped in our remote server's oid space,
@@ -70,11 +74,8 @@ public class PeerNode
                 // properly deal with it
                 event.eventId = PeerNode.this._omgr.getNextEventId(true);
             }
-            @Override
+            @Override protected Communicator createCommunicator () {
-            protected Communicator createCommunicator () {
+                return new ServerCommunicator(this, conmgr, omgr);
-                // TODO: make a custom communicator that uses the ClientManager NIO system to do
-                // its I/O instead of using two threads and blocking socket I/O
-                return new BlockingCommunicator(this);
             }
         };
         _client.addClientObserver(this);
@@ -102,10 +103,8 @@ public class PeerNode
 
     public void refresh (NodeRecord record)
     {
-        // if the hostname of this node changed, kill our existing client connection and connect
+        // if the hostname of this node changed, kill our existing client and connect anew
-        // anew
+        if (!record.hostName.equals(_record.hostName) && _client.isActive()) {
-        if (!record.hostName.equals(_record.hostName) &&
-            _client.isActive()) {
             _client.logoff(false);
         }
 
@@ -120,8 +119,8 @@ public class PeerNode
         // if our client hasn't updated its record since we last tried to logon, then just
         // chill
         if ((_lastConnectStamp - _record.lastUpdated.getTime()) > STALE_INTERVAL) {
-            log.debug("Not reconnecting to stale client [record=" + _record +
+            log.debug("Not reconnecting to stale client", "record", _record,
-                     ", lastTry=" + new Date(_lastConnectStamp) + "].");
+                      "lastTry", new Date(_lastConnectStamp));
             return;
         }
 

src/java/com/threerings/presents/server/PresentsSession.java

@@ -57,6 +57,7 @@ import com.threerings.presents.net.EventNotification;
 import com.threerings.presents.net.FailureResponse;
 import com.threerings.presents.net.ForwardEventRequest;
 import com.threerings.presents.net.LogoffRequest;
+import com.threerings.presents.net.Message;
 import com.threerings.presents.net.ObjectResponse;
 import com.threerings.presents.net.PingRequest;
 import com.threerings.presents.net.PongResponse;
@@ -65,10 +66,8 @@ import com.threerings.presents.net.ThrottleUpdatedMessage;
 import com.threerings.presents.net.UnsubscribeRequest;
 import com.threerings.presents.net.UnsubscribeResponse;
 import com.threerings.presents.net.UpdateThrottleMessage;
-import com.threerings.presents.net.UpstreamMessage;
 import com.threerings.presents.server.net.Connection;
 import com.threerings.presents.server.net.ConnectionManager;
-import com.threerings.presents.server.net.MessageHandler;
 
 import static com.threerings.presents.Log.log;
 
@@ -84,7 +83,7 @@ import static com.threerings.presents.Log.log;
  * from the conmgr thread and therefore also need not be synchronized.
  */
 public class PresentsSession
-    implements MessageHandler, ClientResolutionListener
+    implements Connection.MessageHandler, ClientResolutionListener
 {
     /** Used by {@link PresentsSession#setUsername} to report success or failure. */
     public static interface UserChangeListener
@@ -396,8 +395,8 @@ public class PresentsSession
         endSession();
     }
 
-    // from interface MessageHandler
+    // from interface Connection.MessageHandler
-    public void handleMessage (UpstreamMessage message)
+    public void handleMessage (Message message)
     {
         _messagesIn++; // count 'em up!
 
@@ -957,16 +956,16 @@ public class PresentsSession
     }
 
     /**
-     * Message dispatchers are used to dispatch each different type of upstream message. We can
+     * Message dispatchers are used to dispatch each different type of message. We can look the
-     * look the dispatcher up in a table and invoke it through an overloaded member which is faster
+     * dispatcher up in a table and invoke it through an overloaded member which is faster (so we
-     * (so we think) than doing a bunch of instanceofs.
+     * think) than doing a bunch of instanceofs.
      */
     protected static interface MessageDispatcher
     {
         /**
          * Dispatch the supplied message for the specified client.
          */
-        void dispatch (PresentsSession client, UpstreamMessage mge);
+        void dispatch (PresentsSession client, Message mge);
     }
 
     /**
@@ -974,7 +973,7 @@ public class PresentsSession
      */
     protected static class SubscribeDispatcher implements MessageDispatcher
     {
-        public void dispatch (PresentsSession client, UpstreamMessage msg)
+        public void dispatch (PresentsSession client, Message msg)
         {
             SubscribeRequest req = (SubscribeRequest)msg;
 //             log.info("Subscribing [client=" + client + ", oid=" + req.getOid() + "].");
@@ -989,7 +988,7 @@ public class PresentsSession
      */
     protected static class UnsubscribeDispatcher implements MessageDispatcher
     {
-        public void dispatch (PresentsSession client, UpstreamMessage msg)
+        public void dispatch (PresentsSession client, Message msg)
         {
             UnsubscribeRequest req = (UnsubscribeRequest)msg;
             int oid = req.getOid();
@@ -1009,7 +1008,7 @@ public class PresentsSession
      */
     protected static class ForwardEventDispatcher implements MessageDispatcher
     {
-        public void dispatch (PresentsSession client, UpstreamMessage msg)
+        public void dispatch (PresentsSession client, Message msg)
         {
             ForwardEventRequest req = (ForwardEventRequest)msg;
             DEvent fevt = req.getEvent();
@@ -1036,7 +1035,7 @@ public class PresentsSession
      */
     protected static class PingDispatcher implements MessageDispatcher
     {
-        public void dispatch (PresentsSession client, UpstreamMessage msg)
+        public void dispatch (PresentsSession client, Message msg)
         {
             // send a pong response using the transport with which the message was received
             PingRequest req = (PingRequest)msg;
@@ -1049,7 +1048,7 @@ public class PresentsSession
      */
     protected static class ThrottleUpdatedDispatcher implements MessageDispatcher
     {
-        public void dispatch (final PresentsSession client, UpstreamMessage msg)
+        public void dispatch (final PresentsSession client, Message msg)
         {
             log.debug("Client ACKed throttle update", "client", client);
             client.throttleUpdated();
@@ -1061,7 +1060,7 @@ public class PresentsSession
      */
     protected static class LogoffDispatcher implements MessageDispatcher
     {
-        public void dispatch (final PresentsSession client, UpstreamMessage msg)
+        public void dispatch (final PresentsSession client, Message msg)
         {
             log.debug("Client requested logoff", "client", client);
             client.safeEndSession();

src/java/com/threerings/presents/server/net/AuthingConnection.java

@@ -21,62 +21,37 @@
 
 package com.threerings.presents.server.net;
 
-import java.io.IOException;
-
-import java.nio.channels.SelectionKey;
-import java.nio.channels.SocketChannel;
-
-import com.samskivert.util.StringUtil;
-
 import com.threerings.presents.net.AuthRequest;
 import com.threerings.presents.net.AuthResponse;
-import com.threerings.presents.net.UpstreamMessage;
+import com.threerings.presents.net.Message;
 
 import static com.threerings.presents.Log.log;
 
 /**
- * The authing connection manages the client connection until
+ * The authing connection manages the client connection until authentication has completed (for
- * authentication has completed (for better or for worse).
+ * better or for worse).
  */
 public class AuthingConnection extends Connection
-    implements MessageHandler
 {
-    /**
+    public AuthingConnection ()
-     * Creates a new authing connection object that will manage the
-     * authentication process for the suppled client socket.
-     */
-    public AuthingConnection (ConnectionManager cmgr, SelectionKey selkey,
-                              SocketChannel channel)
-        throws IOException
     {
-        super(cmgr, selkey, channel, System.currentTimeMillis());
+        setMessageHandler(new MessageHandler() {
-
+            public void handleMessage (Message msg) {
-        // we are our own message handler
+                try {
-        setMessageHandler(this);
+                    // keep a handle on our auth request
+                    _authreq = (AuthRequest)msg;
+                    // post ourselves for processing by the authmgr
+                    _cmgr.authenticateConnection(AuthingConnection.this);
+                } catch (ClassCastException cce) {
+                    log.warning("Received non-authreq message during authentication process",
+                                "conn", AuthingConnection.this, "msg", msg);
+                }
+            }
+        });
     }
 
     /**
-     * Called when a new message has arrived from the client.
+     * Returns a reference to the auth request currently being processed.
-     */
-    public void handleMessage (UpstreamMessage msg)
-    {
-        try {
-            // keep a handle on our auth request
-            _authreq = (AuthRequest)msg;
-
-            // post ourselves for processing by the authmgr
-            _cmgr.authenticateConnection(this);
-
-        } catch (ClassCastException cce) {
-            log.warning("Received non-authreq message during " +
-                        "authentication process [conn=" + this +
-                        ", msg=" + msg + "].");
-        }
-    }
-
-    /**
-     * Returns a reference to the auth request currently being processed
-     * by this authing connection.
      */
     public AuthRequest getAuthRequest ()
     {
@@ -84,8 +59,8 @@ public class AuthingConnection extends Connection
     }
 
     /**
-     * Returns the auth response delivered to the client (only valid after
+     * Returns the auth response delivered to the client (only valid after the auth request has
-     * the auth request has been processed.
+     * been processed.
      */
     public AuthResponse getAuthResponse ()
     {
@@ -93,9 +68,8 @@ public class AuthingConnection extends Connection
     }
 
     /**
-     * Stores a reference to the auth response delivered to this
+     * Stores a reference to the auth response delivered to this connection. This is called by the
-     * connection. This is called by the auth manager after delivering the
+     * auth manager after delivering the auth response to the client.
-     * auth response to the client.
      */
     public void setAuthResponse (AuthResponse authrsp)
     {
@@ -105,8 +79,7 @@ public class AuthingConnection extends Connection
     @Override
     public String toString ()
     {
-        return "[mode=AUTHING, addr=" +
+        return "[mode=AUTHING, addr=" + getInetAddress() + "]";
-            StringUtil.toString(_channel.socket().getInetAddress()) + "]";
     }
 
     protected AuthRequest _authreq;

src/java/com/threerings/presents/server/net/Connection.java

@@ -43,34 +43,39 @@ import com.threerings.io.ObjectOutputStream;
 import com.threerings.io.UnreliableObjectInputStream;
 import com.threerings.io.UnreliableObjectOutputStream;
 
-import com.threerings.presents.net.DownstreamMessage;
+import com.threerings.presents.net.Message;
 import com.threerings.presents.net.PingRequest;
-import com.threerings.presents.net.UpstreamMessage;
 import com.threerings.presents.util.DatagramSequencer;
 
 import static com.threerings.presents.Log.log;
 
 /**
  * The base connection class implements the net event handler interface and processes raw incoming
- * network data into a stream of parsed <code>UpstreamMessage</code> objects. It also provides the
+ * network data into a stream of parsed {@link Message} objects. It also provides the means to send
- * means to send messages to the client and facilities for checking delinquency.
+ * messages to the client and facilities for checking delinquency.
  */
-public abstract class Connection implements NetEventHandler
+public class Connection implements NetEventHandler
 {
+    /** Used with {@link #setMessageHandler}. */
+    public static interface MessageHandler {
+        /** Called when a complete message has been parsed from incoming network data. */
+        void handleMessage (Message message);
+    }
+
+    /** The key used by the NIO code to track this connection. */
+    public SelectionKey selkey;
+
     /**
-     * Constructs a connection object that is associated with the supplied socket.
+     * Initializes a connection object with a socket and related info.
      *
      * @param cmgr The connection manager with which this connection is associated.
-     * @param selkey the key used by the NIO code to track this connection.
      * @param channel The socket channel from which we'll be reading messages.
      * @param createStamp The time at which this connection was created.
      */
-    public Connection (ConnectionManager cmgr, SelectionKey selkey, SocketChannel channel,
+    public void init (ConnectionManager cmgr, SocketChannel channel, long createStamp)
-                       long createStamp)
         throws IOException
     {
         _cmgr = cmgr;
-        _selkey = selkey;
         _channel = channel;
         _lastEvent = createStamp;
         _connectionId = ++_lastConnectionId;
@@ -104,14 +109,6 @@ public abstract class Connection implements NetEventHandler
         return _connectionId;
     }
 
-    /**
-     * Returns the selection key associated with our socket channel.
-     */
-    public SelectionKey getSelectionKey ()
-    {
-        return _selkey;
-    }
-
     /**
      * Returns the non-blocking socket object used to construct this connection.
      */
@@ -178,10 +175,39 @@ public abstract class Connection implements NetEventHandler
     }
 
     /**
-     * Called when there is a failure reading or writing on this connection. We notify the
+     * Queues up a request to have this connection closed by the connection manager once all
+     * messages in its queue have been written to its target.
+     */
+    public void asyncClose ()
+    {
+        _cmgr.postAsyncClose(this);
+    }
+
+    /**
+     * Posts a message for delivery to this connection. The message will be delivered by the conmgr
+     * thread as soon as it gets to it.
+     */
+    public void postMessage (Message msg)
+    {
+        // pass this along to the connection manager
+        _cmgr.postMessage(this, msg);
+    }
+
+    /**
+     * Called when an outgoing socket experiences a connect failure. The connection manager will
+     * have cleaned up the partial registration needed during the connect process, so we are only
+     * responsible for closing our socket.
+     */
+    public void connectFailure (IOException ioe)
+    {
+        closeSocket();
+    }
+
+    /**
+     * Called when there is a failure reading or writing to this connection. We notify the
      * connection manager and close ourselves down.
      */
-    public void handleFailure (IOException ioe)
+    public void networkFailure (IOException ioe)
     {
         // if we're already closed, then something is seriously funny
         if (isClosed()) {
@@ -196,6 +222,135 @@ public abstract class Connection implements NetEventHandler
         closeSocket();
     }
 
+    /**
+     * Processes a datagram sent to this connection.
+     */
+    public void handleDatagram (InetSocketAddress source, ByteBuffer buf, long when)
+    {
+        // lazily create our various bits and bobs
+        if (_digest == null) {
+            try {
+                _digest = MessageDigest.getInstance("MD5");
+            } catch (NoSuchAlgorithmException nsae) {
+                log.warning("Missing MD5 algorithm.");
+                return;
+            }
+            ByteBufferInputStream bin = new ByteBufferInputStream(buf);
+            _sequencer = new DatagramSequencer(
+                new UnreliableObjectInputStream(bin),
+                new UnreliableObjectOutputStream(_cmgr.getFlattener()));
+        }
+
+        // verify the hash
+        buf.position(12);
+        _digest.update(buf);
+        byte[] hash = _digest.digest(_datagramSecret);
+        buf.position(4);
+        for (int ii = 0; ii < 8; ii++) {
+            if (hash[ii] != buf.get()) {
+                log.warning("Datagram failed hash check [connectionId=" + _connectionId +
+                    ", source=" + source + "].");
+                return;
+            }
+        }
+
+        // update our target address
+        _datagramAddress = source;
+
+        // read the contents through the sequencer
+        try {
+            Message msg = _sequencer.readDatagram();
+            if (msg == null) {
+                return; // received out of order
+            }
+            msg.received = when;
+            _handler.handleMessage(msg);
+
+        } catch (ClassNotFoundException cnfe) {
+            log.warning("Error reading datagram [error=" + cnfe + "].");
+
+        } catch (IOException ioe) {
+            log.warning("Error reading datagram [error=" + ioe + "].");
+        }
+    }
+
+    // from interface NetEventHandler
+    public int handleEvent (long when)
+    {
+        // make a note that we received an event as of this time
+        _lastEvent = when;
+
+        int bytesIn = 0;
+        try {
+            // we're lazy about creating our input streams because we may be inheriting them from
+            // our authing connection and we don't want to unnecessarily create them in that case
+            if (_fin == null) {
+                _fin = new FramedInputStream();
+                _oin = new ObjectInputStream(_fin);
+                if (_loader != null) {
+                    _oin.setClassLoader(_loader);
+                }
+            }
+
+            // there may be more than one frame in the buffer, so we keep reading them until we run
+            // out of data
+            while (_fin.readFrame(_channel)) {
+                // make a note of how many bytes are in this frame (including the frame length
+                // bytes which aren't reported in available())
+                bytesIn = _fin.available() + 4;
+                // parse the message and pass it on
+                Message msg = (Message)_oin.readObject();
+                msg.received = when;
+//                 Log.info("Read message " + msg + ".");
+                _handler.handleMessage(msg);
+            }
+
+        } catch (EOFException eofe) {
+            // close down the socket gracefully
+            close();
+
+        } catch (ClassNotFoundException cnfe) {
+            log.warning("Error reading message from socket [channel=" +
+                        StringUtil.safeToString(_channel) + ", error=" + cnfe + "].");
+            // deal with the failure
+            String errmsg = "Unable to decode incoming message.";
+            networkFailure((IOException) new IOException(errmsg).initCause(cnfe));
+
+        } catch (IOException ioe) {
+            // don't log a warning for the ever-popular "the client dropped the connection" failure
+            String msg = ioe.getMessage();
+            if (msg == null || msg.indexOf("reset by peer") == -1) {
+                log.warning("Error reading message from socket [channel=" +
+                            StringUtil.safeToString(_channel) + ", error=" + ioe + "].");
+            }
+            // deal with the failure
+            networkFailure(ioe);
+        }
+
+        return bytesIn;
+    }
+
+    // documentation inherited from interface
+    public boolean checkIdle (long now)
+    {
+        long idleMillis = now - _lastEvent;
+        if (idleMillis < PingRequest.PING_INTERVAL + LATENCY_GRACE) {
+            return false;
+        }
+        if (isClosed()) {
+            return true;
+        }
+        log.info("Disconnecting non-communicative client [conn=" + this +
+                 ", idle=" + idleMillis + "ms].");
+        return true;
+    }
+
+    @Override // from Object
+    public String toString ()
+    {
+        return "[id=" + (hashCode() % 1000) + ", addr=" + getInetAddress() + "]";
+    }
+
     /**
      * Returns the object input stream associated with this connection.  This should only be used
      * by the connection manager.
@@ -270,147 +425,10 @@ public abstract class Connection implements NetEventHandler
         }
 
         // clear out our references to prevent repeat closings
-        _selkey = null;
         _channel = null;
     }
 
-    /**
-     * Called when our client socket has data available for reading.
-     */
-    public int handleEvent (long when)
-    {
-        // make a note that we received an event as of this time
-        _lastEvent = when;
-
-        int bytesIn = 0;
-        try {
-            // we're lazy about creating our input streams because we may be inheriting them from
-            // our authing connection and we don't want to unnecessarily create them in that case
-            if (_fin == null) {
-                _fin = new FramedInputStream();
-                _oin = new ObjectInputStream(_fin);
-                if (_loader != null) {
-                    _oin.setClassLoader(_loader);
-                }
-            }
-
-            // there may be more than one frame in the buffer, so we keep reading them until we run
-            // out of data
-            while (_fin.readFrame(_channel)) {
-                // make a note of how many bytes are in this frame (including the frame length
-                // bytes which aren't reported in available())
-                bytesIn = _fin.available() + 4;
-                // parse the message and pass it on
-                UpstreamMessage msg = (UpstreamMessage)_oin.readObject();
-                msg.received = when;
-//                 Log.info("Read message " + msg + ".");
-                _handler.handleMessage(msg);
-            }
-
-        } catch (EOFException eofe) {
-            // close down the socket gracefully
-            close();
-
-        } catch (ClassNotFoundException cnfe) {
-            log.warning("Error reading message from socket [channel=" +
-                        StringUtil.safeToString(_channel) + ", error=" + cnfe + "].");
-            // deal with the failure
-            String errmsg = "Unable to decode incoming message.";
-            handleFailure((IOException) new IOException(errmsg).initCause(cnfe));
-
-        } catch (IOException ioe) {
-            // don't log a warning for the ever-popular "the client dropped the connection" failure
-            String msg = ioe.getMessage();
-            if (msg == null || msg.indexOf("reset by peer") == -1) {
-                log.warning("Error reading message from socket [channel=" +
-                            StringUtil.safeToString(_channel) + ", error=" + ioe + "].");
-            }
-            // deal with the failure
-            handleFailure(ioe);
-        }
-
-        return bytesIn;
-    }
-
-    /**
-     * Processes a datagram sent to this connection.
-     */
-    public void handleDatagram (InetSocketAddress source, ByteBuffer buf, long when)
-    {
-        // lazily create our various bits and bobs
-        if (_digest == null) {
-            try {
-                _digest = MessageDigest.getInstance("MD5");
-            } catch (NoSuchAlgorithmException nsae) {
-                log.warning("Missing MD5 algorithm.");
-                return;
-            }
-            ByteBufferInputStream bin = new ByteBufferInputStream(buf);
-            _sequencer = new DatagramSequencer(
-                new UnreliableObjectInputStream(bin),
-                new UnreliableObjectOutputStream(_cmgr.getFlattener()));
-        }
-
-        // verify the hash
-        buf.position(12);
-        _digest.update(buf);
-        byte[] hash = _digest.digest(_datagramSecret);
-        buf.position(4);
-        for (int ii = 0; ii < 8; ii++) {
-            if (hash[ii] != buf.get()) {
-                log.warning("Datagram failed hash check [connectionId=" + _connectionId +
-                    ", source=" + source + "].");
-                return;
-            }
-        }
-
-        // update our target address
-        _datagramAddress = source;
-
-        // read the contents through the sequencer
-        try {
-            UpstreamMessage msg = (UpstreamMessage)_sequencer.readDatagram();
-            if (msg == null) {
-                return; // received out of order
-            }
-            msg.received = when;
-            _handler.handleMessage(msg);
-
-        } catch (ClassNotFoundException cnfe) {
-            log.warning("Error reading datagram [error=" + cnfe + "].");
-
-        } catch (IOException ioe) {
-            log.warning("Error reading datagram [error=" + ioe + "].");
-        }
-    }
-
-    // documentation inherited from interface
-    public boolean checkIdle (long now)
-    {
-        long idleMillis = now - _lastEvent;
-        if (idleMillis < PingRequest.PING_INTERVAL + LATENCY_GRACE) {
-            return false;
-        }
-        if (isClosed()) {
-            return true;
-        }
-        log.info("Disconnecting non-communicative client [conn=" + this +
-                 ", idle=" + idleMillis + "ms].");
-        return true;
-    }
-
-    /**
-     * Posts a downstream message for delivery to this connection. The message will be delivered by
-     * the conmgr thread as soon as it gets to it.
-     */
-    public void postMessage (DownstreamMessage msg)
-    {
-        // pass this along to the connection manager
-        _cmgr.postMessage(this, msg);
-    }
-
     protected ConnectionManager _cmgr;
-    protected SelectionKey _selkey;
     protected SocketChannel _channel;
 
     protected long _lastEvent;

src/java/com/threerings/presents/server/net/ConnectionManager.java

@@ -62,7 +62,7 @@ import com.threerings.presents.client.Client;
 import com.threerings.presents.data.ConMgrStats;
 import com.threerings.presents.net.AuthRequest;
 import com.threerings.presents.net.AuthResponse;
-import com.threerings.presents.net.DownstreamMessage;
+import com.threerings.presents.net.Message;
 import com.threerings.presents.server.Authenticator;
 import com.threerings.presents.server.ChainedAuthenticator;
 import com.threerings.presents.server.DummyAuthenticator;
@@ -193,6 +193,74 @@ public class ConnectionManager extends LoopingThread
         }
     }
 
+    /**
+     * Opens an outgoing connection to the supplied address. The connection will be opened in a
+     * non-blocking manner and added to the connection manager's select set. Messages posted to the
+     * connection prior to it being actually connected to its destination will remain in the queue.
+     * If the connection fails those messages will be dropped.
+     *
+     * @param conn the connection to be initialized and opened. Callers may want to provide a
+     * {@link Connection} derived class so that they may intercept calldown methods.
+     * @param hostname the hostname of the server to which to connect.
+     * @param port the port on which to connect to the server.
+     *
+     * @exception IOException thrown if an error occurs opening a connection to the supplied
+     * server. When this method returns the connection may or may not be actually connected to the
+     * server. If the asynchronous connection attempt fails, the Connection will be notified via
+     * {@link Connection#networkFailure}.
+     */
+    public void openOutgoingConnection (final Connection conn, String hostname, int port)
+        throws IOException
+    {
+        // create a non-blocking socket channel to use for this connection
+        final SocketChannel sockchan = SocketChannel.open();
+        sockchan.configureBlocking(false);
+
+        // create our connection instance
+        conn.init(this, sockchan, System.currentTimeMillis());
+        // and register our channel with the selector (if this fails, we abandon ship immediately)
+        conn.selkey = sockchan.register(_selector, SelectionKey.OP_CONNECT);
+
+        // start our connection process (now if we fail we need to clean things up)
+        try {
+            NetEventHandler handler;
+            if (sockchan.connect(new InetSocketAddress(hostname, port))) {
+                // it is possible even for a non-blocking socket to connect immediately, in which
+                // case we stick the connection in as its event handler immediately
+                handler = conn;
+
+            } else {
+                // otherwise we wire up a special event handler that will wait for our socket to
+                // finish the connection process and then wire things up fully
+                handler = new NetEventHandler() {
+                    public int handleEvent (long when) {
+                        try {
+                            if (sockchan.finishConnect()) {
+                                // great, we're ready to roll, wire up the connection
+                                conn.selkey = sockchan.register(_selector, SelectionKey.OP_READ);
+                                _handlers.put(conn.selkey, conn);
+                            }
+                        } catch (IOException ioe) {
+                            _handlers.remove(conn.selkey);
+                            _oflowqs.remove(conn);
+                            conn.connectFailure(ioe);
+                        }
+                        return 0;
+                    }
+                    public boolean checkIdle (long now) {
+                        return conn.checkIdle(now);
+                    }
+                };
+            }
+            _handlers.put(conn.selkey, handler);
+
+        } catch (IOException ioe) {
+            log.warning("Failed to initiate connection for " + sockchan + ".", ioe);
+            conn.connectFailure(ioe); // nothing else to clean up
+            throw ioe;
+        }
+    }
+
     /**
      * Queues a connection up to be closed on the conmgr thread.
      */
@@ -201,25 +269,8 @@ public class ConnectionManager extends LoopingThread
         _deathq.append(conn);
     }
 
-    /**
+    // from interface ReportManager.Reporter
-     * Performs the authentication process on the specified connection. This is called by {@link
+    public void appendReport (StringBuilder report, long now, long sinceLast, boolean reset)
-     * AuthingConnection} itself once it receives its auth request.
-     */
-    public void authenticateConnection (AuthingConnection conn)
-    {
-        _author.authenticateConnection(_authInvoker, conn, new ResultListener<AuthingConnection>() {
-            public void requestCompleted (AuthingConnection conn) {
-                _authq.append(conn);
-            }
-            public void requestFailed (Exception cause) {
-                // this never happens
-            }
-        });
-    }
-
-    // documentation inherited from interface ReportManager.Reporter
-    public void appendReport (
-        StringBuilder report, long now, long sinceLast, boolean reset)
     {
         ConMgrStats stats = getStats();
         int connects = stats.connects - _lastStats.connects;
@@ -263,6 +314,22 @@ public class ConnectionManager extends LoopingThread
         return super.isRunning() || _omgr.isRunning();
     }
 
+    /**
+     * Performs the authentication process on the specified connection. This is called by {@link
+     * AuthingConnection} itself once it receives its auth request.
+     */
+    protected void authenticateConnection (AuthingConnection conn)
+    {
+        _author.authenticateConnection(_authInvoker, conn, new ResultListener<AuthingConnection>() {
+            public void requestCompleted (AuthingConnection conn) {
+                _authq.append(conn);
+            }
+            public void requestFailed (Exception cause) {
+                // this never happens
+            }
+        });
+    }
+
     /**
      * Notifies the connection observers of a connection event. Used internally.
      */
@@ -363,10 +430,6 @@ public class ConnectionManager extends LoopingThread
             }
         }
 
-        // we'll use these for sending messages to clients
-        _framer = new FramingOutputStream();
-        _flattener = new ByteArrayOutputStream();
-
         // notify our startup listener, if we have one
         if (_startlist != null) {
             _startlist.requestCompleted(null);
@@ -461,21 +524,20 @@ public class ConnectionManager extends LoopingThread
             try {
                 // construct a new running connection to handle this connections network traffic
                 // from here on out
-                SelectionKey selkey = conn.getSelectionKey();
+                Connection rconn = new Connection();
-                RunningConnection rconn = new RunningConnection(
+                rconn.init(this, conn.getChannel(), iterStamp);
-                    this, selkey, conn.getChannel(), iterStamp);
+                rconn.selkey = conn.selkey;
 
                 // we need to keep using the same object input and output streams from the
                 // beginning of the session because they have context that needs to be preserved
                 rconn.inheritStreams(conn);
 
                 // replace the mapping in the handlers table from the old conn with the new one
-                _handlers.put(selkey, rconn);
+                _handlers.put(rconn.selkey, rconn);
 
                 // add a mapping for the connection id and set the datagram secret
                 _connections.put(rconn.getConnectionId(), rconn);
-                rconn.setDatagramSecret(
+                rconn.setDatagramSecret(conn.getAuthRequest().getCredentials().getDatagramSecret());
-                    conn.getAuthRequest().getCredentials().getDatagramSecret());
 
                 // transfer any overflow queue for that connection
                 OverflowQueue oflowHandler = _oflowqs.remove(conn);
@@ -494,9 +556,10 @@ public class ConnectionManager extends LoopingThread
 
         Set<SelectionKey> ready = null;
         try {
-            // check for incoming network events
 //             log.debug("Selecting from " + StringUtil.toString(_selector.keys()) + " (" +
 //                       SELECT_LOOP_TIME + ").");
+
+            // check for incoming network events
             int ecount = _selector.select(SELECT_LOOP_TIME);
             ready = _selector.selectedKeys();
             if (ecount == 0) {
@@ -538,8 +601,7 @@ public class ConnectionManager extends LoopingThread
             try {
                 handler = _handlers.get(selkey);
                 if (handler == null) {
-                    log.warning("Received network event but have no registered handler " +
+                    log.warning("Received network event for unknown handler", "key", selkey);
-                                "[selkey=" + selkey + "].");
                     // request that this key be removed from our selection set, which normally
                     // happens automatically but for some reason didn't
                     selkey.cancel();
@@ -592,7 +654,7 @@ public class ConnectionManager extends LoopingThread
                     }
 
                 } catch (IOException ioe) {
-                    oq.conn.handleFailure(ioe);
+                    oq.conn.networkFailure(ioe);
                 }
             }
         }
@@ -641,10 +703,15 @@ public class ConnectionManager extends LoopingThread
             return true;
         }
 
+        // if this is an asynchronous close request, queue the connection up for death
+        if (data == ASYNC_CLOSE_REQUEST) {
+            closeConnection(conn);
+            return true;
+        }
+
         // sanity check the message size
         if (data.length > 1024 * 1024) {
-            log.warning("Refusing to write absurdly large message [conn=" + conn +
+            log.warning("Refusing to write very large message", "conn", conn, "size", data.length);
-                        ", size=" + data.length + "].");
             return true;
         }
 
@@ -664,26 +731,26 @@ public class ConnectionManager extends LoopingThread
             _outbuf.put(data);
             _outbuf.flip();
 
+            // if the connection to which we're writing is not yet ready, the whole message is
+            // "leftover", so we pass it to the partial write handler
+            SocketChannel sochan = conn.getChannel();
+            if (!sochan.isConnected()) {
+                pwh.handlePartialWrite(conn, _outbuf);
+                return false;
+            }
+
             // then write the data to the socket
-            int wrote = conn.getChannel().write(_outbuf);
+            int wrote = sochan.write(_outbuf);
             noteWrite(1, wrote);
 
+            // if we didn't write our entire message, deal with the leftover bytes
             if (_outbuf.remaining() > 0) {
                 fully = false;
-//                 log.info("Partial write [conn=" + conn +
-//                          ", msg=" + StringUtil.shortClassName(outmsg) + ", wrote=" + wrote +
-//                          ", size=" + buffer.limit() + "].");
                 pwh.handlePartialWrite(conn, _outbuf);
-
-//                 } else if (wrote > 10000) {
-//                     log.info("Big write [conn=" + conn +
-//                              ", msg=" + StringUtil.shortClassName(outmsg) +
-//                              ", wrote=" + wrote + "].");
             }
 
         } catch (IOException ioe) {
-            // instruct the connection to deal with its failure
+            conn.networkFailure(ioe); // instruct the connection to deal with its failure
-            conn.handleFailure(ioe);
 
         } finally {
             _outbuf.clear();
@@ -790,8 +857,10 @@ public class ConnectionManager extends LoopingThread
             // connection and register it with our selection set
             SelectableChannel selchan = channel;
             selchan.configureBlocking(false);
-            SelectionKey selkey = selchan.register(_selector, SelectionKey.OP_READ);
+            AuthingConnection aconn = new AuthingConnection();
-            _handlers.put(selkey, new AuthingConnection(this, selkey, channel));
+            aconn.selkey = selchan.register(_selector, SelectionKey.OP_READ);
+            aconn.init(this, channel, System.currentTimeMillis());
+            _handlers.put(aconn.selkey, aconn);
             synchronized (this) {
                 _stats.connects++;
             }
@@ -863,11 +932,11 @@ public class ConnectionManager extends LoopingThread
      * which happens when forwarding an event to a client and at the completion of authentication,
      * both of which <em>must</em> happen only on the distributed object thread.
      */
-    void postMessage (Connection conn, DownstreamMessage msg)
+    void postMessage (Connection conn, Message msg)
     {
         if (!isRunning()) {
-            log.warning(
+            log.warning("Posting message to inactive connection manager",
-                "Posting message to inactive connection manager", "msg", msg, new Exception());
+                        "msg", msg, new Exception());
         }
 
         // sanity check
@@ -901,22 +970,20 @@ public class ConnectionManager extends LoopingThread
             ByteBuffer buffer = _framer.frameAndReturnBuffer();
             byte[] data = new byte[buffer.limit()];
             buffer.get(data);
-
+            // log.info("Flattened " + msg + " into " + data.length + " bytes.");
-//             log.info("Flattened " + msg + " into " + data.length + " bytes.");
 
             // and slap both on the queue
-            _outq.append(new Tuple<Connection, byte[]>(conn, data));
+            _outq.append(Tuple.create(conn, data));
 
         } catch (Exception e) {
-            log.warning("Failure flattening message [conn=" + conn +
+            log.warning("Failure flattening message", "conn", conn, "msg", msg, e);
-                    ", msg=" + msg + "].", e);
         }
     }
 
     /**
      * Helper function for {@link #postMessage}; handles posting the message as a datagram.
      */
-    void postDatagram (Connection conn, DownstreamMessage msg)
+    void postDatagram (Connection conn, Message msg)
         throws Exception
     {
         _flattener.reset();
@@ -929,7 +996,20 @@ public class ConnectionManager extends LoopingThread
         byte[] data = _flattener.toByteArray();
 
         // slap it on the queue
-        _dataq.append(new Tuple<Connection, byte[]>(conn, data));
+        _dataq.append(Tuple.create(conn, data));
+    }
+
+    /**
+     * Posts a fake message to this connection's outgoing message queue that will cause the
+     * connection to be closed when this message is reached. This is only used by outgoing
+     * connections to ensure that they finish sending their queued outgoing messages before closing
+     * their connection. Incoming connections tend only to be closed at the request of the client
+     * or in case of delinquincy. In neither circumstance do we need to flush the client's outgoing
+     * queue before closing.
+     */
+    void postAsyncClose (Connection conn)
+    {
+        _outq.append(Tuple.create(conn, ASYNC_CLOSE_REQUEST));
     }
 
     /**
@@ -939,7 +1019,7 @@ public class ConnectionManager extends LoopingThread
     {
         // remove this connection from our mappings (it is automatically removed from the Selector
         // when the socket is closed)
-        _handlers.remove(conn.getSelectionKey());
+        _handlers.remove(conn.selkey);
         _connections.remove(conn.getConnectionId());
         _oflowqs.remove(conn);
         synchronized (this) {
@@ -957,7 +1037,7 @@ public class ConnectionManager extends LoopingThread
     {
         // remove this connection from our mappings (it is automatically removed from the Selector
         // when the socket is closed)
-        _handlers.remove(conn.getSelectionKey());
+        _handlers.remove(conn.selkey);
         _connections.remove(conn.getConnectionId());
         _oflowqs.remove(conn);
 
@@ -1012,8 +1092,14 @@ public class ConnectionManager extends LoopingThread
         {
             // write any partial message if we have one
             if (_partial != null) {
+                // if our outgoing channel is still not ready, then bail immediately
+                SocketChannel sochan = conn.getChannel();
+                if (!sochan.isConnected()) {
+                    return false;
+                }
+
                 // write all we can of our partial buffer
-                int wrote = conn.getChannel().write(_partial);
+                int wrote = sochan.write(_partial);
                 noteWrite(0, wrote);
 
                 if (_partial.remaining() == 0) {
@@ -1095,8 +1181,8 @@ public class ConnectionManager extends LoopingThread
 
     protected Queue<Tuple<Connection, byte[]>> _outq = new Queue<Tuple<Connection, byte[]>>();
     protected Queue<Tuple<Connection, byte[]>> _dataq = new Queue<Tuple<Connection, byte[]>>();
-    protected FramingOutputStream _framer;
+    protected FramingOutputStream _framer = new FramingOutputStream();
-    protected ByteArrayOutputStream _flattener;
+    protected ByteArrayOutputStream _flattener = new ByteArrayOutputStream();
     protected ByteBuffer _outbuf = ByteBuffer.allocateDirect(64 * 1024);
     protected ByteBuffer _databuf = ByteBuffer.allocateDirect(Client.MAX_DATAGRAM_SIZE);
 
@@ -1135,4 +1221,7 @@ public class ConnectionManager extends LoopingThread
     protected static final int CONNECTION_ESTABLISHED = 0;
     protected static final int CONNECTION_FAILED = 1;
     protected static final int CONNECTION_CLOSED = 2;
+
+    /** Used to denote asynchronous close requests. */
+    protected static final byte[] ASYNC_CLOSE_REQUEST = new byte[0];
 }

src/java/com/threerings/presents/server/net/RunningConnection.java

@@ -1,61 +0,0 @@
-//
-// $Id$
-//
-// Narya library - tools for developing networked games
-// Copyright (C) 2002-2007 Three Rings Design, Inc., All Rights Reserved
-// http://www.threerings.net/code/narya/
-//
-// This library is free software; you can redistribute it and/or modify it
-// under the terms of the GNU Lesser General Public License as published
-// by the Free Software Foundation; either version 2.1 of the License, or
-// (at your option) any later version.
-//
-// This library is distributed in the hope that it will be useful,
-// but WITHOUT ANY WARRANTY; without even the implied warranty of
-// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
-// Lesser General Public License for more details.
-//
-// You should have received a copy of the GNU Lesser General Public
-// License along with this library; if not, write to the Free Software
-// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
-
-package com.threerings.presents.server.net;
-
-import java.io.IOException;
-
-import java.nio.channels.SelectionKey;
-import java.nio.channels.SocketChannel;
-
-import com.threerings.presents.net.UpstreamMessage;
-
-/**
- * A running connection object represents a fully operational client
- * connection to the server.
- */
-public class RunningConnection extends Connection
-{
-    /**
-     * Constructs a new running connection object to manage the supplied
-     * client socket.
-     */
-    public RunningConnection (ConnectionManager cmgr, SelectionKey selkey,
-                              SocketChannel channel, long createStamp)
-        throws IOException
-    {
-        super(cmgr, selkey, channel, createStamp);
-    }
-
-    /**
-     * Called when a new message has arrived from the client.
-     */
-    public void handleMessage (UpstreamMessage msg)
-    {
-    }
-
-    @Override
-    public String toString ()
-    {
-        return "[mode=RUNNING, id=" + (hashCode() % 1000) +
-            ", addr=" + getInetAddress() + "]";
-    }
-}

src/java/com/threerings/presents/server/net/ServerCommunicator.java

@@ -0,0 +1,196 @@
+//
+// $Id$
+//
+// Narya library - tools for developing networked games
+// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved
+// http://www.threerings.net/code/narya/
+//
+// This library is free software; you can redistribute it and/or modify it
+// under the terms of the GNU Lesser General Public License as published
+// by the Free Software Foundation; either version 2.1 of the License, or
+// (at your option) any later version.
+//
+// This library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+// Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public
+// License along with this library; if not, write to the Free Software
+// Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+package com.threerings.presents.server.net;
+
+import java.io.IOException;
+
+import com.threerings.presents.client.Client;
+import com.threerings.presents.client.ClientDObjectMgr;
+import com.threerings.presents.client.ClientObserver;
+import com.threerings.presents.client.Communicator;
+import com.threerings.presents.client.ObserverOps;
+import com.threerings.presents.client.SessionObserver;
+import com.threerings.presents.dobj.RootDObjectManager;
+import com.threerings.presents.net.AuthRequest;
+import com.threerings.presents.net.AuthResponse;
+import com.threerings.presents.net.AuthResponseData;
+import com.threerings.presents.net.LogoffRequest;
+import com.threerings.presents.net.Message;
+import com.threerings.presents.net.UpstreamMessage;
+
+import static com.threerings.presents.Log.log;
+
+/**
+ * Provides Client {@link Communicator} services using non-blocking I/O and the connection manager.
+ */
+public class ServerCommunicator extends Communicator
+{
+    public ServerCommunicator (Client client, ConnectionManager conmgr, RootDObjectManager rootmgr)
+    {
+        super(client);
+        _conmgr = conmgr;
+        _rootmgr = rootmgr;
+        _omgr = new ClientDObjectMgr(this, client);
+    }
+
+    @Override // from Communicator
+    public void logon ()
+    {
+        // make sure things are copacetic
+        if (_conn != null) {
+            throw new RuntimeException("Communicator already started.");
+        }
+
+        // we assume server entities have no firewall issues and can connect on the first port
+        try {
+            Connection conn = new Connection() {
+                @Override public void connectFailure (IOException ioe) {
+                    _logonError = ioe; // report this as a logon failure
+                    super.connectFailure(ioe);
+                }
+
+                @Override public void networkFailure (final IOException ioe) {
+                    notifyClientObservers(new ObserverOps.Client() {
+                        protected void notify (ClientObserver obs) {
+                            obs.clientConnectionFailed(_client, ioe);
+                        }
+                    });
+                    super.networkFailure(ioe);
+                }
+
+                @Override protected void closeSocket () {
+                    super.closeSocket();
+                    shutdown();
+                }
+            };
+            conn.setMessageHandler(new Connection.MessageHandler() {
+                public void handleMessage (Message message) {
+                    try {
+                        // our first message will always be an auth response
+                        gotAuthResponse((AuthResponse)message);
+                    } catch (Exception e) {
+                        _logonError = e;
+                        shutdown();
+                    }
+                }
+            });
+            _conmgr.openOutgoingConnection(conn, _client.getHostname(), _client.getPorts()[0]);
+            _conn = conn;
+            if (_loader != null) {
+                _conn.setClassLoader(_loader);
+            }
+
+            // now send our auth request
+            postMessage(new AuthRequest(_client.getCredentials(), _client.getVersion(),
+                                        _client.getBootGroups()));
+
+        } catch (IOException ioe) {
+            _logonError = ioe;
+            shutdown();
+        }
+    }
+
+    @Override // from Communicator
+    public void logoff ()
+    {
+        if (_conn != null) {
+            _conn.postMessage(new LogoffRequest());
+            _conn.asyncClose();
+            _conn = null;
+        }
+    }
+
+    @Override // from Communicator
+    public void gotBootstrap ()
+    {
+        // nothing needed
+    }
+
+    @Override // from Communicator
+    public void postMessage (final UpstreamMessage msg)
+    {
+        // if we're not on the main dobjmgr thread, we need to get there
+        if (!_rootmgr.isDispatchThread()) {
+            _rootmgr.postRunnable(new Runnable() {
+                public void run () {
+                    postMessage(msg);
+                }
+            });
+            return;
+        }
+
+        if (_conn == null) {
+            log.info("Dropping message for lack of connection.", "client", _client, "msg", msg);
+            return;
+        }
+
+        // pass this message along to our connection
+        _conn.postMessage(msg);
+
+        // we cheat a bit and claim that we "wrote" when we post our messages so that we don't have
+        // to modify the connection manager to call a method on Connection every time it writes a
+        // message from the queue
+        updateWriteStamp();
+    }
+
+    @Override // from Communicator
+    public void setClassLoader (ClassLoader loader)
+    {
+        _loader = loader;
+        if (_conn != null) {
+            _conn.setClassLoader(loader);
+        }
+    }
+
+    @Override // from Communicator
+    protected synchronized void logonSucceeded (AuthResponseData data)
+    {
+        super.logonSucceeded(data);
+
+        // now we can route all messages to the ClientDObjectMgr
+        _conn.setMessageHandler(new Connection.MessageHandler() {
+            public void handleMessage (Message message) {
+                _omgr.processMessage(message);
+            }
+        });
+    }
+
+    protected void shutdown ()
+    {
+        if (_logonError == null) {
+            // we were logged on successfully, so report didLogoff first
+            notifyClientObservers(new ObserverOps.Session() {
+                protected void notify (SessionObserver obs) {
+                    obs.clientDidLogoff(_client);
+                }
+            });
+        }
+        clientCleanup(_logonError);
+    }
+
+    protected ConnectionManager _conmgr;
+    protected RootDObjectManager _rootmgr;
+    protected Connection _conn;
+    protected ClientDObjectMgr _omgr;
+    protected ClassLoader _loader;
+    protected Exception _logonError;
+}