Rewritten to use the Java NIO library rather than the Berkeley NBIO

library. 100% pure baby! At least for the moment.


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

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

@@ -1,16 +1,15 @@
 //
-// $Id: Communicator.java,v 1.24 2002/10/31 18:44:34 mdb Exp $
+// $Id: Communicator.java,v 1.25 2002/11/18 18:53:10 mdb Exp $
 
 package com.threerings.presents.client;
 
 import java.io.EOFException;
 import java.io.IOException;
-import java.io.InputStream;
 import java.io.InterruptedIOException;
-import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
 
 import java.net.InetAddress;
-import java.net.Socket;
 
 import com.samskivert.io.NestableIOException;
 import com.samskivert.util.Interval;
@@ -26,6 +25,7 @@ import com.threerings.io.ObjectOutputStream;
 import com.threerings.presents.Log;
 import com.threerings.presents.dobj.DObjectManager;
 import com.threerings.presents.net.*;
+import java.net.InetSocketAddress;
 
 /**
  * The client performs all network I/O on separate threads (one for
@@ -106,7 +106,7 @@ public class Communicator
     {
         // if our socket is already closed, we've already taken care of
         // this business
-        if (_socket == null) {
+        if (_channel == null) {
             return;
         }
 
@@ -186,7 +186,7 @@ public class Communicator
     {
         // make sure the socket isn't already closed down (meaning we've
         // already dealt with the failed connection)
-        if (_socket == null) {
+        if (_channel == null) {
             return;
         }
 
@@ -208,7 +208,7 @@ public class Communicator
     {
         // make sure the socket isn't already closed down (meaning we've
         // already dealt with the closed connection)
-        if (_socket == null) {
+        if (_channel == null) {
             return;
         }
 
@@ -246,11 +246,11 @@ public class Communicator
         // our socket and let the client know that the logoff process has
         // completed
         try {
-            _socket.close();
+            _channel.close();
-        } catch (IOException cle) {
+        } catch (IOException ioe) {
-            Log.warning("Error closing failed socket: " + cle);
+            Log.warning("Error closing failed socket: " + ioe);
         }
-        _socket = null;
+        _channel = null;
 
         // clear these out because they are probably large and in charge
         _oin = null;
@@ -278,7 +278,20 @@ public class Communicator
         _oout.flush();
 
         // then write the framed message to actual output stream
-        _fout.writeFrameAndReset(_out);
+        try {
+            ByteBuffer buffer = _fout.frameAndReturnBuffer();
+            int wrote = _channel.write(buffer);
+            if (wrote != buffer.limit()) {
+                Log.warning("Aiya! Couldn't write entire message [msg=" + msg +
+                            ", size=" + buffer.limit() +
+                            ", wrote=" + wrote + "].");
+//             } else {
+//                 Log.info("Wrote " + wrote + " bytes.");
+            }
+
+        } finally {
+            _fout.resetFrame();
+        }
 
         // make a note of our most recent write time
         updateWriteStamp();
@@ -315,7 +328,7 @@ public class Communicator
         // which case we simply call it again because we can't do anything
         // until it has a whole frame; it will throw an exception if it
         // hits EOF or if something goes awry)
-        while (!_fin.readFrame(_in));
+        while (!_fin.readFrame(_channel));
 
         try {
             DownstreamMessage msg = (DownstreamMessage)_oin.readObject();
@@ -359,6 +372,7 @@ public class Communicator
 
             } catch (Exception e) {
                 Log.debug("Logon failed: " + e);
+                Log.logStackTrace(e);
                 // let the observers know that we've failed
                 _client.notifyObservers(Client.CLIENT_FAILED_TO_LOGON, e);
                 // and terminate our communicator thread
@@ -370,7 +384,7 @@ public class Communicator
             throws IOException
         {
             // if we're already connected, we freak out
-            if (_socket != null) {
+            if (_channel != null) {
                 throw new IOException("Already connected.");
             }
 
@@ -379,13 +393,9 @@ public class Communicator
             int port = _client.getPort();
 
             // establish a socket connection to said server
-            Log.debug("Connecting to server [host=" + host +
+            Log.debug("Connecting [host=" + host + ", port=" + port + "].");
-                      ", port=" + port + "].");
+            _channel = SocketChannel.open(new InetSocketAddress(host, port));
-            _socket = new Socket(host, port);
+            _channel.configureBlocking(true);
-
-            // get a handle on our input and output streams
-            _in = _socket.getInputStream();
-            _out = _socket.getOutputStream();
 
             // our messages are framed (preceded by their length), so we
             // use these helper streams to manage the framing
@@ -477,7 +487,7 @@ public class Communicator
             // we want to interrupt the reader thread as it may be blocked
             // listening to the socket; this is only called if the reader
             // thread doesn't shut itself down
-            interrupt();
+//             interrupt();
         }
     }
 
@@ -541,9 +551,7 @@ public class Communicator
     protected Reader _reader;
     protected Writer _writer;
 
-    protected Socket _socket;
+    protected SocketChannel _channel;
-    protected InputStream _in;
-    protected OutputStream _out;
     protected Queue _msgq = new Queue();
 
     protected int _piid;

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

@@ -1,10 +1,12 @@
 //
-// $Id: AuthingConnection.java,v 1.8 2002/10/29 23:51:26 mdb Exp $
+// $Id: AuthingConnection.java,v 1.9 2002/11/18 18:53:10 mdb Exp $
 
 package com.threerings.presents.server.net;
 
 import java.io.IOException;
-import ninja2.core.io_core.nbio.NonblockingSocket;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.SocketChannel;
+
 import com.samskivert.util.StringUtil;
 
 import com.threerings.presents.Log;
@@ -23,11 +25,11 @@ public class AuthingConnection extends Connection
      * Creates a new authing connection object that will manage the
      * authentication process for the suppled client socket.
      */
