diff --git a/src/java/com/threerings/io/ByteBufferInputStream.java b/src/java/com/threerings/io/ByteBufferInputStream.java
new file mode 100644
index 000000000..f6a22bda3
--- /dev/null
+++ b/src/java/com/threerings/io/ByteBufferInputStream.java
@@ -0,0 +1,97 @@
+//
+// $Id$
+
+package com.threerings.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+import java.nio.BufferUnderflowException;
+import java.nio.ByteBuffer;
+import java.nio.InvalidMarkException;
+
+/**
+ * Reads input from a {@link ByteBuffer}.
+ */
+public class ByteBufferInputStream extends InputStream
+{
+ /**
+ * Creates a new input stream to read from the specified buffer.
+ */
+ public ByteBufferInputStream (ByteBuffer buffer)
+ {
+ _buffer = buffer;
+ }
+
+ /**
+ * Returns a reference to the underlying buffer.
+ */
+ public ByteBuffer getBuffer ()
+ {
+ return _buffer;
+ }
+
+ @Override // documentation inherited
+ public int read ()
+ {
+ try {
+ return (_buffer.get() & 0xFF);
+ } catch (BufferUnderflowException e) {
+ return -1;
+ }
+ }
+
+ @Override // documentation inherited
+ public int read (byte[] b, int offset, int length)
+ throws IOException
+ {
+ length = Math.min(length, _buffer.remaining());
+ if (length <= 0) {
+ return -1;
+ }
+ _buffer.get(b, offset, length);
+ return length;
+ }
+
+ @Override // documentation inherited
+ public long skip (long n)
+ throws IOException
+ {
+ n = Math.min(n, _buffer.remaining());
+ _buffer.position((int)(_buffer.position() + n));
+ return n;
+ }
+
+ @Override // documentation inherited
+ public int available ()
+ {
+ return _buffer.remaining();
+ }
+
+ @Override // documentation inherited
+ public boolean markSupported ()
+ {
+ return true;
+ }
+
+ @Override // documentation inherited
+ public void mark (int readLimit)
+ {
+ _buffer.mark();
+ }
+
+ @Override // documentation inherited
+ public void reset ()
+ throws IOException
+ {
+ try {
+ _buffer.reset();
+ } catch (InvalidMarkException e) {
+ throw new IOException("No mark set.");
+ }
+ }
+
+ /** The buffer from which we read. */
+ protected ByteBuffer _buffer;
+}
+
diff --git a/src/java/com/threerings/io/ByteBufferOutputStream.java b/src/java/com/threerings/io/ByteBufferOutputStream.java
new file mode 100644
index 000000000..913d9732e
--- /dev/null
+++ b/src/java/com/threerings/io/ByteBufferOutputStream.java
@@ -0,0 +1,102 @@
+//
+// $Id$
+
+package com.threerings.io;
+
+import java.io.OutputStream;
+
+import java.nio.BufferOverflowException;
+import java.nio.ByteBuffer;
+
+/**
+ * Stores output in an {@link ByteBuffer} that grows automatically to accommodate the data.
+ */
+public class ByteBufferOutputStream extends OutputStream
+{
+ /**
+ * Creates a new byte buffer output stream.
+ */
+ public ByteBufferOutputStream ()
+ {
+ _buffer = ByteBuffer.allocate(INITIAL_BUFFER_SIZE);
+ }
+
+ /**
+ * Returns a reference to the underlying buffer.
+ */
+ public ByteBuffer getBuffer ()
+ {
+ return _buffer;
+ }
+
+ /**
+ * Flips and returns the buffer. The returned buffer will have a position of zero and a limit
+ * equal to the number of bytes written. Call {@link #reset} to reset the buffer before
+ * writing again.
+ */
+ public ByteBuffer flip ()
+ {
+ _buffer.flip();
+ return _buffer;
+ }
+
+ /**
+ * Resets our internal buffer.
+ */
+ public void reset ()
+ {
+ _buffer.clear();
+ }
+
+ @Override // documentation inherited
+ public void write (int b)
+ {
+ try {
+ _buffer.put((byte)b);
+ } catch (BufferOverflowException boe) {
+ expand(1);
+ _buffer.put((byte)b);
+ }
+ }
+
+ @Override // documentation inherited
+ public void write (byte[] b, int off, int len)
+ {
+ // sanity check the arguments
+ if ((off < 0) || (off > b.length) || (len < 0) ||
+ ((off + len) > b.length) || ((off + len) < 0)) {
+ throw new IndexOutOfBoundsException();
+ } else if (len == 0) {
+ return;
+ }
+
+ try {
+ _buffer.put(b, off, len);
+ } catch (BufferOverflowException boe) {
+ expand(len);
+ _buffer.put(b, off, len);
+ }
+ }
+
+ /**
+ * Expands our buffer to accomodate the specified capacity.
+ */
+ protected final void expand (int needed)
+ {
+ int ocapacity = _buffer.capacity();
+ int ncapacity = _buffer.position() + needed;
+ if (ncapacity > ocapacity) {
+ // increase the buffer size in large increments
+ ncapacity = Math.max(ocapacity << 1, ncapacity);
+ ByteBuffer newbuf = ByteBuffer.allocate(ncapacity);
+ newbuf.put((ByteBuffer)_buffer.flip());
+ _buffer = newbuf;
+ }
+ }
+
+ /** The buffer in which we store our frame data. */
+ protected ByteBuffer _buffer;
+
+ /** The default initial size of the internal buffer. */
+ protected static final int INITIAL_BUFFER_SIZE = 32;
+}
diff --git a/src/java/com/threerings/io/FramingOutputStream.java b/src/java/com/threerings/io/FramingOutputStream.java
index 65b786a60..0457c221c 100644
--- a/src/java/com/threerings/io/FramingOutputStream.java
+++ b/src/java/com/threerings/io/FramingOutputStream.java
@@ -40,69 +40,23 @@ import java.nio.ByteBuffer;
* to its internal buffer. It is intended to only be accessed from a
* single thread.
*/
-public class FramingOutputStream extends OutputStream
+public class FramingOutputStream extends ByteBufferOutputStream
{
public FramingOutputStream ()
{
- _buffer = ByteBuffer.allocate(INITIAL_BUFFER_SIZE);
_buffer.put(HEADER_PAD);
}
- /**
- * Writes the specified byte to this framing output stream.
- *
- * @param b the byte to be written.
- */
- public void write (int b)
+ @Override // documentation inherited
+ public ByteBuffer flip ()
{
- try {
- _buffer.put((byte)b);
- } catch (BufferOverflowException boe) {
- expand(1);
- _buffer.put((byte)b);
- }
+ throw new UnsupportedOperationException("Use frameAndReturnBuffer() instead.");
}
- /**
- * Writes len bytes from the specified byte array
- * starting at offset off to this framing output stream.
- *
- * @param b the data.
- * @param off the start offset in the data.
- * @param len the number of bytes to write.
- */
- public void write (byte[] b, int off, int len)
+ @Override // documentation inherited
+ public void reset ()
{
- // sanity check the arguments
- if ((off < 0) || (off > b.length) || (len < 0) ||
- ((off + len) > b.length) || ((off + len) < 0)) {
- throw new IndexOutOfBoundsException();
- } else if (len == 0) {
- return;
- }
-
- try {
- _buffer.put(b, off, len);
- } catch (BufferOverflowException boe) {
- expand(len);
- _buffer.put(b, off, len);
- }
- }
-
- /**
- * Expands our buffer to accomodate the specified capacity.
- */
- protected final void expand (int needed)
- {
- int ocapacity = _buffer.capacity();
- int ncapacity = _buffer.position() + needed;
- if (ncapacity > ocapacity) {
- // increase the buffer size in large increments
- ncapacity = Math.max(ocapacity << 1, ncapacity);
- ByteBuffer newbuf = ByteBuffer.allocate(ncapacity);
- newbuf.put((ByteBuffer)_buffer.flip());
- _buffer = newbuf;
- }
+ throw new UnsupportedOperationException("Use resetFrame() instead.");
}
/**
@@ -136,12 +90,6 @@ public class FramingOutputStream extends OutputStream
_buffer.put(HEADER_PAD);
}
- /** The buffer in which we store our frame data. */
- protected ByteBuffer _buffer;
-
- /** The default initial size of the internal buffer. */
- protected static final int INITIAL_BUFFER_SIZE = 32;
-
/** We pad the beginning of our buffer so that we can write the frame
* length when the time comes. */
protected static final byte[] HEADER_PAD = new byte[4];
diff --git a/src/java/com/threerings/io/ObjectInputStream.java b/src/java/com/threerings/io/ObjectInputStream.java
index 66a6e3f36..ba716070a 100644
--- a/src/java/com/threerings/io/ObjectInputStream.java
+++ b/src/java/com/threerings/io/ObjectInputStream.java
@@ -114,22 +114,8 @@ public class ObjectInputStream extends DataInputStream
}
}
- // create a class mapping record, and cache it
- Class sclass = Class.forName(cname, true, _loader);
- Streamer streamer = Streamer.getStreamer(sclass);
- if (STREAM_DEBUG) {
- log.info(hashCode() + ": New class '" + cname + "' [code=" + code + "].");
- }
-
- // sanity check
- if (streamer == null) {
- String errmsg = "Aiya! Unable to create streamer for newly seen class " +
- "[code=" + code + ", class=" + cname + "]";
- throw new RuntimeException(errmsg);
- }
-
- cmap = new ClassMapping(code, sclass, streamer);
- _classmap.add(code, cmap);
+ // create the class mapping
+ cmap = mapClass(code, cname);
} else {
cmap = _classmap.get(code);
@@ -162,6 +148,41 @@ public class ObjectInputStream extends DataInputStream
}
}
+ /**
+ * Creates, adds, and returns the class mapping for the specified code and class name.
+ */
+ protected ClassMapping mapClass (short code, String cname)
+ throws IOException, ClassNotFoundException
+ {
+ // create a class mapping record, and cache it
+ ClassMapping cmap = createClassMapping(code, cname);
+ _classmap.add(code, cmap);
+ return cmap;
+ }
+
+ /**
+ * Creates and returns a class mapping for the specified code and class name.
+ */
+ protected ClassMapping createClassMapping (short code, String cname)
+ throws IOException, ClassNotFoundException
+ {
+ // resolve the class and streamer
+ Class sclass = Class.forName(cname, true, _loader);
+ Streamer streamer = Streamer.getStreamer(sclass);
+ if (STREAM_DEBUG) {
+ log.info(hashCode() + ": New class '" + cname + "' [code=" + code + "].");
+ }
+
+ // sanity check
+ if (streamer == null) {
+ String errmsg = "Aiya! Unable to create streamer for newly seen class " +
+ "[code=" + code + ", class=" + cname + "]";
+ throw new RuntimeException(errmsg);
+ }
+
+ return new ClassMapping(code, sclass, streamer);
+ }
+
/**
* Reads an object from the input stream that was previously written with {@link
* ObjectOutputStream#writeBareObject}.
diff --git a/src/java/com/threerings/io/ObjectOutputStream.java b/src/java/com/threerings/io/ObjectOutputStream.java
index 4042db190..477278a63 100644
--- a/src/java/com/threerings/io/ObjectOutputStream.java
+++ b/src/java/com/threerings/io/ObjectOutputStream.java
@@ -89,32 +89,66 @@ public class ObjectOutputStream extends DataOutputStream
log.info(hashCode() + ": Creating class mapping [code=" + _nextCode +
", class=" + sclass.getName() + "].");
}
- cmap = new ClassMapping(_nextCode++, sclass, streamer);
+ cmap = createClassMapping(_nextCode++, sclass, streamer);
_classmap.put(sclass, cmap);
// make sure we didn't blow past our maximum class count
if (_nextCode <= 0) {
throw new RuntimeException("Too many unique classes written to ObjectOutputStream");
}
-
- // writing a negative class code indicates that the class name will follow
- writeShort(-cmap.code);
- String cname = sclass.getName();
- if (_translations != null) {
- String tname = _translations.get(cname);
- if (tname != null) {
- cname = tname;
- }
- }
- writeUTF(cname);
+ writeNewClassMapping(cmap);
} else {
- writeShort(cmap.code);
+ writeExistingClassMapping(cmap);
}
writeBareObject(object, cmap.streamer, true);
}
+ /**
+ * Creates and returns a new class mapping.
+ */
+ protected ClassMapping createClassMapping (short code, Class sclass, Streamer streamer)
+ {
+ return new ClassMapping(code, sclass, streamer);
+ }
+
+ /**
+ * Writes a new class mapping to the stream.
+ */
+ protected void writeNewClassMapping (ClassMapping cmap)
+ throws IOException
+ {
+ // writing a negative class code indicates that the class name will follow
+ writeClassMapping(-cmap.code, cmap.sclass);
+ }
+
+ /**
+ * Writes an existing class mapping to the stream.
+ */
+ protected void writeExistingClassMapping (ClassMapping cmap)
+ throws IOException
+ {
+ writeShort(cmap.code);
+ }
+
+ /**
+ * Writes out the mapping for a class.
+ */
+ protected void writeClassMapping (int code, Class sclass)
+ throws IOException
+ {
+ writeShort(code);
+ String cname = sclass.getName();
+ if (_translations != null) {
+ String tname = _translations.get(cname);
+ if (tname != null) {
+ cname = tname;
+ }
+ }
+ writeUTF(cname);
+ }
+
/**
* Writes a {@link Streamable} instance or one of the support object types without
* associated class metadata to the output stream. The caller is responsible for knowing
diff --git a/src/java/com/threerings/io/UnreliableObjectInputStream.java b/src/java/com/threerings/io/UnreliableObjectInputStream.java
new file mode 100644
index 000000000..bf773b5d4
--- /dev/null
+++ b/src/java/com/threerings/io/UnreliableObjectInputStream.java
@@ -0,0 +1,46 @@
+//
+// $Id$
+
+package com.threerings.io;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * The counterpart of {@link UnreliableObjectOutputStream}.
+ */
+public class UnreliableObjectInputStream extends ObjectInputStream
+{
+ /**
+ * Constructs an object input stream which will read its data from the supplied source stream.
+ */
+ public UnreliableObjectInputStream (InputStream source)
+ {
+ super(source);
+ }
+
+ @Override // documentation inherited
+ protected ClassMapping mapClass (short code, String cname)
+ throws IOException, ClassNotFoundException
+ {
+ // see if we already have a mapping
+ ClassMapping cmap = (code < _classmap.size()) ? _classmap.get(code) : null;
+ if (cmap != null) {
+ // sanity check
+ if (!cmap.sclass.getName().equals(cname)) {
+ throw new RuntimeException(
+ "Received mapping for class that conflicts with existing mapping " +
+ "[code=" + code + ", oclass=" + cmap.sclass.getName() + ", nclass=" +
+ cname + "]");
+ }
+ return cmap;
+ }
+ // insert null entries for missing mappings
+ cmap = createClassMapping(code, cname);
+ for (int ii = 0, nn = (code + 1) - _classmap.size(); ii < nn; ii++) {
+ _classmap.add(null);
+ }
+ _classmap.set(code, cmap);
+ return cmap;
+ }
+}
diff --git a/src/java/com/threerings/io/UnreliableObjectOutputStream.java b/src/java/com/threerings/io/UnreliableObjectOutputStream.java
new file mode 100644
index 000000000..647f92b51
--- /dev/null
+++ b/src/java/com/threerings/io/UnreliableObjectOutputStream.java
@@ -0,0 +1,112 @@
+//
+// $Id$
+
+package com.threerings.io;
+
+import java.io.IOException;
+import java.io.OutputStream;
+
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+
+/**
+ * Extends {@link ObjectOutputStream} for use in unreliable channels, where we must transmit class
+ * mappings with every object until we are explicitly notified that the receiver has cached the
+ * mappings.
+ */
+public class UnreliableObjectOutputStream extends ObjectOutputStream
+{
+ /**
+ * Constructs an object output stream which will write its data to the supplied target stream.
+ */
+ public UnreliableObjectOutputStream (OutputStream target)
+ {
+ super(target);
+ }
+
+ /**
+ * Sets the reference to the set that will hold the classes for which mappings have been
+ * written.
+ */
+ public void setMappedClasses (Set mappedClasses)
+ {
+ _mappedClasses = mappedClasses;
+ }
+
+ /**
+ * Returns a reference to the set of classes for which mappings have been written.
+ */
+ public Set getMappedClasses ()
+ {
+ return _mappedClasses;
+ }
+
+ /**
+ * Notes that the receiver has received the mappings for a group of classes, and thus that from
+ * now on, only the class codes need be sent.
+ */
+ public void noteMappingsReceived (Collection sclasses)
+ {
+ // sanity check
+ if (_classmap == null) {
+ throw new RuntimeException("Missing class map");
+ }
+
+ // make each class's code positive to signify that we no longer need to send metadata
+ for (Class sclass : sclasses) {
+ ClassMapping cmap = _classmap.get(sclass);
+ if (cmap == null) {
+ throw new RuntimeException("No class mapping for " + sclass.getName());
+ }
+ cmap.code = (short)Math.abs(cmap.code);
+ }
+ }
+
+ @Override // documentation inherited
+ protected ClassMapping createClassMapping (short code, Class sclass, Streamer streamer)
+ {
+ // the negative class code indicates that we must rewrite the metadata for the first
+ // instance in each go-round; when we are notified that the other side has cached the
+ // mapping, we can simply write the (positive) code
+ return new ClassMapping((short)(-code), sclass, streamer);
+ }
+
+ @Override // documentation inherited
+ protected void writeNewClassMapping (ClassMapping cmap)
+ throws IOException
+ {
+ writeClassMapping(cmap.code, cmap.sclass);
+ }
+
+ @Override // documentation inherited
+ protected void writeExistingClassMapping (ClassMapping cmap)
+ throws IOException
+ {
+ // if the other side has received the mapping, we need only send the reference
+ if (cmap.code > 0) {
+ writeShort(cmap.code);
+
+ // likewise if we've written the class once this go-round
+ } else if (_mappedClasses.contains(cmap.sclass)) {
+ writeShort(-cmap.code);
+
+ // otherwise, we must retransmit the mapping
+ } else {
+ writeClassMapping(cmap.code, cmap.sclass);
+ }
+ }
+
+ @Override // documentation inherited
+ protected void writeClassMapping (int code, Class sclass)
+ throws IOException
+ {
+ super.writeClassMapping(code, sclass);
+
+ // note that we've written the mapping
+ _mappedClasses.add(sclass);
+ }
+
+ /** The set of classes for which we have written mappings. */
+ protected Set _mappedClasses = new HashSet();
+}
diff --git a/src/java/com/threerings/presents/client/BlockingCommunicator.java b/src/java/com/threerings/presents/client/BlockingCommunicator.java
index bac16a369..bc86f15b5 100644
--- a/src/java/com/threerings/presents/client/BlockingCommunicator.java
+++ b/src/java/com/threerings/presents/client/BlockingCommunicator.java
@@ -25,20 +25,31 @@ import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.nio.ByteBuffer;
+import java.nio.channels.AsynchronousCloseException;
+import java.nio.channels.DatagramChannel;
+import java.nio.channels.SelectionKey;
+import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.net.ConnectException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+
import com.samskivert.util.LoopingThread;
import com.samskivert.util.Queue;
import com.samskivert.util.StringUtil;
+import com.threerings.io.ByteBufferInputStream;
+import com.threerings.io.ByteBufferOutputStream;
import com.threerings.io.FramedInputStream;
import com.threerings.io.FramingOutputStream;
import com.threerings.io.ObjectInputStream;
import com.threerings.io.ObjectOutputStream;
+import com.threerings.io.UnreliableObjectInputStream;
+import com.threerings.io.UnreliableObjectOutputStream;
import com.threerings.presents.Log;
import com.threerings.presents.data.AuthCodes;
@@ -48,7 +59,10 @@ 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.PingRequest;
+import com.threerings.presents.net.PongResponse;
import com.threerings.presents.net.UpstreamMessage;
+import com.threerings.presents.util.DatagramSequencer;
/**
* The client performs all network I/O on separate threads (one for reading and one for
@@ -107,7 +121,7 @@ public class BlockingCommunicator extends Communicator
// post a logoff message
postMessage(new LogoffRequest());
- // let our reader and writer know that it's time to go
+ // let our readers and writers 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
@@ -127,13 +141,33 @@ public class BlockingCommunicator extends Communicator
// closing our socket and invoking the clientDidLogoff callback
_writer.shutdown();
}
+ if (_datagramWriter != null) {
+ _datagramWriter.shutdown();
+ }
+ if (_datagramReader != null) {
+ _datagramReader.shutdown();
+ }
+ }
+
+ @Override // from Communicator
+ public void gotBootstrap ()
+ {
+ // start the datagram writer thread, if applicable
+ if (_client.getDatagramPorts().length > 0) {
+ _datagramReader = new DatagramReader();
+ _datagramReader.start();
+ }
}
@Override // from Communicator
public void postMessage (UpstreamMessage msg)
{
- // simply append the message to the queue
- _msgq.append(msg);
+ // post as datagram if hinted and possible
+ if (msg.datagram && _datagramWriter != null) {
+ _dataq.append(msg);
+ } else {
+ _msgq.append(msg);
+ }
}
@Override // from Communicator
@@ -279,6 +313,62 @@ public class BlockingCommunicator extends Communicator
}
}
+ /**
+ * Callback called by the datagram reader thread when it goes away.
+ */
+ protected synchronized void datagramReaderDidExit ()
+ {
+ // clear out our reader reference
+ _datagramReader = null;
+
+ if (_datagramWriter == null) {
+ closeDatagramChannel();
+ }
+
+ Log.debug("Datagram reader thread exited.");
+ }
+
+ /**
+ * Callback called by the datagram writer thread when it goes away.
+ */
+ protected synchronized void datagramWriterDidExit ()
+ {
+ // clear out our writer reference
+ _datagramWriter = null;
+ Log.debug("Datagram writer thread exited.");
+
+ closeDatagramChannel();
+ }
+
+ /**
+ * Closes the datagram channel.
+ */
+ protected void closeDatagramChannel ()
+ {
+ if (_selector != null) {
+ try {
+ _selector.close();
+ } catch (IOException ioe) {
+ Log.warning("Error closing selector: " + ioe);
+ }
+ _selector = null;
+ }
+ if (_datagramChannel != null) {
+ Log.debug("Closing datagram socket channel.");
+
+ try {
+ _datagramChannel.close();
+ } catch (IOException ioe) {
+ Log.warning("Error closing datagram socket: " + ioe);
+ }
+ _datagramChannel = null;
+
+ // clear these out because they are probably large and in charge
+ _uout = null;
+ _sequencer = null;
+ }
+ }
+
/**
* Writes the supplied message to the socket.
*/
@@ -316,6 +406,42 @@ public class BlockingCommunicator extends Communicator
updateWriteStamp();
}
+ /**
+ * Sends a datagram over the datagram socket.
+ */
+ protected void sendDatagram (UpstreamMessage msg)
+ throws IOException
+ {
+ // reset the stream and write our connection id and hash placeholder
+ _bout.reset();
+ _uout.writeInt(_client.getConnectionId());
+ _uout.writeLong(0L);
+
+ // write the datagram through the sequencer
+ _sequencer.writeDatagram(msg);
+
+ // flip the buffer and make sure it's not too long
+ ByteBuffer buf = _bout.flip();
+ int size = buf.remaining();
+ if (size > Client.MAX_DATAGRAM_SIZE) {
+ Log.warning("Dropping oversized datagram [size=" + size +
+ ", msg=" + msg + "].");
+ return;
+ }
+
+ // compute the hash
+ buf.position(12);
+ _digest.update(buf);
+ byte[] hash = _digest.digest(_secret);
+
+ // insert the first 64 bits of the hash
+ buf.position(4);
+ buf.put(hash, 0, 8).rewind();
+
+ // send the datagram
+ _datagramChannel.write(buf);
+ }
+
/**
* Makes a note of the time at which we last communicated with the server.
*/
@@ -349,6 +475,36 @@ public class BlockingCommunicator extends Communicator
}
}
+ /**
+ * Reads a datagram from the socket (blocking until a datagram has arrived).
+ */
+ protected DownstreamMessage receiveDatagram ()
+ throws IOException
+ {
+ // clear the buffer and read a datagram
+ _buf.clear();
+ if (_datagramChannel.read(_buf) <= 0) {
+ throw new IOException("No datagram available to read.");
+ }
+
+ // decode through the sequencer
+ try {
+ DownstreamMessage msg = (DownstreamMessage)_sequencer.readDatagram();
+ if (msg == null) {
+ return null; // received out of order
+ }
+ msg.datagram = true;
+ if (debugLogMessages()) {
+ Log.info("DATAGRAM " + msg);
+ }
+ return msg;
+
+ } catch (ClassNotFoundException cnfe) {
+ throw (IOException) new IOException(
+ "Unable to decode incoming datagram.").initCause(cnfe);
+ }
+ }
+
/**
* Callback called by the reader thread when it has parsed a new message from the socket and
* wishes to have it processed.
@@ -385,7 +541,7 @@ public class BlockingCommunicator extends Communicator
public Reader ()
{
}
-
+
protected void willStart ()
{
// first we connect and authenticate with the server
@@ -505,7 +661,7 @@ public class BlockingCommunicator extends 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
-
+
// While it would be nice to be able to handle wacky cases requiring reader-side
// shutdown, doing so causes consternation on the other end's writer which suddenly
// loses its connection. So, we rely on the writer side to take us down.
@@ -562,7 +718,197 @@ public class BlockingCommunicator extends Communicator
}
}
- /** This is used to terminate the writer thread. */
+ /**
+ * Handles the general flow of reading datagrams.
+ */
+ protected class DatagramReader extends LoopingThread
+ {
+ protected void willStart ()
+ {
+ try {
+ connect();
+ } catch (IOException ioe) {
+ Log.warning("Failed to open datagram channel [error=" + ioe + "].");
+ shutdown();
+ }
+ }
+
+ protected void connect ()
+ throws IOException
+ {
+ // create a selector to be used only for the initial connection
+ _selector = Selector.open();
+
+ // create and register the channel
+ _datagramChannel = DatagramChannel.open();
+ _datagramChannel.configureBlocking(false);
+ _datagramChannel.register(_selector, SelectionKey.OP_READ, null);
+
+ // create the message digest
+ try {
+ _digest = MessageDigest.getInstance("MD5");
+ } catch (NoSuchAlgorithmException nsae) {
+ Log.warning("Missing MD5 algorithm.");
+ shutdown();
+ return;
+ }
+ _secret = _client.getCredentials().getDatagramSecret().getBytes("UTF-8");
+
+ // create our various streams
+ _bout = new ByteBufferOutputStream();
+ _uout = new UnreliableObjectOutputStream(_bout);
+ ByteBufferInputStream bin = new ByteBufferInputStream(_buf);
+ UnreliableObjectInputStream uin = new UnreliableObjectInputStream(bin);
+ uin.setClassLoader(_loader);
+
+ // create the datagram sequencer
+ _sequencer = new DatagramSequencer(uin, _uout);
+
+ // try each port in turn
+ int cport = -1;
+ for (int port : _client.getDatagramPorts()) {
+ boolean connected = connect(port);
+ if (!isRunning()) {
+ return; // cancelled
+
+ } else if (connected) {
+ cport = port;
+ break;
+ }
+ }
+
+ // close the selector and return the channel to blocking mode
+ _selector.close();
+ _selector = null;
+ _datagramChannel.configureBlocking(true);
+
+ // check if we managed to establish a connection
+ if (cport > 0) {
+ Log.info("Datagram connection established [port=" + cport + "].");
+
+ // start up the writer thread
+ _datagramWriter = new DatagramWriter();
+ _datagramWriter.start();
+
+ } else {
+ Log.info("Failed to establish datagram connection.");
+ shutdown();
+ }
+ }
+
+ protected boolean connect (int port)
+ throws IOException
+ {
+ _datagramChannel.connect(new InetSocketAddress(_client.getHostname(), port));
+ for (int ii = 0; ii < DATAGRAM_ATTEMPTS_PER_PORT; ii++) {
+ // send a ping datagram
+ sendDatagram(new PingRequest());
+
+ // wait for a response
+ int resp = _selector.select(DATAGRAM_RESPONSE_WAIT);
+ if (!isRunning()) {
+ return false; // cancelled
+
+ } else if (resp > 0) {
+ DownstreamMessage msg = receiveDatagram();
+ return true;
+ }
+ }
+ _datagramChannel.disconnect();
+ return false;
+ }
+
+ protected void iterate ()
+ {
+ DownstreamMessage msg = null;
+
+ try {
+ // read the next message from the socket
+ msg = receiveDatagram();
+
+ // process the message if it wasn't dropped
+ if (msg != null) {
+ processMessage(msg);
+ }
+
+ } catch (AsynchronousCloseException ace) {
+ // 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("Datagram reader thread woken up in time to die.");
+
+ } catch (IOException ioe) {
+ Log.warning("Error receiving datagram [error=" + ioe + "].");
+
+ } catch (Exception e) {
+ Log.warning("Error processing message [msg=" + msg + ", error=" + e + "].");
+ }
+ }
+
+ protected void handleIterateFailure (Exception e)
+ {
+ Log.warning("Uncaught exception in datagram reader thread.");
+ Log.logStackTrace(e);
+ }
+
+ protected void didShutdown ()
+ {
+ datagramReaderDidExit();
+ }
+
+ protected void kick ()
+ {
+ // if we have a selector, wake it up
+ if (_selector != null) {
+ _selector.wakeup();
+ }
+ }
+ }
+
+ /**
+ * Handles the general flow of writing datagrams.
+ */
+ protected class DatagramWriter extends LoopingThread
+ {
+ protected void iterate ()
+ {
+ // fetch the next message from the queue
+ UpstreamMessage msg = _dataq.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
+ sendDatagram(msg);
+
+ } catch (IOException ioe) {
+ Log.warning("Error sending datagram [error=" + ioe + "].");
+ }
+ }
+
+ protected void handleIterateFailure (Exception e)
+ {
+ Log.warning("Uncaught exception in datagram writer thread.");
+ Log.logStackTrace(e);
+ }
+
+ protected void didShutdown ()
+ {
+ datagramWriterDidExit();
+ }
+
+ protected void kick ()
+ {
+ // post a bogus message to the outgoing queue to ensure that the writer thread notices
+ // that it's time to go
+ _dataq.append(new TerminationMessage());
+ }
+ }
+
+ /** This is used to terminate the writer threads. */
protected static class TerminationMessage extends UpstreamMessage
{
}
@@ -570,9 +916,16 @@ public class BlockingCommunicator extends Communicator
protected Reader _reader;
protected Writer _writer;
+ protected DatagramReader _datagramReader;
+ protected DatagramWriter _datagramWriter;
+
protected SocketChannel _channel;
protected Queue _msgq = new Queue();
+ protected Selector _selector;
+ protected DatagramChannel _datagramChannel;
+ protected Queue _dataq = new Queue();
+
protected long _lastWrite;
protected Exception _logonError;
@@ -584,6 +937,23 @@ public class BlockingCommunicator extends Communicator
protected FramedInputStream _fin;
protected ObjectInputStream _oin;
+ /** We use these to write our upstream datagrams. */
+ protected ByteBufferOutputStream _bout;
+ protected UnreliableObjectOutputStream _uout;
+ protected MessageDigest _digest;
+ protected byte[] _secret;
+
+ /** We use these to read our downstream datagrams. */
+ protected ByteBuffer _buf = ByteBuffer.allocateDirect(Client.MAX_DATAGRAM_SIZE);
+
+ protected DatagramSequencer _sequencer;
+
protected ClientDObjectMgr _omgr;
protected ClassLoader _loader;
+
+ /** The number of times per port to try to establish a datagram "connection." */
+ protected static final int DATAGRAM_ATTEMPTS_PER_PORT = 10;
+
+ /** The number of milliseconds to wait for a response datagram. */
+ protected static final long DATAGRAM_RESPONSE_WAIT = 1000L;
}
diff --git a/src/java/com/threerings/presents/client/Client.java b/src/java/com/threerings/presents/client/Client.java
index af2e5643d..a64fff10b 100644
--- a/src/java/com/threerings/presents/client/Client.java
+++ b/src/java/com/threerings/presents/client/Client.java
@@ -53,6 +53,12 @@ public class Client
/** The default ports on which the server listens for client connections. */
public static final int[] DEFAULT_SERVER_PORTS = { 47624 };
+ /** The default ports on which the server listens for datagrams. */
+ public static final int[] DEFAULT_DATAGRAM_PORTS = { };
+
+ /** The maximum size of a datagram. */
+ public static final int MAX_DATAGRAM_SIZE = 1500;
+
/**
* Constructs a client object with the supplied credentials and RunQueue. The creds will be
* used to authenticate with any server to which this client attempts to connect. The RunQueue
@@ -114,9 +120,22 @@ public class Client
* @see #moveToServer
*/
public void setServer (String hostname, int[] ports)
+ {
+ setServer(hostname, ports, new int[0]);
+ }
+
+ /**
+ * Configures the client to communicate with the server on the supplied hostname, set of
+ * ports (which will be tried in succession), and datagram ports.
+ *
+ * @see #logon
+ * @see #moveToServer
+ */
+ public void setServer (String hostname, int[] ports, int[] datagramPorts)
{
_hostname = hostname;
_ports = ports;
+ _datagramPorts = datagramPorts;
}
/**
@@ -144,6 +163,15 @@ public class Client
return _ports;
}
+ /**
+ * Returns the ports on the server to which the client can send datagrams. Returns an empty
+ * array if datagrams are not supported.
+ */
+ public int[] getDatagramPorts ()
+ {
+ return _datagramPorts;
+ }
+
/**
* Returns the credentials with which this client is currently configured to connect to the
* server.
@@ -230,6 +258,15 @@ public class Client
omgr.registerFlushDelay(objclass, delay);
}
+ /**
+ * Returns the unique id of the client's connection to the server. It is only valid for the
+ * duration of the session.
+ */
+ public int getConnectionId ()
+ {
+ return _connectionId;
+ }
+
/**
* Returns the oid of the client object associated with this session. It is only valid for the
* duration of the session.
@@ -416,9 +453,29 @@ public class Client
* other server, or if the move failed.
*/
public void moveToServer (String hostname, int[] ports, InvocationService.ConfirmListener obs)
+ {
+ moveToServer(hostname, ports, new int[0], obs);
+ }
+
+ /**
+ * Transitions a logged on client from its current server to the specified new server.
+ * Currently this simply logs the client off of its current server (if it is logged on) and
+ * logs it onto the new server, but in the future we may aim to do something fancier.
+ *
+ *
If we fail to connect to the new server, the client will not be automatically
+ * reconnected to the old server. It will be in a logged off state. However, it will be
+ * reconfigured with the hostname and ports of the old server so that the caller can notify the
+ * user of the failure and then simply call {@link #logon} to attempt to reconnect to the old
+ * server.
+ *
+ * @param observer an observer that will be notified when we have successfully logged onto the
+ * other server, or if the move failed.
+ */
+ public void moveToServer (
+ String hostname, int[] ports, int[] datagramPorts, InvocationService.ConfirmListener obs)
{
// the server switcher will take care of everything for us
- new ServerSwitcher(hostname, ports, obs).switchServers();
+ new ServerSwitcher(hostname, ports, datagramPorts, obs).switchServers();
}
/**
@@ -501,8 +558,12 @@ public class Client
_omgr = omgr;
// extract bootstrap information
+ _connectionId = data.connectionId;
_cloid = data.clientOid;
+ // notify the communicator that we got our bootstrap data
+ _comm.gotBootstrap();
+
// initialize our invocation director
_invdir.init(omgr, _cloid, this);
@@ -676,7 +737,7 @@ public class Client
_comm = null;
_omgr = null;
_clobj = null;
- _cloid = -1;
+ _connectionId = _cloid = -1;
_standalone = false;
// and let our invocation director know we're logged off
@@ -799,10 +860,19 @@ public class Client
/** Handles the process of switching between servers. See {@link #moveToServer}. */
protected class ServerSwitcher extends ClientAdapter
{
- public ServerSwitcher (String hostname, int[] ports, InvocationService.ConfirmListener obs)
+ public ServerSwitcher (
+ String hostname, int[] ports, InvocationService.ConfirmListener obs)
+ {
+ this(hostname, ports, new int[0], obs);
+ }
+
+ public ServerSwitcher (
+ String hostname, int[] ports, int[] datagramPorts,
+ InvocationService.ConfirmListener obs)
{
_hostname = hostname;
_ports = ports;
+ _datagramPorts = datagramPorts;
_observer = obs;
}
@@ -818,6 +888,7 @@ public class Client
// attempt fails
_oldHostname = Client.this._hostname;
_oldPorts = Client.this._ports;
+ _oldDatagramPorts = Client.this._datagramPorts;
// otherwise logoff and wait for all of our callbacks to clear
logoff(true);
@@ -827,11 +898,12 @@ public class Client
public void clientDidClear (Client client)
{
// configure the client to point to the new server and logon
- setServer(_hostname, _ports);
+ setServer(_hostname, _ports, _datagramPorts);
if (!logon()) {
log.warning("logon() failed during server switch? [hostname=" + _hostname +
- ", ports=" + StringUtil.toString(_ports) + "].");
+ ", ports=" + StringUtil.toString(_ports) + ", datagramPorts=" +
+ StringUtil.toString(_datagramPorts) + "].");
clientFailedToLogon(Client.this, new Exception("logon() failed?"));
}
}
@@ -848,7 +920,7 @@ public class Client
{
removeClientObserver(this);
if (_oldHostname != null) { // restore our previous server and ports
- setServer(_oldHostname, _oldPorts);
+ setServer(_oldHostname, _oldPorts, _oldDatagramPorts);
}
if (_observer != null) {
_observer.requestFailed((cause instanceof LogonException) ?
@@ -858,6 +930,7 @@ public class Client
protected String _hostname, _oldHostname;
protected int[] _ports, _oldPorts;
+ protected int[] _datagramPorts, _oldDatagramPorts;
protected InvocationService.ConfirmListener _observer;
}
@@ -876,6 +949,9 @@ public class Client
/** The data associated with our authentication response. */
protected AuthResponseData _authData;
+ /** The unique id of our connection. */
+ protected int _connectionId = -1;
+
/** Our client distribted object id. */
protected int _cloid = -1;
@@ -891,6 +967,9 @@ public class Client
/** The ports on which we connect to the game server. */
protected int[] _ports;
+ /** The server ports to which we can send datagrams. */
+ protected int[] _datagramPorts;
+
/** Our list of client observers. */
protected ObserverList _observers =
new ObserverList(ObserverList.SAFE_IN_ORDER_NOTIFY);
diff --git a/src/java/com/threerings/presents/client/Communicator.java b/src/java/com/threerings/presents/client/Communicator.java
index c2f64ed37..20c852148 100644
--- a/src/java/com/threerings/presents/client/Communicator.java
+++ b/src/java/com/threerings/presents/client/Communicator.java
@@ -47,6 +47,11 @@ public abstract class Communicator
*/
public abstract void logoff ();
+ /**
+ * Notifies the communicator that the client has received its bootstrap data.
+ */
+ public abstract void gotBootstrap ();
+
/**
* Queues up the specified message for delivery upstream.
*/
diff --git a/src/java/com/threerings/presents/net/BootstrapData.java b/src/java/com/threerings/presents/net/BootstrapData.java
index ad37381cb..56ab8b57a 100644
--- a/src/java/com/threerings/presents/net/BootstrapData.java
+++ b/src/java/com/threerings/presents/net/BootstrapData.java
@@ -34,6 +34,9 @@ import com.threerings.presents.data.InvocationMarshaller;
*/
public class BootstrapData extends SimpleStreamableObject
{
+ /** The unique id of the client's connection (used to address datagrams). */
+ public int connectionId;
+
/** The oid of this client's associated distributed object. */
public int clientOid;
diff --git a/src/java/com/threerings/presents/net/Credentials.java b/src/java/com/threerings/presents/net/Credentials.java
index 535400162..ffd9095cb 100644
--- a/src/java/com/threerings/presents/net/Credentials.java
+++ b/src/java/com/threerings/presents/net/Credentials.java
@@ -63,6 +63,14 @@ public abstract class Credentials implements Streamable
_username = name;
}
+ /**
+ * Returns a string to use in a hash on the datagram contents to authenticate client datagrams.
+ */
+ public String getDatagramSecret ()
+ {
+ return "";
+ }
+
// documentation inherited
public int hashCode ()
{
diff --git a/src/java/com/threerings/presents/net/DownstreamMessage.java b/src/java/com/threerings/presents/net/DownstreamMessage.java
index 0eca0768a..db237cf75 100644
--- a/src/java/com/threerings/presents/net/DownstreamMessage.java
+++ b/src/java/com/threerings/presents/net/DownstreamMessage.java
@@ -37,6 +37,12 @@ public abstract class DownstreamMessage extends SimpleStreamableObject
*/
public short messageId = -1;
+ /**
+ * When sending, this acts as a hint that the message may be transmitted as a datagram.
+ * When receiving, it indicates that the message was received as a datagram.
+ */
+ public transient boolean datagram;
+
/**
* Generates a string representation of this instance.
*/
diff --git a/src/java/com/threerings/presents/net/UpstreamMessage.java b/src/java/com/threerings/presents/net/UpstreamMessage.java
index 1d5eb4b3c..965f88f04 100644
--- a/src/java/com/threerings/presents/net/UpstreamMessage.java
+++ b/src/java/com/threerings/presents/net/UpstreamMessage.java
@@ -41,6 +41,12 @@ public abstract class UpstreamMessage extends SimpleStreamableObject
/** A timestamp indicating when this upstream message was received. */
public transient long received;
+ /**
+ * When sending, this acts as a hint that the message may be transmitted as a datagram.
+ * When receiving, it indicates that the message was received as a datagram.
+ */
+ public transient boolean datagram;
+
/**
* Each upstream message derived class must provide a zero argument
* constructor so that it can be unserialized when read from the
diff --git a/src/java/com/threerings/presents/net/UsernamePasswordCreds.java b/src/java/com/threerings/presents/net/UsernamePasswordCreds.java
index d3834657c..fb620c9a7 100644
--- a/src/java/com/threerings/presents/net/UsernamePasswordCreds.java
+++ b/src/java/com/threerings/presents/net/UsernamePasswordCreds.java
@@ -47,6 +47,12 @@ public class UsernamePasswordCreds extends Credentials
return _password;
}
+ @Override // documentation inherited
+ public String getDatagramSecret ()
+ {
+ return _password;
+ }
+
// documentation inherited
public int hashCode ()
{
diff --git a/src/java/com/threerings/presents/server/PresentsClient.java b/src/java/com/threerings/presents/server/PresentsClient.java
index dcec78ddb..265905d80 100644
--- a/src/java/com/threerings/presents/server/PresentsClient.java
+++ b/src/java/com/threerings/presents/server/PresentsClient.java
@@ -343,7 +343,7 @@ public class PresentsClient
{
// we'll be keeping this bad boy
_clobj = clobj;
-
+
// if our connection was closed while we were resolving our client object, then just
// abandon ship
if (getConnection() == null) {
@@ -670,7 +670,10 @@ public class PresentsClient
*/
protected void populateBootstrapData (BootstrapData data)
{
- // give them the client object id
+ // give them the connection id
+ data.connectionId = _conn.getConnectionId();
+
+ // and the client object id
data.clientOid = _clobj.getOid();
// fill in the list of bootstrap services
@@ -977,7 +980,9 @@ public class PresentsClient
{
// send a pong response
PingRequest req = (PingRequest)msg;
- client.safePostMessage(new PongResponse(req.getUnpackStamp()));
+ PongResponse resp = new PongResponse(req.getUnpackStamp());
+ resp.datagram = req.datagram;
+ client.safePostMessage(resp);
}
}
diff --git a/src/java/com/threerings/presents/server/PresentsServer.java b/src/java/com/threerings/presents/server/PresentsServer.java
index 35842b41a..faca86c01 100644
--- a/src/java/com/threerings/presents/server/PresentsServer.java
+++ b/src/java/com/threerings/presents/server/PresentsServer.java
@@ -174,7 +174,7 @@ public class PresentsServer
invoker.start();
// create our connection manager
- conmgr = new ConnectionManager(getListenPorts());
+ conmgr = new ConnectionManager(getListenPorts(), getDatagramPorts());
conmgr.setAuthenticator(createAuthenticator());
// create our client manager
@@ -236,6 +236,14 @@ public class PresentsServer
return Client.DEFAULT_SERVER_PORTS;
}
+ /**
+ * Returns the ports on which the connection manager will listen for datagrams.
+ */
+ protected int[] getDatagramPorts ()
+ {
+ return Client.DEFAULT_DATAGRAM_PORTS;
+ }
+
/**
* Starts up all of the server services and enters the main server event loop.
*/
@@ -387,7 +395,7 @@ public class PresentsServer
protected void invokerDidShutdown ()
{
}
-
+
/** Our interval that generates "state of server" reports. */
protected Interval _reportInterval;
diff --git a/src/java/com/threerings/presents/server/net/Connection.java b/src/java/com/threerings/presents/server/net/Connection.java
index c2953ac6d..911fdb1a5 100644
--- a/src/java/com/threerings/presents/server/net/Connection.java
+++ b/src/java/com/threerings/presents/server/net/Connection.java
@@ -25,20 +25,30 @@ import java.io.EOFException;
import java.io.IOException;
import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+
import com.samskivert.util.StringUtil;
+import com.threerings.io.ByteBufferInputStream;
+import com.threerings.io.ByteBufferOutputStream;
import com.threerings.io.FramedInputStream;
import com.threerings.io.FramingOutputStream;
import com.threerings.io.ObjectInputStream;
import com.threerings.io.ObjectOutputStream;
+import com.threerings.io.UnreliableObjectInputStream;
+import com.threerings.io.UnreliableObjectOutputStream;
import com.threerings.presents.Log;
import com.threerings.presents.net.DownstreamMessage;
import com.threerings.presents.net.PingRequest;
import com.threerings.presents.net.UpstreamMessage;
+import com.threerings.presents.util.DatagramSequencer;
/**
* The base connection class implements the net event handler interface and processes raw incoming
@@ -63,6 +73,7 @@ public abstract class Connection implements NetEventHandler
_selkey = selkey;
_channel = channel;
_lastEvent = createStamp;
+ _connectionId = ++_lastConnectionId;
}
/**
@@ -85,6 +96,14 @@ public abstract class Connection implements NetEventHandler
}
}
+ /**
+ * Returns the connection's unique identifier.
+ */
+ public int getConnectionId ()
+ {
+ return _connectionId;
+ }
+
/**
* Returns the selection key associated with our socket channel.
*/
@@ -110,6 +129,27 @@ public abstract class Connection implements NetEventHandler
return (_channel == null) ? null : _channel.socket().getInetAddress();
}
+ /**
+ * Returns the address to which datagrams should be sent or null if no datagram address has
+ * been established.
+ */
+ public InetSocketAddress getDatagramAddress ()
+ {
+ return _datagramAddress;
+ }
+
+ /**
+ * Sets the secret string used to authenticate datagrams from the client.
+ */
+ public void setDatagramSecret (String secret)
+ {
+ try {
+ _datagramSecret = secret.getBytes("UTF-8");
+ } catch (Exception e) {
+ _datagramSecret = new byte[0]; // shouldn't happen
+ }
+ }
+
/**
* Returns true if this connection is closed.
*/
@@ -205,6 +245,15 @@ public abstract class Connection implements NetEventHandler
_oout = oout;
}
+ /**
+ * Returns a reference to the connection's datagram sequencer. This should only be called by
+ * the connection manager.
+ */
+ protected DatagramSequencer getDatagramSequencer ()
+ {
+ return _sequencer;
+ }
+
/**
* Closes the socket associated with this connection. This happens when we receive EOF, are
* requested to close down or when our connection fails.
@@ -285,6 +334,59 @@ public abstract class Connection implements NetEventHandler
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;
+ msg.datagram = true;
+ _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)
{
@@ -316,13 +418,24 @@ public abstract class Connection implements NetEventHandler
protected long _lastEvent;
+ protected int _connectionId;
+
protected FramedInputStream _fin;
protected ObjectInputStream _oin;
protected ObjectOutputStream _oout;
+ protected InetSocketAddress _datagramAddress;
+ protected byte[] _datagramSecret;
+
+ protected MessageDigest _digest;
+ protected DatagramSequencer _sequencer;
+
protected MessageHandler _handler;
protected ClassLoader _loader;
+ /** The last connection id assigned. */
+ protected static int _lastConnectionId;
+
/** The number of milliseconds beyond the ping interval that we allow a client's network
* connection to be idle before we forcibly disconnect them. */
protected static final long LATENCY_GRACE = 30 * 1000L;
diff --git a/src/java/com/threerings/presents/server/net/ConnectionManager.java b/src/java/com/threerings/presents/server/net/ConnectionManager.java
index 6d9f498ea..d2bbe1168 100644
--- a/src/java/com/threerings/presents/server/net/ConnectionManager.java
+++ b/src/java/com/threerings/presents/server/net/ConnectionManager.java
@@ -21,9 +21,11 @@
package com.threerings.presents.server.net;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.Channel;
+import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
@@ -44,10 +46,12 @@ import com.samskivert.util.*;
import com.threerings.io.FramingOutputStream;
import com.threerings.io.ObjectOutputStream;
+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.util.DatagramSequencer;
import com.threerings.presents.server.Authenticator;
import com.threerings.presents.server.PresentsServer;
@@ -79,8 +83,19 @@ public class ConnectionManager extends LoopingThread
*/
public ConnectionManager (int[] ports)
throws IOException
+ {
+ this(ports, new int[0]);
+ }
+
+ /**
+ * Constructs and initialized a connection manager (binding socket on which it will listen for
+ * client connections to each of the specified ports).
+ */
+ public ConnectionManager (int[] ports, int[] datagramPorts)
+ throws IOException
{
_ports = ports;
+ _datagramPorts = datagramPorts;
_selector = SelectorProvider.provider().openSelector();
// create our stats record
@@ -311,8 +326,27 @@ public class ConnectionManager extends LoopingThread
return;
}
- // we'll use this for sending messages to clients
+ // open up the datagram ports as well
+ for (int port : _datagramPorts) {
+ try {
+ // create a channel and add it to the select set
+ _datagramChannel = DatagramChannel.open();
+ _datagramChannel.configureBlocking(false);
+
+ InetSocketAddress isa = new InetSocketAddress(port);
+ _datagramChannel.socket().bind(isa);
+ registerChannel(_datagramChannel);
+ log.info("Server accepting datagrams on " + isa + ".");
+
+ } catch (IOException ioe) {
+ log.log(Level.WARNING, "Failure opening datagram channel on port '" +
+ port + "'.", ioe);
+ }
+ }
+
+ // 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) {
@@ -340,6 +374,30 @@ public class ConnectionManager extends LoopingThread
});
}
+ /** Helper function for {@link #willStart}. */
+ protected void registerChannel (final DatagramChannel listener)
+ throws IOException
+ {
+ SelectionKey sk = listener.register(_selector, SelectionKey.OP_READ);
+ _handlers.put(sk, new NetEventHandler() {
+ public int handleEvent (long when) {
+ return readDatagram(listener, when);
+ }
+ public boolean checkIdle (long now) {
+ return false; // we're never idle
+ }
+ });
+ }
+
+ /**
+ * Returns a reference to the output stream used to flatten messages into byte arrays. Should
+ * only be called by {@link Connection}.
+ */
+ protected ByteArrayOutputStream getFlattener ()
+ {
+ return _flattener;
+ }
+
/**
* Performs the select loop. This is the body of the conmgr thread.
*/
@@ -385,6 +443,11 @@ public class ConnectionManager extends LoopingThread
// replace the mapping in the handlers table from the old conn with the new one
_handlers.put(selkey, rconn);
+ // add a mapping for the connection id and set the datagram secret
+ _connections.put(rconn.getConnectionId(), rconn);
+ rconn.setDatagramSecret(
+ conn.getAuthRequest().getCredentials().getDatagramSecret());
+
// transfer any overflow queue for that connection
OverflowQueue oflowHandler = _oflowqs.remove(conn);
if (oflowHandler != null) {
@@ -526,6 +589,11 @@ public class ConnectionManager extends LoopingThread
// otherwise write the message out to the client directly
writeMessage(conn, tup.right, _oflowHandler);
}
+
+ // send any datagrams
+ while ((tup = _dataq.getNonBlocking()) != null) {
+ writeDatagram(tup.left, tup.right);
+ }
}
/**
@@ -595,6 +663,29 @@ public class ConnectionManager extends LoopingThread
return fully;
}
+ /**
+ * Sends a datagram to the specified connection.
+ *
+ * @return true if the datagram was sent, false if we failed to send for any reason.
+ */
+ protected boolean writeDatagram (Connection conn, byte[] data)
+ {
+ InetSocketAddress target = conn.getDatagramAddress();
+ if (target == null) {
+ log.warning("No address to send datagram [conn=" + conn + "].");
+ return false;
+ }
+
+ _databuf.clear();
+ _databuf.put(data).flip();
+ try {
+ return _datagramChannel.send(_databuf, target) > 0;
+ } catch (IOException ioe) {
+ log.log(Level.WARNING, "Failed to send datagram.", ioe);
+ return false;
+ }
+ }
+
/** Called by {@link #writeMessage} and friends when they write data over the network. */
protected final synchronized void noteWrite (int msgs, int bytes)
{
@@ -624,6 +715,11 @@ public class ConnectionManager extends LoopingThread
log.log(Level.WARNING, "Failed to close listening socket.", ioe);
}
+ // and the datagram socket, if any
+ if (_datagramChannel != null) {
+ _datagramChannel.socket().close();
+ }
+
// report if there's anything left on the outgoing message queue
if (_outq.size() > 0) {
log.warning("Connection Manager failed to deliver " + _outq.size() + " message(s).");
@@ -684,6 +780,51 @@ public class ConnectionManager extends LoopingThread
}
}
+ /**
+ * Called by our net event handler when a datagram is ready to be read from the channel.
+ *
+ * @return the size of the datagram.
+ */
+ protected int readDatagram (DatagramChannel listener, long when)
+ {
+ InetSocketAddress source;
+ _databuf.clear();
+ try {
+ source = (InetSocketAddress)listener.receive(_databuf);
+ } catch (IOException ioe) {
+ log.log(Level.WARNING, "Failure receiving datagram.", ioe);
+ return 0;
+ }
+
+ // make sure we actually read a packet
+ if (source == null) {
+ log.info("Psych! Got READ_READY, but no datagram.");
+ return 0;
+ }
+
+ // flip the buffer and record the size (which must be at least 14 to contain the connection
+ // id, authentication hash, and a class reference)
+ int size = _databuf.flip().remaining();
+ if (size < 14) {
+ log.warning("Received undersized datagram [source=" + source +
+ ", size=" + size + "].");
+ return 0;
+ }
+
+ // the first four bytes are the connection id
+ int connectionId = _databuf.getInt();
+ Connection conn = _connections.get(connectionId);
+ if (conn != null) {
+ conn.handleDatagram(source, _databuf, when);
+ } else {
+ log.warning("Received datagram for unknown connection [id=" + connectionId +
+ ", source=" + source + "].");
+ }
+
+ // return the size of the datagram
+ return size;
+ }
+
/**
* Called by a connection when it has a downstream message that needs to be delivered.
* Note: this method is called as a result of a call to {@link Connection#postMessage}
@@ -708,6 +849,12 @@ public class ConnectionManager extends LoopingThread
}
try {
+ // send it as a datagram if hinted and possible
+ if (msg.datagram && conn.getDatagramAddress() != null) {
+ postDatagram(conn, msg);
+ return;
+ }
+
_framer.resetFrame();
// flatten this message using the connection's output stream
@@ -731,14 +878,34 @@ public class ConnectionManager extends LoopingThread
}
}
+ /**
+ * Helper function for {@link #postMessage}; handles posting the message as a datagram.
+ */
+ void postDatagram (Connection conn, DownstreamMessage msg)
+ throws Exception
+ {
+ _flattener.reset();
+
+ // flatten the message using the connection's sequencer
+ DatagramSequencer sequencer = conn.getDatagramSequencer();
+ sequencer.writeDatagram(msg);
+
+ // extract as a byte array
+ byte[] data = _flattener.toByteArray();
+
+ // slap it on the queue
+ _dataq.append(new Tuple(conn, data));
+ }
+
/**
* Called by a connection if it experiences a network failure.
*/
void connectionFailed (Connection conn, IOException ioe)
{
- // remove this connection from our mapping (it is automatically removed from the Selector
+ // remove this connection from our mappings (it is automatically removed from the Selector
// when the socket is closed)
_handlers.remove(conn.getSelectionKey());
+ _connections.remove(conn.getConnectionId());
_oflowqs.remove(conn);
synchronized (this) {
_stats.disconnects++;
@@ -753,9 +920,10 @@ public class ConnectionManager extends LoopingThread
*/
void connectionClosed (Connection conn)
{
- // remove this connection from our mapping (it is automatically removed from the Selector
+ // remove this connection from our mappings (it is automatically removed from the Selector
// when the socket is closed)
_handlers.remove(conn.getSelectionKey());
+ _connections.remove(conn.getConnectionId());
_oflowqs.remove(conn);
// let our observers know what's up
@@ -860,10 +1028,11 @@ public class ConnectionManager extends LoopingThread
protected int _msgs, _partials;
}
- protected int[] _ports;
+ protected int[] _ports, _datagramPorts;
protected Authenticator _author;
protected Selector _selector;
protected ServerSocketChannel _ssocket;
+ protected DatagramChannel _datagramChannel;
protected ResultListener