diff --git a/src/main/java/com/threerings/nio/conman/ConnectionManager.java b/src/main/java/com/threerings/nio/conman/ConnectionManager.java index f22207a72..052058daa 100644 --- a/src/main/java/com/threerings/nio/conman/ConnectionManager.java +++ b/src/main/java/com/threerings/nio/conman/ConnectionManager.java @@ -32,7 +32,6 @@ import java.nio.channels.SelectionKey; import java.nio.channels.Selector; import java.nio.channels.SocketChannel; -import com.google.common.collect.Lists; import com.google.common.collect.Maps; import com.google.inject.Inject; import com.google.inject.name.Named; @@ -172,7 +171,7 @@ public abstract class ConnectionManager extends LoopingThread // close connections that have had no network traffic for too long long idleStamp = iterStamp - _idleTime; - for (NetEventHandler handler : Lists.newArrayList(_handlers.values())) { + for (NetEventHandler handler : _handlers.values()) { if (handler.checkIdle(idleStamp)) { // this will queue the connection for closure on our next tick handler.becameIdle(); diff --git a/src/main/java/com/threerings/presents/server/net/PresentsConnectionManager.java b/src/main/java/com/threerings/presents/server/net/PresentsConnectionManager.java index b5c2213fd..8a8abcf35 100644 --- a/src/main/java/com/threerings/presents/server/net/PresentsConnectionManager.java +++ b/src/main/java/com/threerings/presents/server/net/PresentsConnectionManager.java @@ -367,32 +367,7 @@ public class PresentsConnectionManager extends ConnectionManager } else { // otherwise we wire up a special event handler that will wait for our socket to // finish the connection process and then wire things up fully - handler = new NetEventHandler() { - public int handleEvent (long when) { - try { - if (sockchan.finishConnect()) { - // great, we're ready to roll, wire up the connection - conn.selkey = sockchan.register(_selector, SelectionKey.OP_READ); - _handlers.put(conn.selkey, conn); - log.info("Outgoing connection ready", "conn", conn); - } - } catch (IOException ioe) { - handleError(ioe); - } - return 0; - } - public boolean checkIdle (long idleStamp) { - return conn.checkIdle(idleStamp); - } - public void becameIdle () { - handleError(new IOException("Pending connection became idle.")); - } - protected void handleError (IOException ioe) { - _handlers.remove(conn.selkey); - _oflowqs.remove(conn); - conn.connectFailure(ioe); - } - }; + handler = new OutgoingConnectionHandler(conn); } _handlers.put(conn.selkey, handler); @@ -402,6 +377,18 @@ public class PresentsConnectionManager extends ConnectionManager } } + @Override // from LoopingThread + protected void iterate () + { + super.iterate(); + + // reap any outgoing connection handlers that failed to connect due to idleness + OutgoingConnectionHandler handler; + while ((handler = _outfailq.getNonBlocking()) != null) { + handler.handleError(new IOException("Pending connection became idle.")); + } + } + @Override // from LoopingThread public boolean isRunning () { @@ -552,6 +539,50 @@ public class PresentsConnectionManager extends ConnectionManager } } + protected class OutgoingConnectionHandler implements NetEventHandler + { + public OutgoingConnectionHandler (Connection conn) + { + _conn = conn; + } + + public int handleEvent (long when) + { + SocketChannel sockchan = _conn.getChannel(); + try { + if (sockchan.finishConnect()) { + // great, we're ready to roll, wire up the connection + _conn.selkey = sockchan.register(_selector, SelectionKey.OP_READ); + _handlers.put(_conn.selkey, _conn); + log.info("Outgoing connection ready", "conn", _conn); + } + } catch (IOException ioe) { + handleError(ioe); + } + return 0; + } + + public boolean checkIdle (long idleStamp) + { + return _conn.checkIdle(idleStamp); + } + + public void becameIdle () + { + // this failed connection will be cleaned up in the next iterate() tick + _outfailq.append(this); + } + + protected void handleError (IOException ioe) + { + _handlers.remove(_conn.selkey); + _oflowqs.remove(_conn); + _conn.connectFailure(ioe); + } + + protected final Connection _conn; + } + /** Handles client authentication. The base authenticator is injected but optional services * like the PeerManager may replace this authenticator with one that intercepts certain types * of authentication and then passes normal authentications through. */ @@ -562,6 +593,9 @@ public class PresentsConnectionManager extends ConnectionManager protected Queue _authq = Queue.newQueue(); protected Queue> _connectq = Queue.newQueue(); + /** failed (idled out) outgoing connections that need to be cleaned up */ + protected Queue _outfailq = Queue.newQueue(); + protected FramingOutputStream _framer = new FramingOutputStream(); protected ByteArrayOutputStream _flattener = new ByteArrayOutputStream();