-    public AuthingConnection (ConnectionManager cmgr,
+    public AuthingConnection (ConnectionManager cmgr, SelectionKey selkey,
-                              NonblockingSocket socket)
+                              SocketChannel channel)
         throws IOException
     {
-        super(cmgr, socket, System.currentTimeMillis());
+        super(cmgr, selkey, channel, System.currentTimeMillis());
 
         // we are our own message handler
         setMessageHandler(this);
@@ -80,10 +82,13 @@ public class AuthingConnection extends Connection
         _authrsp = authrsp;
     }
 
+    /**
+     * Generates a string representation of this instance.
+     */
     public String toString ()
     {
         return "[mode=AUTHING, addr=" +
-            StringUtil.toString(_socket.getInetAddress()) + "]";
+            StringUtil.toString(_channel.socket().getInetAddress()) + "]";
     }
 
     protected AuthRequest _authreq;

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

@@ -1,5 +1,5 @@
 //
-// $Id: Connection.java,v 1.12 2002/11/06 21:05:13 shaper Exp $
+// $Id: Connection.java,v 1.13 2002/11/18 18:53:10 mdb Exp $
 
 package com.threerings.presents.server.net;
 
@@ -7,9 +7,11 @@ import java.io.EOFException;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
-import java.net.SocketTimeoutException;
 
-import ninja2.core.io_core.nbio.*;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.SocketChannel;
 
 import com.samskivert.io.NestableIOException;
 
@@ -37,24 +39,20 @@ public abstract class Connection implements NetEventHandler
      *
      * @param cmgr The connection manager with which this connection is
      * associated.
-     * @param socket The socket from which we'll be reading messages.
+     * @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, NonblockingSocket socket,
+    public Connection (ConnectionManager cmgr, SelectionKey selkey,
-                       long createStamp)
+                       SocketChannel channel, long createStamp)
         throws IOException
     {
         _cmgr = cmgr;
-        _socket = socket;
+        _selkey = selkey;
+        _channel = channel;
         _lastEvent = createStamp;
-
-        // create a select item that will allow us to be incorporated into
-        // the main event polling loop
-        _selitem = new SelectItem(_socket, this, Selectable.READ_READY);
-
-        // get a handle on our socket streams
-        _in = _socket.getInputStream();
-        _out = _socket.getOutputStream();
     }
 
     /**
@@ -68,20 +66,20 @@ public abstract class Connection implements NetEventHandler
     }
 
     /**
-     * Returns the select item associated with this connection.
+     * Returns the selection key associated with our socket channel.
      */
