diff --git a/src/java/com/threerings/presents/client/BlockingCommunicator.java b/src/java/com/threerings/presents/client/BlockingCommunicator.java
new file mode 100644
index 000000000..e35530415
--- /dev/null
+++ b/src/java/com/threerings/presents/client/BlockingCommunicator.java
@@ -0,0 +1,582 @@
+//
+// $Id: Communicator.java 4602 2007-02-24 00:39:27Z mdb $
+//
+// 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.client;
+
+import java.io.EOFException;
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+
+import java.net.ConnectException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+
+import com.samskivert.util.LoopingThread;
+import com.samskivert.util.Queue;
+import com.samskivert.util.StringUtil;
+
+import com.threerings.io.FramedInputStream;
+import com.threerings.io.FramingOutputStream;
+import com.threerings.io.ObjectInputStream;
+import com.threerings.io.ObjectOutputStream;
+
+import com.threerings.presents.Log;
+import com.threerings.presents.data.AuthCodes;
+import com.threerings.presents.dobj.DObjectManager;
+import com.threerings.presents.net.AuthRequest;
+import com.threerings.presents.net.AuthResponse;
+import com.threerings.presents.net.AuthResponseData;
+import com.threerings.presents.net.DownstreamMessage;
+import com.threerings.presents.net.LogoffRequest;
+import com.threerings.presents.net.UpstreamMessage;
+
+/**
+ * The client performs all network I/O on separate threads (one for reading and one for
+ * writing). The communicator class encapsulates that functionality.
+ *
+ *
+ * Logon synopsis:
+ *
+ * Client.logon():
+ * - Calls Communicator.start()
+ * Communicator.start():
+ * - spawn Reader thread
+ * Reader.run():
+ * { - connect
+ * - authenticate
+ * } if either fail, notify observers of failed logon
+ * - start writer thread
+ * - notify observers that we're logged on
+ * - read loop
+ * Writer.run():
+ * - write loop
+ *
+ */
+public class BlockingCommunicator extends Communicator
+{
+ /**
+ * Creates a new communicator instance which is associated with the supplied client.
+ */
+ public BlockingCommunicator (Client client)
+ {
+ super(client);
+ }
+
+ @Override // from Communicator
+ public void logon ()
+ {
+ // make sure things are copacetic
+ if (_reader != null) {
+ throw new RuntimeException("Communicator already started.");
+ }
+
+ // start up the reader thread. it will connect to the server and start up the writer thread
+ // if everything went successfully
+ _reader = new Reader();
+ _reader.start();
+ }
+
+ @Override // from Communicator
+ public synchronized void logoff ()
+ {
+ // if our socket is already closed, we've already taken care of this business
+ if (_channel == null) {
+ return;
+ }
+
+ // post a logoff message
+ postMessage(new LogoffRequest());
+
+ // let our reader and writer know that it's time to go
+ if (_reader != null) {
+ // if logoff() is being called by the client as part of a normal shutdown, this will
+ // cause the reader thread to be interrupted and shutdown gracefully. if logoff is
+ // being called by the reader thread as a result of a failed socket, it won't interrupt
+ // itself as it is already shutting down gracefully. if the JVM is buggy and calling
+ // interrupt() on a thread that is blocked on a socket doesn't wake it up, then when we
+ // close() the socket a bit further down, we have another chance that the reader thread
+ // will wake up; this time slightly less gracefully because it will think there's a
+ // network error when in fact we're just shutting down, but at least it will cleanly
+ // exit
+ _reader.shutdown();
+ }
+ if (_writer != null) {
+ // shutting down the writer thread is simpler because we can post a termination message
+ // on the queue and be sure that it will receive it. when the writer thread has
+ // delivered our logoff request and exited, we will complete the logoff process by
+ // closing our socket and invoking the clientDidLogoff callback
+ _writer.shutdown();
+ }
+ }
+
+ @Override // from Communicator
+ public void postMessage (UpstreamMessage msg)
+ {
+ // simply append the message to the queue
+ _msgq.append(msg);
+ }
+
+ @Override // from Communicator
+ public void setClassLoader (ClassLoader loader)
+ {
+ _loader = loader;
+ if (_oin != null) {
+ _oin.setClassLoader(loader);
+ }
+ }
+
+ @Override // from Communicator
+ public synchronized long getLastWrite ()
+ {
+ return _lastWrite;
+ }
+
+ /**
+ * 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);
+
+ // create our distributed object manager
+ _omgr = new ClientDObjectMgr(this, _client);
+
+ // create a new writer thread and start it up
+ if (_writer != null) {
+ throw new RuntimeException("Writer already started!?");
+ }
+ _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)
+ {
+ // make sure the socket isn't already closed down (meaning we've already dealt with the
+ // failed connection)
+ if (_channel == null) {
+ return;
+ }
+
+ Log.info("Connection failed: " + ioe);
+ Log.logStackTrace(ioe);
+
+ // let the client know that things went south
+ _client.notifyObservers(Client.CLIENT_CONNECTION_FAILED, ioe);
+
+ // and request that we go through the motions of logging off
+ logoff();
+ }
+
+ /**
+ * Callback called by the reader if the server closes the other end of the connection.
+ */
+ protected synchronized void connectionClosed ()
+ {
+ // make sure the socket isn't already closed down (meaning we've already dealt with the
+ // closed connection)
+ if (_channel == null) {
+ return;
+ }
+
+ Log.debug("Connection closed.");
+ // now do the whole logoff thing
+ logoff();
+ }
+
+ /**
+ * Callback called by the reader thread when it goes away.
+ */
+ protected synchronized void readerDidExit ()
+ {
+ // clear out our reader reference
+ _reader = null;
+
+ if (_writer == null) {
+ // there's no writer during authentication, so we may be responsible for closing the
+ // socket channel
+ closeChannel();
+
+ // let the client know when we finally go away
+ _client.cleanup(_logonError);
+ }
+
+ Log.debug("Reader thread exited.");
+ }
+
+ /**
+ * Callback called by the writer thread when it goes away.
+ */
+ protected synchronized void writerDidExit ()
+ {
+ // clear out our writer reference
+ _writer = null;
+ Log.debug("Writer thread exited.");
+
+ // let the client observers know that we're logged off
+ _client.notifyObservers(Client.CLIENT_DID_LOGOFF, null);
+
+ // 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
+ closeChannel();
+
+ // let the client know when we finally go away
+ if (_reader == null) {
+ _client.cleanup(_logonError);
+ }
+ }
+
+ /**
+ * Closes the socket channel that we have open to the server. Called by either {@link
+ * #readerDidExit} or {@link #writerDidExit} whichever is called last.
+ */
+ protected void closeChannel ()
+ {
+ if (_channel != null) {
+ Log.debug("Closing socket channel.");
+
+ try {
+ _channel.close();
+ } catch (IOException ioe) {
+ Log.warning("Error closing failed socket: " + ioe);
+ }
+ _channel = null;
+
+ // clear these out because they are probably large and in charge
+ _oin = null;
+ _oout = null;
+ }
+ }
+
+ /**
+ * Writes the supplied message to the socket.
+ */
+ protected void sendMessage (UpstreamMessage msg)
+ throws IOException
+ {
+ if (debugLogMessages()) {
+ Log.info("SEND " + msg);
+ }
+
+ // first we write the message so that we can measure it's length
+ _oout.writeObject(msg);
+ _oout.flush();
+
+ // then write the framed message to actual output stream
+ try {
+ ByteBuffer buffer = _fout.frameAndReturnBuffer();
+ if (buffer.limit() > 4096) {
+ String txt = StringUtil.truncate(String.valueOf(msg), 80, "...");
+ Log.info("Whoa, writin' a big one [msg=" + txt + ", size=" + buffer.limit() + "].");
+ }
+ 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();
+ }
+
+ /**
+ * 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).
+ */
+ protected DownstreamMessage receiveMessage ()
+ throws IOException
+ {
+ // read in the next message frame (readFrame() can return false meaning it only read part
+ // of the frame from the network, in 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(_channel));
+
+ try {
+ DownstreamMessage msg = (DownstreamMessage)_oin.readObject();
+ if (debugLogMessages()) {
+ Log.info("RECEIVE " + msg);
+ }
+ return msg;
+
+ } catch (ClassNotFoundException cnfe) {
+ throw (IOException) new IOException(
+ "Unable to decode incoming message.").initCause(cnfe);
+ }
+ }
+
+ /**
+ * 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
+ {
+ // the default implementation just connects to the first port and does no cycling
+ int port = _client.getPorts()[0];
+ Log.info("Connecting [host=" + host + ", port=" + port + "].");
+ synchronized (BlockingCommunicator.this) {
+ _channel = SocketChannel.open(new InetSocketAddress(host, port));
+ }
+ }
+
+ protected boolean debugLogMessages ()
+ {
+ return false;
+ }
+
+ /**
+ * The reader encapsulates the authentication and message reading process. It calls back to the
+ * {@link Communicator} class to do things, but the general flow of the reader thread is
+ * encapsulated in this class.
+ */
+ protected class Reader extends LoopingThread
+ {
+ protected void willStart ()
+ {
+ // first we connect and authenticate with the server
+ try {
+ // connect to the server
+ connect();
+
+ // then authenticate
+ logon();
+
+ } catch (Exception e) {
+ Log.debug("Logon failed: " + e);
+ // Log.logStackTrace(e);
+ // once we're shutdown we'll report this error
+ _logonError = e;
+ // terminate our communicator thread
+ shutdown();
+ }
+ }
+
+ protected void connect ()
+ throws IOException
+ {
+ // if we're already connected, we freak out
+ if (_channel != null) {
+ throw new IOException("Already connected.");
+ }
+
+ // look up the address of the target server
+ InetAddress host = InetAddress.getByName(_client.getHostname());
+ openChannel(host);
+ _channel.configureBlocking(true);
+
+ // our messages are framed (preceded by their length), so we use these helper streams
+ // to manage the framing
+ _fin = new FramedInputStream();
+ _fout = new FramingOutputStream();
+
+ // create our object input and output streams
+ _oin = new ObjectInputStream(_fin);
+ _oin.setClassLoader(_loader);
+ _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
+ protected void iterate ()
+ {
+ DownstreamMessage msg = null;
+
+ try {
+ // read the next message from the socket
+ msg = receiveMessage();
+
+ // process the message
+ processMessage(msg);
+
+ } catch (InterruptedIOException iioe) {
+ // somebody set up us the bomb! we've been interrupted which means that we're being
+ // shut down, so we just report it and return from iterate() like a good monkey
+ Log.debug("Reader thread woken up in time to die.");
+
+ } catch (EOFException eofe) {
+ // let the communicator know that our connection was closed
+ connectionClosed();
+ // and shut ourselves down
+ shutdown();
+
+ } catch (IOException ioe) {
+ // let the communicator know that our connection failed
+ connectionFailed(ioe);
+ // and shut ourselves down
+ shutdown();
+
+ } catch (Exception e) {
+ Log.warning("Error processing message [msg=" + msg + ", error=" + e + "].");
+ }
+ }
+
+ protected void handleIterateFailure (Exception e)
+ {
+ Log.warning("Uncaught exception it reader thread.");
+ Log.logStackTrace(e);
+ }
+
+ protected void didShutdown ()
+ {
+ // let the communicator know when we finally go away
+ readerDidExit();
+ }
+
+ protected void kick ()
+ {
+ // 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();
+ }
+ }
+
+ /**
+ * The writer encapsulates the message writing process. It calls back to the {@link
+ * Communicator} class to do things, but the general flow of the writer thread is encapsulated
+ * in this class.
+ */
+ protected class Writer extends LoopingThread
+ {
+ protected void iterate ()
+ {
+ // fetch the next message from the queue
+ UpstreamMessage msg = _msgq.get();
+
+ // if this is a termination message, we're being requested to exit, so we want to bail
+ // now rather than continuing
+ if (msg instanceof TerminationMessage) {
+ return;
+ }
+
+ try {
+ // write the message out the socket
+ sendMessage(msg);
+
+ } catch (IOException ioe) {
+ // let the communicator know if we have any problems
+ connectionFailed(ioe);
+ // and bail
+ shutdown();
+ }
+ }
+
+ protected void handleIterateFailure (Exception e)
+ {
+ Log.warning("Uncaught exception it writer thread.");
+ Log.logStackTrace(e);
+ }
+
+ protected void didShutdown ()
+ {
+ writerDidExit();
+ }
+
+ protected void kick ()
+ {
+ // post a bogus message to the outgoing queue to ensure that the writer thread notices
+ // that it's time to go
+ postMessage(new TerminationMessage());
+ }
+ }
+
+ /** This is used to terminate the writer thread. */
+ protected static class TerminationMessage extends UpstreamMessage
+ {
+ }
+
+ protected Client _client;
+ protected Reader _reader;
+ protected Writer _writer;
+
+ protected SocketChannel _channel;
+ protected Queue _msgq = new Queue();
+
+ protected long _lastWrite;
+ protected Exception _logonError;
+
+ /** We use this to frame our upstream messages. */
+ protected FramingOutputStream _fout;
+ protected ObjectOutputStream _oout;
+
+ /** We use this to frame our downstream messages. */
+ protected FramedInputStream _fin;
+ protected ObjectInputStream _oin;
+
+ protected ClientDObjectMgr _omgr;
+ protected ClassLoader _loader;
+}
diff --git a/src/java/com/threerings/presents/client/Client.java b/src/java/com/threerings/presents/client/Client.java
index 128fac1fa..d1edd1807 100644
--- a/src/java/com/threerings/presents/client/Client.java
+++ b/src/java/com/threerings/presents/client/Client.java
@@ -379,7 +379,7 @@ public class Client
// otherwise create a new communicator instance and start it up. this will initiate the
// logon process
- _comm = new Communicator(this);
+ _comm = createCommunicator();
_comm.setClassLoader(_loader);
_comm.logon();
@@ -491,7 +491,9 @@ public class Client
*/
protected void gotBootstrap (BootstrapData data, DObjectManager omgr)
{
- log.fine("Got bootstrap " + data + ".");
+ if (debugLogMessages()) {
+ log.info("Got bootstrap " + data + ".");
+ }
// keep these around for interested parties
_bstrap = data;
@@ -521,6 +523,14 @@ public class Client
// nothing by default
}
+ /**
+ * Creates the communicator that this client will use to send and receive messages.
+ */
+ protected Communicator createCommunicator ()
+ {
+ return new ClientCommunicator(this);
+ }
+
/**
* Called every five seconds; ensures that we ping the server if we haven't communicated in a
* long while and periodically resyncs the client and server clock deltas.
@@ -536,7 +546,9 @@ public class Client
if (_dcalc != null) {
// if our current calculator is done, clear it out
if (_dcalc.isDone()) {
- log.fine("Time offset from server: " + _serverDelta + "ms.");
+ if (debugLogMessages()) {
+ log.info("Time offset from server: " + _serverDelta + "ms.");
+ }
_dcalc = null;
} else if (_dcalc.shouldSendPing()) {
// otherwise, send another ping
@@ -692,6 +704,17 @@ public class Client
}
}
+ /**
+ * Whether or not to log low-level debug messages. This is used by the communicator as well
+ * which may be running on the server as a peer client, so we want to avoid constructing log
+ * messages when we're not logging and thus need to use this pattern rather than just
+ * log.fine.
+ */
+ protected boolean debugLogMessages ()
+ {
+ return false;
+ }
+
/**
* Used to notify client observers of events.
*/
diff --git a/src/java/com/threerings/presents/client/ClientCommunicator.java b/src/java/com/threerings/presents/client/ClientCommunicator.java
new file mode 100644
index 000000000..605fbf774
--- /dev/null
+++ b/src/java/com/threerings/presents/client/ClientCommunicator.java
@@ -0,0 +1,158 @@
+//
+// $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.client;
+
+import java.io.IOException;
+
+import java.net.ConnectException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.channels.SocketChannel;
+
+import com.samskivert.swing.RuntimeAdjust;
+import com.samskivert.util.IntListUtil;
+import com.samskivert.util.Interval;
+
+import com.threerings.presents.Log;
+import com.threerings.presents.data.AuthCodes;
+
+/**
+ * Customizes the blocking communicator with some things that we only do on users' machines (where
+ * there's only one client running, not potentially dozens, and where we're not sending high
+ * volumes of traffic through the client like we do for inter-server communications). This will go
+ * away when we create a special non-blocking communicator for use on the server that integrates
+ * with the ClientManager.
+ */
+public class ClientCommunicator extends BlockingCommunicator
+{
+ public ClientCommunicator (Client client)
+ {
+ super(client);
+ }
+
+ @Override // from BlockingCommunicator
+ protected void openChannel (InetAddress host)
+ throws IOException
+ {
+ // obtain the list of available ports on which to attempt our client connection and
+ // determine our preferred port
+ String pportKey = _client.getHostname() + ".preferred_port";
+ int[] ports = _client.getPorts();
+ int pport = PresentsPrefs.config.getValue(pportKey, ports[0]);
+ int ppidx = Math.max(0, IntListUtil.indexOf(ports, pport));
+
+ // try connecting on each of the ports in succession
+ for (int ii = 0; ii < ports.length; ii++) {
+ int port = ports[(ii+ppidx)%ports.length];
+ int nextPort = ports[(ii+ppidx+1)%ports.length];
+ Log.info("Connecting [host=" + host + ", port=" + port + "].");
+ InetSocketAddress addr = new InetSocketAddress(host, port);
+ try {
+ synchronized (this) {
+ clearPPI(true);
+ _prefPortInterval = new PrefPortInterval(pportKey, port, nextPort);
+ _channel = SocketChannel.open(addr);
+ _prefPortInterval.schedule(PREF_PORT_DELAY);
+ }
+ break;
+
+ } catch (IOException ioe) {
+ if (ioe instanceof ConnectException && ii < (ports.length-1)) {
+ _client.reportLogonTribulations(
+ new LogonException(AuthCodes.TRYING_NEXT_PORT, true));
+ continue; // try the next port
+ }
+ throw ioe;
+ }
+ }
+ }
+
+ @Override // from BlockingCommunicator
+ protected void readerDidExit ()
+ {
+ // if we haven't recorded a preferred port yet, instead do the failure action since we
+ // didn't stay connected long enough
+ clearPPI(true);
+ super.readerDidExit();
+ }
+
+ @Override // from BlockingCommunicator
+ protected boolean debugLogMessages ()
+ {
+ return _logMessages.getValue();
+ }
+
+ /**
+ * Cancels our preferred port saving interval. This method is called from the communication
+ * reader thread and the interval thread and must thus be synchronized.
+ */
+ protected synchronized boolean clearPPI (boolean cancel)
+ {
+ if (_prefPortInterval != null) {
+ if (cancel) {
+ _prefPortInterval.cancel();
+ _prefPortInterval.failed();
+ }
+ _prefPortInterval = null;
+ return true;
+ }
+ return false;
+ }
+
+ /** Used to save our preferred port once we know our connection is not going to be
+ * unceremoniously closed by Windows Connection Sharing. */
+ protected class PrefPortInterval extends Interval
+ {
+ public PrefPortInterval (String key, int thisPort, int nextPort) {
+ super();
+ _key = key;
+ _thisPort = thisPort;
+ _nextPort = nextPort;
+ }
+
+ public void expired () {
+ if (clearPPI(false)) {
+ PresentsPrefs.config.setValue(_key, _thisPort);
+ }
+ }
+
+ public void failed () {
+ PresentsPrefs.config.setValue(_key, _nextPort);
+ }
+
+ protected String _key;
+ protected int _thisPort;
+ protected int _nextPort;
+ }
+
+ /** We use this interval to record the preferred port if it stays connected long enough. */
+ protected PrefPortInterval _prefPortInterval;
+
+ /** Used to control low-level message logging. */
+ protected static RuntimeAdjust.BooleanAdjust _logMessages =
+ new RuntimeAdjust.BooleanAdjust("Toggles whether or not all sent and received low-level " +
+ "network events are logged.", "narya.presents.log_events",
+ PresentsPrefs.config, false);
+
+ /** Time a port must remain connected before we mark it as preferred. */
+ protected static long PREF_PORT_DELAY = 5000L;
+}
diff --git a/src/java/com/threerings/presents/client/Communicator.java b/src/java/com/threerings/presents/client/Communicator.java
index a74a0ee98..c2f64ed37 100644
--- a/src/java/com/threerings/presents/client/Communicator.java
+++ b/src/java/com/threerings/presents/client/Communicator.java
@@ -21,61 +21,12 @@
package com.threerings.presents.client;
-import java.io.EOFException;
-import java.io.IOException;
-import java.io.InterruptedIOException;
-import java.nio.ByteBuffer;
-import java.nio.channels.SocketChannel;
-
-import java.net.ConnectException;
-import java.net.InetAddress;
-import java.net.InetSocketAddress;
-
-import com.samskivert.swing.RuntimeAdjust;
-import com.samskivert.util.IntListUtil;
-import com.samskivert.util.Interval;
-import com.samskivert.util.LoopingThread;
-import com.samskivert.util.Queue;
-import com.samskivert.util.StringUtil;
-
-import com.threerings.io.FramedInputStream;
-import com.threerings.io.FramingOutputStream;
-import com.threerings.io.ObjectInputStream;
-import com.threerings.io.ObjectOutputStream;
-
-import com.threerings.presents.Log;
-import com.threerings.presents.data.AuthCodes;
-import com.threerings.presents.dobj.DObjectManager;
-import com.threerings.presents.net.AuthRequest;
-import com.threerings.presents.net.AuthResponse;
-import com.threerings.presents.net.AuthResponseData;
-import com.threerings.presents.net.DownstreamMessage;
-import com.threerings.presents.net.LogoffRequest;
import com.threerings.presents.net.UpstreamMessage;
/**
- * The client performs all network I/O on separate threads (one for reading and one for
- * writing). The communicator class encapsulates that functionality.
- *
- *
- * Logon synopsis:
- *
- * Client.logon():
- * - Calls Communicator.start()
- * Communicator.start():
- * - spawn Reader thread
- * Reader.run():
- * { - connect
- * - authenticate
- * } if either fail, notify observers of failed logon
- * - start writer thread
- * - notify observers that we're logged on
- * - read loop
- * Writer.run():
- * - write loop
- *
+ * Handles sending and receiving messages for the client.
*/
-public class Communicator
+public abstract class Communicator
{
/**
* Creates a new communicator instance which is associated with the supplied client.
@@ -88,585 +39,29 @@ public class Communicator
/**
* Logs on to the server and initiates our full-duplex message exchange.
*/
- public void logon ()
- {
- // make sure things are copacetic
- if (_reader != null) {
- throw new RuntimeException("Communicator already started.");
- }
-
- // start up the reader thread. it will connect to the server and start up the writer thread
- // if everything went successfully
- _reader = new Reader();
- _reader.start();
- }
+ public abstract void logon ();
/**
* Delivers a logoff notification to the server and shuts down the network connection. Also
* causes all communication threads to terminate.
*/
- public synchronized void logoff ()
- {
- // if our socket is already closed, we've already taken care of this business
- if (_channel == null) {
- return;
- }
-
- // post a logoff message
- postMessage(new LogoffRequest());
-
- // let our reader and writer know that it's time to go
- if (_reader != null) {
- // if logoff() is being called by the client as part of a normal shutdown, this will
- // cause the reader thread to be interrupted and shutdown gracefully. if logoff is
- // being called by the reader thread as a result of a failed socket, it won't interrupt
- // itself as it is already shutting down gracefully. if the JVM is buggy and calling
- // interrupt() on a thread that is blocked on a socket doesn't wake it up, then when we
- // close() the socket a bit further down, we have another chance that the reader thread
- // will wake up; this time slightly less gracefully because it will think there's a
- // network error when in fact we're just shutting down, but at least it will cleanly
- // exit
- _reader.shutdown();
- }
- if (_writer != null) {
- // shutting down the writer thread is simpler because we can post a termination message
- // on the queue and be sure that it will receive it. when the writer thread has
- // delivered our logoff request and exited, we will complete the logoff process by
- // closing our socket and invoking the clientDidLogoff callback
- _writer.shutdown();
- }
- }
+ public abstract void logoff ();
/**
* Queues up the specified message for delivery upstream.
*/
- public void postMessage (UpstreamMessage msg)
- {
- // simply append the message to the queue
- _msgq.append(msg);
- }
+ public abstract void postMessage (UpstreamMessage msg);
/**
* Configures this communicator with a custom class loader to be used when reading and writing
* objects over the network.
*/
- public void setClassLoader (ClassLoader loader)
- {
- _loader = loader;
- if (_oin != null) {
- _oin.setClassLoader(loader);
- }
- }
-
- /**
- * 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);
-
- // create our distributed object manager
- _omgr = new ClientDObjectMgr(this, _client);
-
- // create a new writer thread and start it up
- if (_writer != null) {
- throw new RuntimeException("Writer already started!?");
- }
- _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)
- {
- // make sure the socket isn't already closed down (meaning we've already dealt with the
- // failed connection)
- if (_channel == null) {
- return;
- }
-
- Log.info("Connection failed: " + ioe);
- Log.logStackTrace(ioe);
-
- // let the client know that things went south
- _client.notifyObservers(Client.CLIENT_CONNECTION_FAILED, ioe);
-
- // and request that we go through the motions of logging off
- logoff();
- }
-
- /**
- * Callback called by the reader if the server closes the other end of the connection.
- */
- protected synchronized void connectionClosed ()
- {
- // make sure the socket isn't already closed down (meaning we've already dealt with the
- // closed connection)
- if (_channel == null) {
- return;
- }
-
- Log.debug("Connection closed.");
- // now do the whole logoff thing
- logoff();
- }
-
- /**
- * Callback called by the reader thread when it goes away.
- */
- protected synchronized void readerDidExit ()
- {
- // clear out our reader reference
- _reader = null;
-
- if (_writer == null) {
- // there's no writer during authentication, so we may be responsible for closing the
- // socket channel
- closeChannel();
-
- // let the client know when we finally go away
- _client.cleanup(_logonError);
- }
-
- Log.debug("Reader thread exited.");
- }
-
- /**
- * Callback called by the writer thread when it goes away.
- */
- protected synchronized void writerDidExit ()
- {
- // clear out our writer reference
- _writer = null;
- Log.debug("Writer thread exited.");
-
- // let the client observers know that we're logged off
- _client.notifyObservers(Client.CLIENT_DID_LOGOFF, null);
-
- // 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
- closeChannel();
-
- // let the client know when we finally go away
- if (_reader == null) {
- _client.cleanup(_logonError);
- }
- }
-
- /**
- * Closes the socket channel that we have open to the server. Called by either {@link
- * #readerDidExit} or {@link #writerDidExit} whichever is called last.
- */
- protected void closeChannel ()
- {
- if (_channel != null) {
- Log.debug("Closing socket channel.");
-
- try {
- _channel.close();
- } catch (IOException ioe) {
- Log.warning("Error closing failed socket: " + ioe);
- }
- _channel = null;
-
- // clear these out because they are probably large and in charge
- _oin = null;
- _oout = null;
- }
- }
-
- /**
- * Writes the supplied message to the socket.
- */
- protected void sendMessage (UpstreamMessage msg)
- throws IOException
- {
- if (_logMessages.getValue()) {
- Log.info("SEND " + msg);
- }
-
- // first we write the message so that we can measure it's length
- _oout.writeObject(msg);
- _oout.flush();
-
- // then write the framed message to actual output stream
- try {
- ByteBuffer buffer = _fout.frameAndReturnBuffer();
- if (buffer.limit() > 4096) {
- String txt = StringUtil.truncate(String.valueOf(msg), 80, "...");
- Log.info("Whoa, writin' a big one [msg=" + txt + ", size=" + buffer.limit() + "].");
- }
- 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();
- }
+ public abstract void setClassLoader (ClassLoader loader);
/**
* Returns the time at which we last sent a packet to the server.
*/
- protected synchronized 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();
- }
-
- /**
- * Reads a new message from the socket (blocking until a message has arrived).
- */
- protected DownstreamMessage receiveMessage ()
- throws IOException
- {
- // read in the next message frame (readFrame() can return false meaning it only read part
- // of the frame from the network, in 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(_channel));
-
- try {
- DownstreamMessage msg = (DownstreamMessage)_oin.readObject();
- if (_logMessages.getValue()) {
- Log.info("RECEIVE " + msg);
- }
- return msg;
-
- } catch (ClassNotFoundException cnfe) {
- throw (IOException) new IOException(
- "Unable to decode incoming message.").initCause(cnfe);
- }
- }
-
- /**
- * 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);
- }
-
- /**
- * Cancels our preferred port saving interval. This method is called from the communication
- * reader thread and the interval thread and must thus be synchronized.
- */
- protected synchronized boolean clearPPI (boolean cancel)
- {
- if (_prefPortInterval != null) {
- if (cancel) {
- _prefPortInterval.cancel();
- _prefPortInterval.failed();
- }
- _prefPortInterval = null;
- return true;
- }
- return false;
- }
-
- /**
- * The reader encapsulates the authentication and message reading process. It calls back to the
- * {@link Communicator} class to do things, but the general flow of the reader thread is
- * encapsulated in this class.
- */
- protected class Reader extends LoopingThread
- {
- protected void willStart ()
- {
- // first we connect and authenticate with the server
- try {
- // connect to the server
- connect();
-
- // then authenticate
- logon();
-
- } catch (Exception e) {
- Log.debug("Logon failed: " + e);
- // Log.logStackTrace(e);
- // once we're shutdown we'll report this error
- _logonError = e;
- // terminate our communicator thread
- shutdown();
- }
- }
-
- protected void connect ()
- throws IOException
- {
- // if we're already connected, we freak out
- if (_channel != null) {
- throw new IOException("Already connected.");
- }
-
- // look up the address of the target server
- InetAddress host = InetAddress.getByName(_client.getHostname());
-
- // obtain the list of available ports on which to attempt our client connection and
- // determine our preferred port
- String pportKey = _client.getHostname() + ".preferred_port";
- int[] ports = _client.getPorts();
- int pport = PresentsPrefs.config.getValue(pportKey, ports[0]);
- int ppidx = Math.max(0, IntListUtil.indexOf(ports, pport));
-
- // try connecting on each of the ports in succession
- for (int ii = 0; ii < ports.length; ii++) {
- int port = ports[(ii+ppidx)%ports.length];
- int nextPort = ports[(ii+ppidx+1)%ports.length];
- Log.info("Connecting [host=" + host + ", port=" + port + "].");
- InetSocketAddress addr = new InetSocketAddress(host, port);
- try {
- synchronized (Communicator.this) {
- clearPPI(true);
- _prefPortInterval = new PrefPortInterval(pportKey, port, nextPort);
- _channel = SocketChannel.open(addr);
- _prefPortInterval.schedule(PREF_PORT_DELAY);
- }
- break;
-
- } catch (IOException ioe) {
- if (ioe instanceof ConnectException && ii < (ports.length-1)) {
- _client.reportLogonTribulations(
- new LogonException(AuthCodes.TRYING_NEXT_PORT, true));
- continue; // try the next port
- }
- throw ioe;
- }
- }
-
- _channel.configureBlocking(true);
-
- // our messages are framed (preceded by their length), so we use these helper streams
- // to manage the framing
- _fin = new FramedInputStream();
- _fout = new FramingOutputStream();
-
- // create our object input and output streams
- _oin = new ObjectInputStream(_fin);
- _oin.setClassLoader(_loader);
- _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
- protected void iterate ()
- {
- DownstreamMessage msg = null;
-
- try {
- // read the next message from the socket
- msg = receiveMessage();
-
- // process the message
- processMessage(msg);
-
- } catch (InterruptedIOException iioe) {
- // somebody set up us the bomb! we've been interrupted which means that we're being
- // shut down, so we just report it and return from iterate() like a good monkey
- Log.debug("Reader thread woken up in time to die.");
-
- } catch (EOFException eofe) {
- // let the communicator know that our connection was closed
- connectionClosed();
- // and shut ourselves down
- shutdown();
-
- } catch (IOException ioe) {
- // let the communicator know that our connection failed
- connectionFailed(ioe);
- // and shut ourselves down
- shutdown();
-
- } catch (Exception e) {
- Log.warning("Error processing message [msg=" + msg + ", error=" + e + "].");
- }
- }
-
- protected void handleIterateFailure (Exception e)
- {
- Log.warning("Uncaught exception it reader thread.");
- Log.logStackTrace(e);
- }
-
- protected void didShutdown ()
- {
- // if we haven't recorded a preferred port yet, instead do the failure action since we
- // didn't stay connected long enough
- clearPPI(true);
-
- // let the communicator know when we finally go away
- readerDidExit();
- }
-
- protected void kick ()
- {
- // 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();
- }
- }
-
- /**
- * The writer encapsulates the message writing process. It calls back to the {@link
- * Communicator} class to do things, but the general flow of the writer thread is encapsulated
- * in this class.
- */
- protected class Writer extends LoopingThread
- {
- protected void iterate ()
- {
- // fetch the next message from the queue
- UpstreamMessage msg = _msgq.get();
-
- // if this is a termination message, we're being requested to exit, so we want to bail
- // now rather than continuing
- if (msg instanceof TerminationMessage) {
- return;
- }
-
- try {
- // write the message out the socket
- sendMessage(msg);
-
- } catch (IOException ioe) {
- // let the communicator know if we have any problems
- connectionFailed(ioe);
- // and bail
- shutdown();
- }
- }
-
- protected void handleIterateFailure (Exception e)
- {
- Log.warning("Uncaught exception it writer thread.");
- Log.logStackTrace(e);
- }
-
- protected void didShutdown ()
- {
- writerDidExit();
- }
-
- protected void kick ()
- {
- // post a bogus message to the outgoing queue to ensure that the writer thread notices
- // that it's time to go
- postMessage(new TerminationMessage());
- }
- }
-
- /** This is used to terminate the writer thread. */
- protected static class TerminationMessage extends UpstreamMessage
- {
- }
-
- /** Used to save our preferred port once we know our connection is not going to be
- * unceremoniously closed by Windows Connection Sharing. */
- protected class PrefPortInterval extends Interval
- {
- public PrefPortInterval (String key, int thisPort, int nextPort) {
- super();
- _key = key;
- _thisPort = thisPort;
- _nextPort = nextPort;
- }
-
- public void expired () {
- if (clearPPI(false)) {
- PresentsPrefs.config.setValue(_key, _thisPort);
- }
- }
-
- public void failed () {
- PresentsPrefs.config.setValue(_key, _nextPort);
- }
-
- protected String _key;
- protected int _thisPort;
- protected int _nextPort;
- }
+ public abstract long getLastWrite ();
protected Client _client;
- protected Reader _reader;
- protected Writer _writer;
-
- protected SocketChannel _channel;
- protected Queue _msgq = new Queue();
-
- protected long _lastWrite;
- protected Exception _logonError;
-
- /** We use this to frame our upstream messages. */
- protected FramingOutputStream _fout;
- protected ObjectOutputStream _oout;
-
- /** We use this to frame our downstream messages. */
- protected FramedInputStream _fin;
- protected ObjectInputStream _oin;
-
- protected ClientDObjectMgr _omgr;
- protected ClassLoader _loader;
-
- /** We use this interval to record the preferred port if it stays connected long enough. */
- protected PrefPortInterval _prefPortInterval;
-
- /** Time a port must remain connected before we mark it as preferred. */
- protected static long PREF_PORT_DELAY = 5000L;
-
- /** Used to control low-level message logging. */
- protected static RuntimeAdjust.BooleanAdjust _logMessages =
- new RuntimeAdjust.BooleanAdjust("Toggles whether or not all sent and received low-level " +
- "network events are logged.", "narya.presents.log_events",
- PresentsPrefs.config, false);
}