diff --git a/src/java/com/threerings/io/FramingOutputStream.java b/src/java/com/threerings/io/FramingOutputStream.java
index b1d9c0ecd..a7e4c8119 100644
--- a/src/java/com/threerings/io/FramingOutputStream.java
+++ b/src/java/com/threerings/io/FramingOutputStream.java
@@ -44,7 +44,7 @@ public class FramingOutputStream extends OutputStream
{
public FramingOutputStream ()
{
- _buffer = ByteBuffer.allocate(INITIAL_BUFFER_SIZE);
+ _buffer = ByteBuffer.allocateDirect(INITIAL_BUFFER_SIZE);
_buffer.put(HEADER_PAD);
}
@@ -99,7 +99,7 @@ public class FramingOutputStream extends OutputStream
if (ncapacity > ocapacity) {
// increase the buffer size in large increments
ncapacity = Math.max(ocapacity << 1, ncapacity);
- ByteBuffer newbuf = ByteBuffer.allocate(ncapacity);
+ ByteBuffer newbuf = ByteBuffer.allocateDirect(ncapacity);
newbuf.put((ByteBuffer)_buffer.flip());
_buffer = newbuf;
}
diff --git a/src/java/com/threerings/presents/client/Communicator.java b/src/java/com/threerings/presents/client/Communicator.java
index fbcaec9f9..396965ca3 100644
--- a/src/java/com/threerings/presents/client/Communicator.java
+++ b/src/java/com/threerings/presents/client/Communicator.java
@@ -316,8 +316,7 @@ public class Communicator
try {
ByteBuffer buffer = _fout.frameAndReturnBuffer();
if (buffer.limit() > 4096) {
- String txt = StringUtil.truncate(
- String.valueOf(msg), 80, "...");
+ String txt = StringUtil.truncate(String.valueOf(msg), 80, "...");
Log.info("Whoa, writin' a big one [msg=" + txt +
", size=" + buffer.limit() + "].");
}
diff --git a/src/java/com/threerings/presents/server/Authenticator.java b/src/java/com/threerings/presents/server/Authenticator.java
index f9d57a75c..eda0a38a0 100644
--- a/src/java/com/threerings/presents/server/Authenticator.java
+++ b/src/java/com/threerings/presents/server/Authenticator.java
@@ -53,8 +53,7 @@ public abstract class Authenticator
* off a task on the {@link PresentsServer#invoker} to perform the
* authentication. When the authentication is complete, the
* authenticator implementation should call {@link
- * #connectionWasAuthenticated} from the distributed object
- * thread.
+ * #connectionWasAuthenticated}.
*/
public abstract void authenticateConnection (AuthingConnection conn);
@@ -62,9 +61,7 @@ public abstract class Authenticator
* This is called by authenticator implementations when they have
* completed the authentication process. It will deliver the response
* to the user and let the connection manager know to report the
- * authentication if it succeeded. Note: this method should
- * only be called on the distributed object thread as all messages
- * being sent to the client should originate therefrom.
+ * authentication if it succeeded.
*/
protected void connectionWasAuthenticated (
AuthingConnection conn, AuthResponse rsp)
diff --git a/src/java/com/threerings/presents/server/DummyAuthenticator.java b/src/java/com/threerings/presents/server/DummyAuthenticator.java
index db16643e3..823d10089 100644
--- a/src/java/com/threerings/presents/server/DummyAuthenticator.java
+++ b/src/java/com/threerings/presents/server/DummyAuthenticator.java
@@ -35,15 +35,11 @@ public class DummyAuthenticator extends Authenticator
/**
* Accept all authentication requests.
*/
- public void authenticateConnection (final AuthingConnection conn)
+ public void authenticateConnection (AuthingConnection conn)
{
Log.info("Accepting request: " + conn.getAuthRequest());
- PresentsServer.omgr.postRunnable(new Runnable() {
- public void run () {
- AuthResponseData rdata = new AuthResponseData();
- rdata.code = AuthResponseData.SUCCESS;
- connectionWasAuthenticated(conn, new AuthResponse(rdata));
- }
- });
+ AuthResponseData rdata = new AuthResponseData();
+ rdata.code = AuthResponseData.SUCCESS;
+ connectionWasAuthenticated(conn, new AuthResponse(rdata));
}
}
diff --git a/src/java/com/threerings/presents/server/PresentsClient.java b/src/java/com/threerings/presents/server/PresentsClient.java
index 2814803c0..30e4066d2 100644
--- a/src/java/com/threerings/presents/server/PresentsClient.java
+++ b/src/java/com/threerings/presents/server/PresentsClient.java
@@ -758,17 +758,6 @@ public class PresentsClient
}
}
- /** Callable from non-dobjmgr thread, this queues up a runnable on the
- * dobjmgr thread to post the supplied message to this client. */
- protected final void safePostMessage (final DownstreamMessage msg)
- {
- PresentsServer.omgr.postRunnable(new Runnable() {
- public void run () {
- postMessage(msg);
- }
- });
- }
-
/** Queues a message for delivery to the client. */
protected final boolean postMessage (DownstreamMessage msg)
{
@@ -867,7 +856,7 @@ public class PresentsClient
// post a response to the client letting them know that we
// will no longer send them events regarding this object
- client.safePostMessage(new UnsubscribeResponse(oid));
+ client.postMessage(new UnsubscribeResponse(oid));
}
}
@@ -901,7 +890,7 @@ public class PresentsClient
{
// send a pong response
PingRequest req = (PingRequest)msg;
- client.safePostMessage(new PongResponse(req.getUnpackStamp()));
+ client.postMessage(new PongResponse(req.getUnpackStamp()));
}
}
diff --git a/src/java/com/threerings/presents/server/Rejector.java b/src/java/com/threerings/presents/server/Rejector.java
index 4681c339f..8a14ed0fa 100644
--- a/src/java/com/threerings/presents/server/Rejector.java
+++ b/src/java/com/threerings/presents/server/Rejector.java
@@ -55,16 +55,12 @@ public class Rejector extends PresentsServer
protected class RejectingAuthenticator extends Authenticator
{
/** Reject all authentication requests. */
- public void authenticateConnection (final AuthingConnection conn)
+ public void authenticateConnection (AuthingConnection conn)
{
Log.info("Rejecting request: " + conn.getAuthRequest());
- omgr.postRunnable(new Runnable() {
- public void run () {
- AuthResponseData rdata = new AuthResponseData();
- rdata.code = _errmsg;
- connectionWasAuthenticated(conn, new AuthResponse(rdata));
- }
- });
+ AuthResponseData rdata = new AuthResponseData();
+ rdata.code = _errmsg;
+ connectionWasAuthenticated(conn, new AuthResponse(rdata));
}
}
diff --git a/src/java/com/threerings/presents/server/net/ConnectionManager.java b/src/java/com/threerings/presents/server/net/ConnectionManager.java
index 5e0e52820..0e3abf9c0 100644
--- a/src/java/com/threerings/presents/server/net/ConnectionManager.java
+++ b/src/java/com/threerings/presents/server/net/ConnectionManager.java
@@ -333,7 +333,8 @@ public class ConnectionManager extends LoopingThread
if (oq.writeOverflowMessages(iterStamp)) {
// if they were all written, we can remove it
oqiter.remove();
- Log.info("Flushed overflow queue " + oq + ".");
+// Log.info("Flushed overflow queue for " +
+// oq.conn + ".");
}
} catch (IOException ioe) {
@@ -363,7 +364,7 @@ public class ConnectionManager extends LoopingThread
}
// otherwise write the message out to the client directly
- writeMessage(conn, (byte[])tup.right, _oflowHandler);
+ writeMessage(conn, (DownstreamMessage)tup.right, _oflowHandler);
}
// check for connections that have completed authentication
@@ -471,7 +472,7 @@ public class ConnectionManager extends LoopingThread
* have been invoked).
*/
protected boolean writeMessage (
- Connection conn, byte[] data, PartialWriteHandler pwh)
+ Connection conn, DownstreamMessage outmsg, PartialWriteHandler pwh)
{
// if the connection to which this message is destined is closed,
// drop the message and move along quietly; this is perfectly
@@ -481,55 +482,44 @@ public class ConnectionManager extends LoopingThread
if (conn.isClosed()) {
return true;
}
-
- // sanity check the message size
- if (data.length > 1024 * 1024) {
- Log.warning("Refusing to write absurdly large message " +
- "[conn=" + conn + ", size=" + data.length + "].");
- return true;
- }
-
- // expand our output buffer if needed to accomodate this message
- if (data.length > _outbuf.capacity()) {
- // increase the buffer size in large increments
- int ncapacity = Math.max(_outbuf.capacity() << 1, data.length);
- Log.info("Expanding output buffer size [nsize=" + ncapacity + "].");
- _outbuf = ByteBuffer.allocateDirect(ncapacity);
- }
-
+
boolean fully = true;
try {
-// Log.info("Writing " + data.length + " byte message to " +
-// conn + ".");
+ // write the message via the connection's object output stream
+ // (which we configure to write data to our framing output
+ // stream)
+ ObjectOutputStream oout = conn.getObjectOutputStream(_framer);
+// Log.info("Sending " + outmsg + ".");
+ oout.writeObject(outmsg);
+ oout.flush();
- // first copy the data into our "direct" output buffer
- _outbuf.put(data);
- _outbuf.flip();
+ try {
+ // then write the framed message to the socket
+ ByteBuffer buffer = _framer.frameAndReturnBuffer();
+ int wrote = conn.getChannel().write(buffer);
+ noteWrite(1, wrote);
- // then write the data to the socket
- int wrote = conn.getChannel().write(_outbuf);
- noteWrite(1, wrote);
-
- if (_outbuf.remaining() > 0) {
- fully = false;
+ if (buffer.remaining() > 0) {
+ fully = false;
// Log.info("Partial write [conn=" + conn +
// ", msg=" + StringUtil.shortClassName(outmsg) +
// ", wrote=" + wrote +
// ", size=" + buffer.limit() + "].");
- pwh.handlePartialWrite(conn, _outbuf);
+ pwh.handlePartialWrite(conn, buffer);
// } else if (wrote > 10000) {
// Log.info("Big write [conn=" + conn +
// ", msg=" + StringUtil.shortClassName(outmsg) +
// ", wrote=" + wrote + "].");
+ }
+
+ } finally {
+ _framer.resetFrame();
}
} catch (IOException ioe) {
// instruct the connection to deal with its failure
conn.handleFailure(ioe);
-
- } finally {
- _outbuf.clear();
}
return fully;
@@ -619,11 +609,7 @@ public class ConnectionManager extends LoopingThread
/**
* 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} which happens when
- * forwarding an event to a client and at the completion of
- * authentication, both of which should happen only on the
- * distributed object thread.
+ * to be delivered.
*/
void postMessage (Connection conn, DownstreamMessage msg)
{
@@ -633,29 +619,8 @@ public class ConnectionManager extends LoopingThread
Thread.dumpStack();
} else {
- // flatten this message using the connection's output stream
- try {
- ObjectOutputStream oout = conn.getObjectOutputStream(_framer);
- oout.writeObject(msg);
- oout.flush();
-
- // now extract that data into a byte array
- ByteBuffer buffer = _framer.frameAndReturnBuffer();
- byte[] data = new byte[buffer.limit()];
- buffer.get(data);
- _framer.resetFrame();
-
-// Log.info("Flattened " + msg + " into " +
-// data.length + " bytes.");
-
- // and slap both on the queue
- _outq.append(new Tuple(conn, data));
-
- } catch (Exception e) {
- Log.warning("Failure flattening message [conn=" + conn +
- ", msg=" + msg + "]. Dropping.");
- Log.logStackTrace(e);
- }
+ // slap both these suckers onto the outgoing message queue
+ _outq.append(new Tuple(conn, msg));
}
}
@@ -745,7 +710,6 @@ public class ConnectionManager extends LoopingThread
if (_partial.remaining() == 0) {
_partial = null;
- _partials++;
} else {
// Log.info("Still going [conn=" + conn +
// ", wrote=" + wrote +
@@ -755,11 +719,10 @@ public class ConnectionManager extends LoopingThread
}
while (size() > 0) {
- byte[] data = (byte[])remove(0);
+ DownstreamMessage outmsg = (DownstreamMessage)remove(0);
// if any of these messages are partially written, we have
// to stop and wait for the next tick
- _msgs++;
- if (!writeMessage(conn, data, this)) {
+ if (!writeMessage(conn, outmsg, this)) {
return false;
}
}
@@ -776,21 +739,9 @@ public class ConnectionManager extends LoopingThread
_partial.flip();
}
- /**
- * Returns a string representation of this instance.
- */
- public String toString ()
- {
- return "[conn=" + conn + ", partials=" + _partials +
- ", msgs=" + _msgs + "]";
- }
-
/** The remains of a message that was only partially written on
* its first attempt. */
protected ByteBuffer _partial;
-
- /** A couple of counters. */
- protected int _msgs, _partials;
}
protected int _port;
@@ -807,7 +758,6 @@ public class ConnectionManager extends LoopingThread
protected Queue _outq = new Queue();
protected FramingOutputStream _framer;
- protected ByteBuffer _outbuf = ByteBuffer.allocateDirect(64 * 1024);
protected HashMap _oflowqs = new HashMap();
@@ -830,7 +780,6 @@ public class ConnectionManager extends LoopingThread
public void handlePartialWrite (Connection conn, ByteBuffer msgbuf) {
// if we couldn't write all the data for this message, we'll
// need to establish an overflow queue
- Log.info("Starting overflow queue for " + conn + ".");
_oflowqs.put(conn, new OverflowQueue(conn, msgbuf));
}
};