-    public SelectItem getSelectItem ()
+    public SelectionKey getSelectionKey ()
     {
-        return _selitem;
+        return _selkey;
     }
 
     /**
      * Returns the non-blocking socket object used to construct this
      * connection.
      */
-    public NonblockingSocket getSocket ()
+    public SocketChannel getChannel ()
     {
-        return _socket;
+        return _channel;
     }
 
     /**
@@ -89,7 +87,7 @@ public abstract class Connection implements NetEventHandler
      */
     public boolean isClosed ()
     {
-        return (_socket == null);
+        return (_channel == null);
     }
 
     /**
@@ -133,15 +131,6 @@ public abstract class Connection implements NetEventHandler
         closeSocket();
     }
 
-    /**
-     * Returns the output stream associated with this connection. This
-     * should only be used by the connection manager.
-     */
-    protected OutputStream getOutputStream ()
-    {
-        return _out;
-    }
-
     /**
      * Returns the object input stream associated with this connection.
      * This should only be used by the connection manager.
@@ -197,23 +186,26 @@ public abstract class Connection implements NetEventHandler
      */
     protected void closeSocket ()
     {
-        if (_socket != null) {
+        if (_channel != null) {
+            Log.debug("Closing channel " + this + ".");
+
             try {
-                _socket.close();
+                _channel.close();
             } catch (IOException ioe) {
                 Log.warning("Error closing connection [conn=" + this +
                             ", error=" + ioe + "].");
             }
 
-            // clear out our socket reference to prevent repeat closings
+            // clear out our references to prevent repeat closings
-            _socket = null;
+            _selkey = null;
+            _channel = null;
         }
     }
 
     /**
      * Called when our client socket has data available for reading.
      */
-    public int handleEvent (long when, Selectable source, short events)
+    public int handleEvent (long when)
     {
         // make a note that we received an event as of this time
         _lastEvent = when;
@@ -228,8 +220,9 @@ public abstract class Connection implements NetEventHandler
                 _oin = new ObjectInputStream(_fin);
             }
 
-            // read the available data and see if we have a whole frame
+            // there may be more than one frame in the buffer, so we keep
-            if (_fin.readFrame(_in)) {
+            // 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())
@@ -246,14 +239,14 @@ public abstract class Connection implements NetEventHandler
 
         } catch (ClassNotFoundException cnfe) {
             Log.warning("Error reading message from socket " +
-                        "[socket=" + _socket + ", error=" + cnfe + "].");
+                        "[channel=" + _channel + ", error=" + cnfe + "].");
             // deal with the failure
             handleFailure(new NestableIOException(
                               "Unable to decode incoming message.", cnfe));
 
         } catch (IOException ioe) {
             Log.warning("Error reading message from socket " +
-                        "[socket=" + _socket + ", error=" + ioe + "].");
+                        "[channel=" + _channel + ", error=" + ioe + "].");
             // deal with the failure
             handleFailure(ioe);
         }
@@ -262,18 +255,18 @@ public abstract class Connection implements NetEventHandler
     }
 
     // documentation inherited from interface
-    public void checkIdle (long now)
+    public boolean checkIdle (long now)
     {
         long idleMillis = now - _lastEvent;
         if (idleMillis < PingRequest.PING_INTERVAL + LATENCY_GRACE) {
-            return;
+            return false;
         }
         if (isClosed()) {
-            return;
+            return false;
         }
-        Log.info("Disconnecting idle client [conn=" + this +
+        Log.info("Disconnecting non-communicative client [conn=" + this +
                  ", idle=" + idleMillis + "ms].");
-        handleFailure(new SocketTimeoutException("Idle too long"));
+        return true;
     }
 
     /**
@@ -288,16 +281,14 @@ public abstract class Connection implements NetEventHandler
     }
 
     protected ConnectionManager _cmgr;
-    protected NonblockingSocket _socket;
+    protected SelectionKey _selkey;
-    protected SelectItem _selitem;
+    protected SocketChannel _channel;
 
     protected long _lastEvent;
 
-    protected InputStream _in;
     protected FramedInputStream _fin;
     protected ObjectInputStream _oin;
 
-    protected OutputStream _out;
     protected ObjectOutputStream _oout;
 
     protected MessageHandler _handler;

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

@@ -1,13 +1,26 @@
 //
-// $Id: ConnectionManager.java,v 1.24 2002/11/05 02:17:56 mdb Exp $
+// $Id: ConnectionManager.java,v 1.25 2002/11/18 18:53:10 mdb Exp $
 
 package com.threerings.presents.server.net;
 
 import java.io.IOException;
-import java.net.SocketException;
+import java.nio.ByteBuffer;
-import java.util.ArrayList;
+import java.nio.channels.SelectableChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
+import java.nio.channels.ServerSocketChannel;
+import java.nio.channels.SocketChannel;
+import java.nio.channels.spi.SelectorProvider;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketException;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.Set;
 
-import ninja2.core.io_core.nbio.*;
 import com.samskivert.util.*;
 
 import com.threerings.io.FramingOutputStream;
@@ -41,34 +54,42 @@ public class ConnectionManager extends LoopingThread
         throws IOException
     {
         _port = port;
-        _selset = new SelectSet();
+        _selector = SelectorProvider.provider().openSelector();
 
         // register as a "state of server" reporter
         PresentsServer.registerReporter(this);
 
         try {
             // create our listening socket and add it to the select set
-            _listener = new NonblockingServerSocket(_port);
+            _listener = ServerSocketChannel.open();
-        } catch (SocketException se) {
+            _listener.configureBlocking(false);
+
+            InetAddress lh = InetAddress.getLocalHost();
+            InetSocketAddress isa = new InetSocketAddress(lh, _port);
+            _listener.socket().bind(isa);
+            Log.info("Server listening on " + isa + ".");
+
+        } catch (IOException ioe) {
             Log.warning("Failure creating listen socket " +
                         "[port=" + _port + "].");
-            throw se;
+            throw ioe;
         }
 
-        _litem = new SelectItem(_listener, Selectable.ACCEPT_READY);
+        // register our listening socket and map its select key to a net
-        // when an ACCEPT_READY event happens, we do this:
+        // event handler that will accept new connections
-        _litem.obj = new NetEventHandler() {
+        SelectionKey lkey =
-            public int handleEvent (long when, Selectable item, short events) {
+            _listener.register(_selector, SelectionKey.OP_ACCEPT);
+        _handlers.put(lkey, new NetEventHandler() {
+            public int handleEvent (long when) {
                 acceptConnection();
                 // there's no easy way to measure bytes read when
                 // accepting a connection, so we claim nothing
                 return 0;
             }
-            public void checkIdle (long now) {
+            public boolean checkIdle (long now) {
-                // we're never idle
+                return false; // we're never idle
             }
-        };
+        });
-        _selset.add(_litem);
 
         // we'll use this for sending messages to clients
         _framer = new FramingOutputStream();
@@ -217,18 +238,22 @@ public class ConnectionManager extends LoopingThread
             }
         }
 
-        // close any connections that have seen no network traffic for too
+        // close connections that have had no network traffic for too long
-        // long
+        Iterator hiter = _handlers.values().iterator();
-        int ccount = _selset.size();
+        while (hiter.hasNext()) {
-        for (int ii = 0; ii < ccount; ii++) {
+            NetEventHandler handler = (NetEventHandler)hiter.next();
-            ((NetEventHandler)_selset.elementAt(ii).getObj()).
+            if (handler.checkIdle(iterStamp)) {
-                checkIdle(iterStamp);
+                // this will queue the connection up for closure on our
+                // next tick
+                closeConnection(((Connection)handler));
+            }
         }
 
         // send any messages that are waiting on the outgoing queue
         Tuple tup;
         while ((tup = (Tuple)_outq.getNonBlocking()) != null) {
             Connection conn = (Connection)tup.left;
+
             // if the connection to which this message is destined is
             // closed, drop the message and move along quietly; this is
             // perfectly legal, a user can logoff whenever they like, even
@@ -249,10 +274,27 @@ public class ConnectionManager extends LoopingThread
                 oout.flush();
 
                 // then write framed message to real output stream
-                int out = _framer.writeFrameAndReset(conn.getOutputStream());
+                try {
-                synchronized (this) {
+                    ByteBuffer buffer = _framer.frameAndReturnBuffer();
-                    _bytesOut += out;
+                    int wrote = conn.getChannel().write(buffer);
-                    _msgsOut++;
+                    if (wrote != buffer.limit()) {
+                        Log.warning("Aiya! Failed to write complete message " +
+                                    "[conn=" + conn + ", wrote=" + wrote +
+                                    ", needed=" + buffer.limit() + "].");
+                        // TODO: give them the boot
+//                     } else {
+//                         Log.info("Sent [msg=" + outmsg +
+//                                  ", size=" + wrote + "].");
+                    }
+
+                    int out = buffer.limit();
+                    synchronized (this) {
+                        _bytesOut += wrote;
+                        _msgsOut++;
+                    }
+
+                } finally {
+                    _framer.resetFrame();
                 }
 
             } catch (IOException ioe) {
@@ -264,20 +306,21 @@ public class ConnectionManager extends LoopingThread
         // check for connections that have completed authentication
         AuthingConnection conn;
         while ((conn = (AuthingConnection)_authq.getNonBlocking()) != null) {
-            // remove the old connection from the select set
-            _selset.remove(conn.getSelectItem());
-
-            // construct a new running connection to handle this
-            // connections network traffic from here on out
             try {
-                RunningConnection rconn =
+                // construct a new running connection to handle this
-                    new RunningConnection(this, conn.getSocket(), iterStamp);
+                // connections network traffic from here on out
+                SelectionKey selkey = conn.getSelectionKey();
+                RunningConnection rconn = new RunningConnection(
+                    this, selkey, conn.getChannel(), iterStamp);
+
                 // we need to keep using the same object input and output
                 // streams from the beginning of the session because they
                 // have contextual state that needs to be preserved
                 rconn.inheritStreams(conn);
-                // wire this connection up to receive network events
+
-                _selset.add(rconn.getSelectItem());
+                // replace the mapping in the handlers table from the old
+                // connection with the new one
+                _handlers.put(selkey, rconn);
 
                 // and let our observers know about our new connection
                 notifyObservers(CONNECTION_ESTABLISHED, rconn,
@@ -290,20 +333,47 @@ public class ConnectionManager extends LoopingThread
             }
         }
 
-        // check for incoming network events
+        Set ready = null;
-        int ecount = _selset.select(SELECT_LOOP_TIME);
+        try {
-        if (ecount == 0) {
+            // check for incoming network events
+//             Log.debug("Selecting from " +
+//                       StringUtil.toString(_selector.keys()) + " (" +
+//                       SELECT_LOOP_TIME + ").");
+            int ecount = _selector.select(SELECT_LOOP_TIME);
+            ready = _selector.selectedKeys();
+            if (ecount == 0) {
+                if (ready.size() == 0) {
+                    return;
+                } else {
+                    Log.warning("select() returned no selected sockets, " +
+                                "but there are " + ready.size() +
+                                " in the ready set.");
+                }
+            }
+
+        } catch (IOException ioe) {
+            Log.warning("Failure select()ing [ioe=" + ioe + "].");
             return;
         }
 
         // process those events
-        SelectItem[] active = _selset.getEvents();
+//         Log.info("Ready set " + StringUtil.toString(ready) + ".");
-        for (int i = 0; i < active.length; i++) {
+        Iterator siter = ready.iterator();
+        while (siter.hasNext()) {
+            SelectionKey selkey = (SelectionKey)siter.next();
             try {
-                SelectItem item = active[i];
+                NetEventHandler handler = (NetEventHandler)
-                NetEventHandler handler = (NetEventHandler)item.obj;
+                    _handlers.get(selkey);
-                int got = handler.handleEvent(
+                if (handler == null) {
-                    iterStamp, item.getSelectable(), item.revents);
+                    Log.warning("Received network event but have no " +
+                                "registered handler [selkey=" + selkey + "].");
+                    continue;
+                }
+
+//                 Log.info("Got event [selkey=" + selkey +
+//                          ", handler=" + handler + "].");
+
+                int got = handler.handleEvent(iterStamp);
                 if (got != 0) {
                     synchronized (this) {
                         _bytesIn += got;
@@ -319,6 +389,7 @@ public class ConnectionManager extends LoopingThread
                 Log.logStackTrace(e);
             }
         }
+        ready.clear();
     }
 
     // documentation inherited
@@ -341,34 +412,48 @@ public class ConnectionManager extends LoopingThread
      */
     protected void acceptConnection ()
     {
-        NonblockingSocket socket = null;
+        SocketChannel channel = null;
 
         try {
-            socket = _listener.nbAccept();
+            channel = _listener.accept();
-            if (socket == null) {
+            if (channel == null) {
                 // in theory this shouldn't happen because we got an
                 // ACCEPT_READY event, but better safe than sorry
-                // Log.info("Psych! Got ACCEPT_READY, but no connection.");
+                Log.info("Psych! Got ACCEPT_READY, but no connection.");
                 return;
             }
 
-            // create a new authing connection object to manage the
+            if (!(channel instanceof SelectableChannel)) {
-            // authentication of this client connection
+                Log.warning("Provided with un-selectable socket as " +
-            AuthingConnection acon = new AuthingConnection(this, socket);
+                            "result of accept(), can't cope " +
+                            "[channel=" + channel + "].");
+                // stick a fork in the socket
+                channel.socket().close();
+                return;
+            }
 
-            // wire this connection into the select set
+            Log.debug("Accepted connection " + channel + ".");
-            _selset.add(acon.getSelectItem());
+
+            // create a new authing connection object to manage the
+            // authentication of this client connection and register it
+            // with our selection set
+            SelectableChannel selchan = (SelectableChannel)channel;
+            selchan.configureBlocking(false);
+            SelectionKey selkey = selchan.register(
+                _selector, SelectionKey.OP_READ);
+            _handlers.put(selkey, new AuthingConnection(this, selkey, channel));
+            return;
 
         } catch (IOException ioe) {
             Log.warning("Failure accepting new connection: " + ioe);
+        }
 
-            // make sure we don't leak a socket if something went awry
+        // make sure we don't leak a socket if something went awry
-            if (socket != null) {
+        if (channel != null) {
-                try {
+            try {
-                    socket.close();
+                channel.socket().close();
-                } catch (IOException ioe2) {
+            } catch (IOException ioe) {
-                    Log.warning("Failed closing aborted connection: " + ioe2);
+                Log.warning("Failed closing aborted connection: " + ioe);
-                }
             }
         }
     }
@@ -395,8 +480,9 @@ public class ConnectionManager extends LoopingThread
      */
     void connectionFailed (Connection conn, IOException ioe)
     {
-        // remove this connection from the select set
+        // remove this connection from our mapping (it is automatically
-        _selset.remove(conn.getSelectItem());
+        // removed from the Selector when the socket is closed)
+        _handlers.remove(conn.getSelectionKey());
 
         // let our observers know what's up
         notifyObservers(CONNECTION_FAILED, conn, ioe, null);
@@ -407,8 +493,9 @@ public class ConnectionManager extends LoopingThread
      */
     void connectionClosed (Connection conn)
     {
-        // remove this connection from the select set
+        // remove this connection from our mapping (it is automatically
-        _selset.remove(conn.getSelectItem());
+        // removed from the Selector when the socket is closed)
+        _handlers.remove(conn.getSelectionKey());
 
         // let our observers know what's up
         notifyObservers(CONNECTION_CLOSED, conn, null, null);
@@ -416,10 +503,11 @@ public class ConnectionManager extends LoopingThread
 
     protected int _port;
     protected Authenticator _author;
-    protected SelectSet _selset;
+    protected Selector _selector;
+    protected ServerSocketChannel _listener;
 
-    protected NonblockingServerSocket _listener;
+    /** Maps selection keys to network event handlers. */
-    protected SelectItem _litem;
+    protected HashMap _handlers = new HashMap();
 
     protected Queue _deathq = new Queue();
     protected Queue _authq = new Queue();
@@ -437,12 +525,12 @@ public class ConnectionManager extends LoopingThread
 
     /**
      * How long we wait for network events before checking our running
-     * flag to see if we should still be running.
+     * flag to see if we should still be running. We don't want to loop
+     * too tightly, but we need to make sure we don't sit around listening
+     * for incoming network events too long when there are outgoing
+     * messages in the queue.
      */
-    // protected static final int SELECT_LOOP_TIME = 30 * 1000;
+    protected static final int SELECT_LOOP_TIME = 100;
-
-    // while we're testing, use a short loop time
-    protected static final int SELECT_LOOP_TIME = 10;
 
     // codes for notifyObservers()
     protected static final int CONNECTION_ESTABLISHED = 0;

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

@@ -1,36 +1,36 @@
 //
-// $Id: NetEventHandler.java,v 1.5 2002/11/05 02:17:56 mdb Exp $
+// $Id: NetEventHandler.java,v 1.6 2002/11/18 18:53:10 mdb Exp $
 
 package com.threerings.presents.server.net;
 
-import ninja2.core.io_core.nbio.Selectable;
-
 /**
- * When a network event arrives on a particular <code>Selectable</code>,
+ * When a network event occurs, the connection manager calls the net event
- * the connection manager calls the net event handler associated with that
+ * handler associated with that channel to process the event. There are
- * selectable to process the event. There are only a few handlers (and
+ * only a few handlers (and probably only ever will be): the one that
- * probably only ever will be): the one that accepts new connections, the
+ * accepts new connections, the one that deals with a connection while the
- * one that deals with a connection while the client is authenticating and
+ * client is authenticating and the one that processes messages from
- * the one that processes messages from authenticated connections.
+ * authenticated connections.
- * Providing this interface prevents us from having to do a bunch of
+ *
+ * <p> Utilising this interface prevents us from having to do a bunch of
  * inefficient and ugly comparisons; instead we can simply call through an
  * interface method to the proper code.
  */
 public interface NetEventHandler
 {
     /**
-     * Called when a network event has occurred on the supplied source.
+     * Called when a network event has occurred on this handler's source.
-     * The <code>events</code> parameter indicates which event or events
-     * have occurred.
      *
      * @return the number of bytes read from the network as a result of
      * handling this event.
      */
-    public int handleEvent (long when, Selectable source, short events);
+    public int handleEvent (long when);
 
     /**
-     * Called to ensure that this selectable has not been idle for longer
+     * Called to ensure that this channel has not been idle for longer
      * than is possible in happily operating circumstances.
+     *
+     * @return true if the handler is idle (in which case it will be
+     * closed shortly), false if it is not.
      */
-    public void checkIdle (long now);
+    public boolean checkIdle (long now);
 }

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

@@ -1,10 +1,12 @@
 //
-// $Id: RunningConnection.java,v 1.8 2002/10/29 23:51:26 mdb Exp $
+// $Id: RunningConnection.java,v 1.9 2002/11/18 18:53:10 mdb Exp $
 
 package com.threerings.presents.server.net;
 
 import java.io.IOException;
-import ninja2.core.io_core.nbio.NonblockingSocket;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.SocketChannel;
+
 import com.samskivert.util.StringUtil;
 
 import com.threerings.presents.net.UpstreamMessage;
@@ -19,12 +21,11 @@ public class RunningConnection extends Connection
      * Constructs a new running connection object to manage the supplied
      * client socket.
      */
-    public RunningConnection (ConnectionManager cmgr,
+    public RunningConnection (ConnectionManager cmgr, SelectionKey selkey,
-                              NonblockingSocket socket,
+                              SocketChannel channel, long createStamp)
-                              long createStamp)
         throws IOException
     {
-        super(cmgr, socket, createStamp);
+        super(cmgr, selkey, channel, createStamp);
     }
 
     /**
@@ -36,12 +37,9 @@ public class RunningConnection extends Connection
 
     public String toString ()
     {
-        if (_socket != null) {
+        String addr = (_channel == null) ? "<disconnected>" :
-            return "[mode=RUNNING, id=" + (hashCode() % 1000) +
+            StringUtil.toString(_channel.socket().getInetAddress());
-                ", addr=" + StringUtil.toString(_socket.getInetAddress()) + "]";
+        return "[mode=RUNNING, id=" + (hashCode() % 1000) +
-        } else {
+            ", addr=" + addr + "]";
-            return "[mode=RUNNING, id=" + (hashCode() % 1000) +
-                ", addr=<disconnected>]";
-        }
     }
 }