diff --git a/src/main/java/com/threerings/bureau/server/BureauRegistry.java b/src/main/java/com/threerings/bureau/server/BureauRegistry.java index 8ea76a1a7..b58024e27 100644 --- a/src/main/java/com/threerings/bureau/server/BureauRegistry.java +++ b/src/main/java/com/threerings/bureau/server/BureauRegistry.java @@ -44,7 +44,7 @@ import com.threerings.presents.server.InvocationManager; import com.threerings.presents.server.PresentsSession; import com.threerings.presents.server.ServiceAuthenticator; import com.threerings.presents.server.SessionFactory; -import com.threerings.presents.server.net.ConnectionManager; +import com.threerings.presents.server.net.PresentsConnectionManager; import com.threerings.bureau.data.AgentObject; import com.threerings.bureau.data.BureauAuthName; @@ -100,7 +100,7 @@ public class BureauRegistry * Creates an uninitialized registry. */ @Inject public BureauRegistry ( - InvocationManager invmgr, ConnectionManager conmgr, ClientManager clmgr) + InvocationManager invmgr, PresentsConnectionManager conmgr, ClientManager clmgr) { invmgr.registerDispatcher(new BureauDispatcher(new BureauProvider() { public void bureauInitialized (ClientObject client, String bureauId) { diff --git a/src/main/java/com/threerings/nio/PolicyServer.java b/src/main/java/com/threerings/nio/PolicyServer.java new file mode 100644 index 000000000..81663a2f1 --- /dev/null +++ b/src/main/java/com/threerings/nio/PolicyServer.java @@ -0,0 +1,125 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2010 Three Rings Design, Inc., All Rights Reserved +// http://code.google.com/p/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.nio; + +import java.io.IOException; +import java.nio.channels.SocketChannel; + +import com.google.common.base.Charsets; +import com.google.inject.Inject; +import com.google.inject.Singleton; + +import com.samskivert.util.Lifecycle; +import com.samskivert.util.Tuple; + +import com.threerings.presents.server.ReportManager; + +import com.threerings.nio.conman.Connection; +import com.threerings.nio.conman.ConnectionManager; +import com.threerings.nio.conman.ServerSocketChannelAcceptor; + +import static com.threerings.NaryaLog.log; + +/** + * Binds to a port and responds to "xmlsocket" requests on it with a policy file allowing access + * to a given set of ports on a given hostname. + */ +@Singleton +public class PolicyServer extends ConnectionManager +{ + @Inject + public PolicyServer (Lifecycle cycle, ReportManager mgr) + throws IOException + { + super(cycle, mgr); + } + + public void init (int socketPolicyPort, String publicServerHost, int[] serverPorts, + int gameServerPort) + throws IOException + { + _acceptor = + new ServerSocketChannelAcceptor(publicServerHost, new int[] { socketPolicyPort }, this); + + // build the XML once and for all + StringBuilder policy = new StringBuilder("\n"). + append("\n"); + // if we're running on 843, serve a master policy file + if (socketPolicyPort == MASTER_PORT) { + policy.append(" \n"); + } + + for (String host : new String[] { publicServerHost }) { + policy.append(" \n"); + */ + policy.append("*\"/>\n"); + } + policy.append("\n"); + + _policy = policy.toString().getBytes(Charsets.UTF_8); + start(); + } + + @Override + protected void willStart () + { + super.willStart(); + if (!_acceptor.bind()) { + log.warning("Policy server failed to bind!"); + } + } + + + @Override + protected void didShutdown () + { + super.didShutdown(); + _acceptor.shutdown(); + } + + @Override + protected void handleAcceptedSocket (SocketChannel channel) + { + Connection conn = new Connection() { + @Override public int handleEvent (long when) { + // Ignore incoming data. Should just be "" + return 0; + } + }; + handleAcceptedSocket(channel, conn); + _outq.append(Tuple.newTuple(conn, _policy)); + postAsyncClose(conn); + } + + protected ServerSocketChannelAcceptor _acceptor; + + protected byte[] _policy; + + protected static final int MASTER_PORT = 843; +} diff --git a/src/main/java/com/threerings/nio/conman/Connection.java b/src/main/java/com/threerings/nio/conman/Connection.java new file mode 100644 index 000000000..a417a1933 --- /dev/null +++ b/src/main/java/com/threerings/nio/conman/Connection.java @@ -0,0 +1,206 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2010 Three Rings Design, Inc., All Rights Reserved +// http://code.google.com/p/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.nio.conman; + +import java.io.IOException; +import java.net.InetAddress; +import java.nio.channels.SelectionKey; +import java.nio.channels.SocketChannel; +import com.threerings.presents.net.PingRequest; +import static com.threerings.presents.Log.log; + +/** + * Implements the net event handler interface to check for delinquency and manages a client + * connection. Subclasses must handle incoming data in {@link #handleEvent}. + */ +public abstract class Connection implements NetEventHandler +{ + /** The key used by the NIO code to track this connection. */ + public SelectionKey selkey; + + /** + * Initializes a connection object with a socket and related info. + * + * @param cmgr The connection manager with which this connection is associated. + * @param channel The socket channel from which we'll be reading messages. + * @param createStamp The time at which this connection was created. + */ + public void init (ConnectionManager cmgr, SocketChannel channel, long createStamp) + throws IOException + { + _cmgr = cmgr; + _channel = channel; + _lastEvent = createStamp; + _connectionId = ++_lastConnectionId; + } + + /** + * Returns the connection's unique identifier. + */ + public int getConnectionId () + { + return _connectionId; + } + + /** + * Returns the non-blocking socket object used to construct this connection. + */ + public SocketChannel getChannel () + { + return _channel; + } + + /** + * Returns the address associated with this connection or null if it has no underlying socket + * channel. + */ + public InetAddress getInetAddress () + { + return (_channel == null) ? null : _channel.socket().getInetAddress(); + } + + /** + * Returns true if this connection is closed. + */ + public boolean isClosed () + { + return (_channel == null); + } + + /** + * Closes this connection and unregisters it from the connection manager. This should only be + * called from the conmgr thread. + */ + public void close () + { + // we shouldn't be closed twice + if (isClosed()) { + log.warning("Attempted to re-close connection " + this + ".", new Exception()); + return; + } + + // unregister from the select set + _cmgr.connectionClosed(this); + + // close our socket + closeSocket(); + } + + /** + * Queues up a request to have this connection closed by the connection manager once all + * messages in its queue have been written to its target. + */ + public void asyncClose () + { + _cmgr.postAsyncClose(this); + } + + /** + * Called when an outgoing socket experiences a connect failure. The connection manager will + * have cleaned up the partial registration needed during the connect process, so we are only + * responsible for closing our socket. + */ + public void connectFailure (IOException ioe) + { + closeSocket(); + } + + /** + * Called when there is a failure reading or writing to this connection. We notify the + * connection manager and close ourselves down. + */ + public void networkFailure (IOException ioe) + { + // if we're already closed, then something is seriously funny + if (isClosed()) { + log.warning("Failure reported on closed connection " + this + ".", new Exception()); + return; + } + + // let the connection manager know we're hosed + _cmgr.connectionFailed(this, ioe); + + // and close our socket + closeSocket(); + } + + // from interface NetEventHandler + public boolean checkIdle (long now) + { + long idleMillis = now - _lastEvent; + if (idleMillis < PingRequest.PING_INTERVAL + LATENCY_GRACE) { + return false; + } + if (isClosed()) { + return true; + } + log.info("Disconnecting non-communicative client", "conn", this, "idle", idleMillis + "ms"); + return true; + } + + // from interface NetEventHandler + public void becameIdle () + { + _cmgr.closeConnection(this); + } + + @Override // from Object + public String toString () + { + return "[id=" + (hashCode() % 1000) + ", addr=" + getInetAddress() + "]"; + } + + /** + * Closes the socket associated with this connection. This happens when we receive EOF, are + * requested to close down or when our connection fails. + */ + protected void closeSocket () + { + if (_channel == null) { + return; + } + + log.debug("Closing channel " + this + "."); + try { + _channel.close(); + } catch (IOException ioe) { + log.warning("Error closing connection", "conn", this, "error", ioe); + } + + // clear out our references to prevent repeat closings + _channel = null; + } + + protected ConnectionManager _cmgr; + protected SocketChannel _channel; + + protected long _lastEvent; + + protected int _connectionId; + + /** 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/main/java/com/threerings/presents/server/net/ConnectionManager.java b/src/main/java/com/threerings/nio/conman/ConnectionManager.java similarity index 61% rename from src/main/java/com/threerings/presents/server/net/ConnectionManager.java rename to src/main/java/com/threerings/nio/conman/ConnectionManager.java index b06ff37be..33064ba9a 100644 --- a/src/main/java/com/threerings/presents/server/net/ConnectionManager.java +++ b/src/main/java/com/threerings/nio/conman/ConnectionManager.java @@ -19,77 +19,52 @@ // 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.server.net; +package com.threerings.nio.conman; import java.util.ArrayList; -import java.util.List; import java.util.Map; -import java.io.ByteArrayOutputStream; import java.io.IOException; -import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.NotYetConnectedException; +import java.nio.channels.SelectableChannel; 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.Singleton; import com.google.inject.name.Named; import com.samskivert.util.IntMap; import com.samskivert.util.IntMaps; -import com.samskivert.util.Invoker; import com.samskivert.util.Lifecycle; import com.samskivert.util.LoopingThread; import com.samskivert.util.Queue; -import com.samskivert.util.ResultListener; import com.samskivert.util.Tuple; -import com.threerings.io.ByteBufferInputStream; -import com.threerings.io.FramingOutputStream; -import com.threerings.io.ObjectOutputStream; -import com.threerings.io.UnreliableObjectInputStream; -import com.threerings.io.UnreliableObjectOutputStream; +import com.threerings.presents.data.ConMgrStats; +import com.threerings.presents.server.ReportManager; import com.threerings.nio.SelectorIterable; -import com.threerings.presents.annotation.AuthInvoker; -import com.threerings.presents.client.Client; -import com.threerings.presents.data.ConMgrStats; -import com.threerings.presents.net.Message; -import com.threerings.presents.net.PongResponse; -import com.threerings.presents.net.Transport; -import com.threerings.presents.server.Authenticator; -import com.threerings.presents.server.ChainedAuthenticator; -import com.threerings.presents.server.ClientManager; -import com.threerings.presents.server.DummyAuthenticator; -import com.threerings.presents.server.PresentsDObjectMgr; -import com.threerings.presents.server.PresentsServer; -import com.threerings.presents.server.ReportManager; -import com.threerings.presents.util.DatagramSequencer; - -import static com.threerings.presents.Log.log; +import static com.threerings.NaryaLog.log; /** * Manages socket connections. It creates connection objects for each socket connection, but those * connection objects interact closely with the connection manager because network I/O is done via * a poll()-like mechanism rather than via threads.

* - * ConnectionManager doesn't directly accept TCP connections; it expects a custom derived class or - * external entity to do so and call its {@link #handleAcceptedSocket} method. - * See {@link BindingConnectionManager} for the standalone implementation. + * ConnectionManager doesn't directly accept TCP connections; it expects + * {@link ServerSocketChannelAcceptor} or an external entity to do so and call its + * {@link #handleAcceptedSocket} method */ -@Singleton -public class ConnectionManager extends LoopingThread +public abstract class ConnectionManager extends LoopingThread implements Lifecycle.ShutdownComponent, ReportManager.Reporter { /** * Creates a connection manager instance. */ - @Inject public ConnectionManager (Lifecycle cycle, ReportManager repmgr) + public ConnectionManager (Lifecycle cycle, ReportManager repmgr) throws IOException { super("ConnectionManager"); @@ -98,15 +73,6 @@ public class ConnectionManager extends LoopingThread _selector = Selector.open(); } - /** - * Adds an authenticator to the authentication chain. This authenticator will be offered a - * chance to authenticate incoming connections before falling back to the main authenticator. - */ - public void addChainedAuthenticator (ChainedAuthenticator author) - { - _authors.add(author); - } - /** * Instructs us to execute the specified runnable when the connection manager thread exits. * Note: this will be executed on the connection manager thread, so don't do anything @@ -127,7 +93,6 @@ public class ConnectionManager extends LoopingThread // fill in our snapshot values _stats.connectionCount = _connections.size(); _stats.handlerCount = _handlers.size(); - _stats.authQueueSize = _authq.size(); _stats.deathQueueSize = _deathq.size(); _stats.outQueueSize = _outq.size(); if (_oflowqs.size() > 0) { @@ -139,6 +104,18 @@ public class ConnectionManager extends LoopingThread return _stats.clone(); } + /** + * Registers ops on chan on this manager's selector and hooks + * netEventHandler up to receive events whenever the selection occurs. + */ + public SelectionKey register (SelectableChannel chan, int ops, NetEventHandler netEventHandler) + throws IOException + { + SelectionKey key = chan.register(_selector, ops); + _handlers.put(key, netEventHandler); + return key; + } + /** * Introduces a new active socket into Presents from off the ConnectionManager thread. If * Presents is embedded in another framework that handles socket acceptance, this will be @@ -149,32 +126,6 @@ public class ConnectionManager extends LoopingThread _acceptedq.append(channel); } - /** - * Opens an outgoing connection to the supplied address. The connection will be opened in a - * non-blocking manner and added to the connection manager's select set. Messages posted to the - * connection prior to it being actually connected to its destination will remain in the queue. - * If the connection fails those messages will be dropped. - * - * @param conn the connection to be initialized and opened. Callers may want to provide a - * {@link Connection} derived class so that they may intercept calldown methods. - * @param hostname the hostname of the server to which to connect. - * @param port the port on which to connect to the server. - * - * @exception IOException thrown if an error occurs creating our socket. Everything else - * happens asynchronously. If the connection attempt fails, the Connection will be notified via - * {@link Connection#networkFailure}. - */ - public void openOutgoingConnection (Connection conn, String hostname, int port) - throws IOException - { - // create a socket channel to use for this connection, initialize it and queue it up to - // have the non-blocking connect process started - SocketChannel sockchan = SocketChannel.open(); - sockchan.configureBlocking(false); - conn.init(this, sockchan, System.currentTimeMillis()); - _connectq.append(Tuple.newTuple(conn, new InetSocketAddress(hostname, port))); - } - /** * Queues a connection up to be closed on the conmgr thread. */ @@ -228,13 +179,6 @@ public class ConnectionManager extends LoopingThread report.append(bytesOut*1000/sinceLast).append(" bps\n"); } - @Override // from LoopingThread - public boolean isRunning () - { - // Prevent exiting our thread until the object manager is done. - return super.isRunning() || _omgr.isRunning(); - } - @Override // from LoopingThread protected void willStart () { @@ -262,6 +206,7 @@ public class ConnectionManager extends LoopingThread _lastDebugStamp = iterStamp; } + // close any connections that have been queued up to die Connection dconn; while ((dconn = _deathq.getNonBlocking()) != null) { @@ -289,21 +234,7 @@ public class ConnectionManager extends LoopingThread // those last moments we don't want to accept new connections or read any incoming messages if (super.isRunning()) { - SocketChannel accepted; - while ((accepted = _acceptedq.getNonBlocking()) != null) { - handleAcceptedSocket(accepted); - } - // start up any outgoing connections that need to be connected - Tuple pconn; - while ((pconn = _connectq.getNonBlocking()) != null) { - startOutgoingConnection(pconn.left, pconn.right); - } - - // check for connections that have completed authentication - processAuthedConnections(iterStamp); - - // listen for and process incoming network events - processIncomingEvents(iterStamp); + handleIncoming(iterStamp); } if (DEBUG_REPORT && generateDebugReport) { @@ -311,109 +242,31 @@ public class ConnectionManager extends LoopingThread } } - /** - * Performs the authentication process on the specified connection. This is called by {@link - * AuthingConnection} itself once it receives its auth request. - */ - protected void authenticateConnection (AuthingConnection conn) - { - Authenticator author = _author; - for (ChainedAuthenticator cauthor : _authors) { - if (cauthor.shouldHandleConnection(conn)) { - author = cauthor; - break; - } + protected void handleIncoming(long iterStamp) { + + SocketChannel accepted; + while ((accepted = _acceptedq.getNonBlocking()) != null) { + handleAcceptedSocket(accepted); } - author.authenticateConnection(_authInvoker, conn, new ResultListener() { - public void requestCompleted (AuthingConnection conn) { - _authq.append(conn); - } - public void requestFailed (Exception cause) { - // this never happens - } - }); + // listen for and process incoming network events + processIncomingEvents(iterStamp); } /** - * Creates a datagram sequencer for use by a {@link Connection}. + * Adds a connection for the given socket to the managed set. */ - protected DatagramSequencer createDatagramSequencer () - { - return new DatagramSequencer( - new UnreliableObjectInputStream(new ByteBufferInputStream(_databuf)), - new UnreliableObjectOutputStream(_flattener)); - } + protected abstract void handleAcceptedSocket (SocketChannel channel); - /** - * Starts the connection process for an outgoing connection. This is called as part of the - * conmgr tick for any pending outgoing connections. - */ - protected void startOutgoingConnection (final Connection conn, InetSocketAddress addr) - { - final SocketChannel sockchan = conn.getChannel(); - try { - // register our channel with the selector (if this fails, we abandon ship immediately) - conn.selkey = sockchan.register(_selector, SelectionKey.OP_CONNECT); - - // start our connection process (now if we fail we need to clean things up) - NetEventHandler handler; - if (sockchan.connect(addr)) { - // it is possible even for a non-blocking socket to connect immediately, in which - // case we stick the connection in as its event handler immediately - handler = conn; - - } 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 now) { - return conn.checkIdle(now); - } - 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); - } - }; - } - _handlers.put(conn.selkey, handler); - - } catch (IOException ioe) { - log.warning("Failed to initiate connection for " + sockchan + ".", ioe); - conn.connectFailure(ioe); // nothing else to clean up - } - } - - /** - * Starts an accepted socket down the path to authorization. - */ - protected void handleAcceptedSocket (SocketChannel channel) + protected void handleAcceptedSocket (SocketChannel channel, Connection conn) { try { // create a new authing connection object to manage the authentication of this client // connection and register it with our selection set channel.configureBlocking(false); - AuthingConnection aconn = new AuthingConnection(); - aconn.selkey = channel.register(_selector, SelectionKey.OP_READ); - aconn.init(this, channel, System.currentTimeMillis()); - _handlers.put(aconn.selkey, aconn); + conn.init(this, channel, System.currentTimeMillis()); + conn.selkey = register(channel, SelectionKey.OP_READ, conn); + _handlers.put(conn.selkey, conn); synchronized (this) { _stats.connects++; } @@ -430,48 +283,6 @@ public class ConnectionManager extends LoopingThread } } - /** - * Converts connections that have completed the authentication process into full running - * connections and notifies the client manager that new connections have been established. - */ - protected void processAuthedConnections (long iterStamp) - { - AuthingConnection conn; - while ((conn = _authq.getNonBlocking()) != null) { - try { - // construct a new running connection to handle this connections network traffic - // from here on out - Connection rconn = new Connection(); - rconn.init(this, conn.getChannel(), iterStamp); - rconn.selkey = conn.selkey; - - // we need to keep using the same object input and output streams from the - // beginning of the session because they have context that needs to be preserved - rconn.inheritStreams(conn); - - // replace the mapping in the handlers table from the old conn with the new one - _handlers.put(rconn.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) { - _oflowqs.put(rconn, oflowHandler); - } - - // and let the client manager know about our new connection - _clmgr.connectionEstablished(rconn, conn.getAuthName(), conn.getAuthRequest(), - conn.getAuthResponse()); - - } catch (IOException ioe) { - log.warning("Failure upgrading authing connection to running.", ioe); - } - } - } - /** * Checks for any network events on our set of sockets and passes those events down to their * associated {@link NetEventHandler}s for processing. @@ -568,11 +379,6 @@ 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); - } } /** @@ -652,29 +458,6 @@ 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 conn.getDatagramChannel().send(_databuf, target) > 0; - } catch (IOException ioe) { - log.warning("Failed to send datagram.", ioe); - return false; - } - } - /** Called by {@link #writeMessage} and friends when they write data over the network. */ protected synchronized void noteWrite (int msgs, int bytes) { @@ -682,96 +465,6 @@ public class ConnectionManager extends LoopingThread _stats.bytesOut += bytes; } - /** - * 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 must happen only on the distributed object thread. - */ - protected void postMessage (Connection conn, Message msg) - { - if (!isRunning()) { - log.warning("Posting message to inactive connection manager", - "msg", msg, new Exception()); - } - - // sanity check - if (conn == null || msg == null) { - log.warning("postMessage() bogosity", "conn", conn, "msg", msg, new Exception()); - return; - } - - // more sanity check; messages must only be posted from the dobjmgr thread - if (!_omgr.isDispatchThread()) { - log.warning("Message posted on non-distributed object thread", "conn", conn, - "msg", msg, "thread", Thread.currentThread(), new Exception()); - // let it through though as we don't want to break things unnecessarily - } - - try { - // send it as a datagram if hinted and possible (pongs must be sent as part of the - // negotation process) - if (!msg.getTransport().isReliable() && - (conn.getTransmitDatagrams() || msg instanceof PongResponse) && - postDatagram(conn, msg)) { - return; - } - - // note the actual transport - msg.noteActualTransport(Transport.RELIABLE_ORDERED); - - _framer.resetFrame(); - - // flatten this message using the connection's output stream - 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); - // log.info("Flattened " + msg + " into " + data.length + " bytes."); - - // and slap both on the queue - _outq.append(Tuple.newTuple(conn, data)); - - } catch (Exception e) { - log.warning("Failure flattening message", "conn", conn, "msg", msg, e); - } - } - - /** - * Helper function for {@link #postMessage}; handles posting the message as a datagram. - * - * @return true if the datagram was successfully posted, false if it was too big. - */ - protected boolean postDatagram (Connection conn, Message msg) - throws Exception - { - _flattener.reset(); - - // flatten the message using the connection's sequencer - DatagramSequencer sequencer = conn.getDatagramSequencer(); - sequencer.writeDatagram(msg); - - // if the message is too big, we must fall back to sending it through the stream channel - if (_flattener.size() > Client.MAX_DATAGRAM_SIZE) { - return false; - } - - // note the actual transport - msg.noteActualTransport(Transport.UNRELIABLE_UNORDERED); - - // extract as a byte array - byte[] data = _flattener.toByteArray(); - - // slap it on the queue - _dataq.append(Tuple.newTuple(conn, data)); - - return true; - } - /** * Posts a fake message to this connection's outgoing message queue that will cause the * connection to be closed when this message is reached. This is only used by outgoing @@ -798,9 +491,6 @@ public class ConnectionManager extends LoopingThread synchronized (this) { _stats.disconnects++; } - - // let the client manager know what's up - _clmgr.connectionFailed(conn, ioe); } /** @@ -816,9 +506,6 @@ public class ConnectionManager extends LoopingThread synchronized (this) { _stats.closes++; } - - // let the client manager know what's up - _clmgr.connectionClosed(conn); } @Override @@ -963,12 +650,6 @@ public class ConnectionManager extends LoopingThread } }; - /** 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. */ - @Inject(optional=true) protected Authenticator _author = new DummyAuthenticator(); - protected List _authors = Lists.newArrayList(); - protected Selector _selector; protected SelectorIterable _selectorSelector; @@ -979,17 +660,11 @@ public class ConnectionManager extends LoopingThread protected IntMap _connections = IntMaps.newHashIntMap(); protected Queue _deathq = Queue.newQueue(); - protected Queue _authq = Queue.newQueue(); - protected Queue> _connectq = Queue.newQueue(); protected Queue _acceptedq = Queue.newQueue(); protected Queue> _outq = Queue.newQueue(); - protected Queue> _dataq = Queue.newQueue(); - protected FramingOutputStream _framer = new FramingOutputStream(); - protected ByteArrayOutputStream _flattener = new ByteArrayOutputStream(); protected ByteBuffer _outbuf = ByteBuffer.allocateDirect(64 * 1024); - protected ByteBuffer _databuf = ByteBuffer.allocateDirect(Client.MAX_DATAGRAM_SIZE); protected Map _oflowqs = Maps.newHashMap(); @@ -1012,12 +687,6 @@ public class ConnectionManager extends LoopingThread @Inject(optional=true) @Named("presents.net.selectLoopTime") protected int _selectLoopTime = 100; - // some dependencies - @Inject @AuthInvoker protected Invoker _authInvoker; - @Inject protected ClientManager _clmgr; - @Inject protected PresentsDObjectMgr _omgr; - @Inject protected PresentsServer _server; - /** Used to denote asynchronous close requests. */ protected static final byte[] ASYNC_CLOSE_REQUEST = new byte[0]; diff --git a/src/main/java/com/threerings/presents/server/net/NetEventHandler.java b/src/main/java/com/threerings/nio/conman/NetEventHandler.java similarity index 91% rename from src/main/java/com/threerings/presents/server/net/NetEventHandler.java rename to src/main/java/com/threerings/nio/conman/NetEventHandler.java index 6e7801f91..e85c560bf 100644 --- a/src/main/java/com/threerings/presents/server/net/NetEventHandler.java +++ b/src/main/java/com/threerings/nio/conman/NetEventHandler.java @@ -19,7 +19,7 @@ // 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.server.net; +package com.threerings.nio.conman; /** * When a network event occurs, the connection manager calls the net event @@ -29,8 +29,8 @@ package com.threerings.presents.server.net; * client is authenticating and the one that processes messages from * authenticated connections. * - *

Utilising this interface prevents us from having to do a bunch of - * inefficient and ugly comparisons; instead we can simply call through an + *

Using this interface prevents us from having to do a bunch of + * inefficient and ugly comparisons; instead we can call through an * interface method to the proper code. */ public interface NetEventHandler diff --git a/src/main/java/com/threerings/nio/conman/ServerSocketChannelAcceptor.java b/src/main/java/com/threerings/nio/conman/ServerSocketChannelAcceptor.java new file mode 100644 index 000000000..f53f0fcb5 --- /dev/null +++ b/src/main/java/com/threerings/nio/conman/ServerSocketChannelAcceptor.java @@ -0,0 +1,134 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2010 Three Rings Design, Inc., All Rights Reserved +// http://code.google.com/p/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.nio.conman; + +import java.util.List; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.channels.SelectionKey; +import java.nio.channels.ServerSocketChannel; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Lists; + +import com.samskivert.net.AddressUtil; +import com.samskivert.util.Lifecycle; + +import static com.threerings.NaryaLog.log; + +/** + * Binds sockets on a given hostname for a set of ports for tcp connections, and passes accepted + * connections into a connection manager. + */ +public class ServerSocketChannelAcceptor + implements Lifecycle.ShutdownComponent +{ + /** + * Configures the connection manager with the hostname and ports on which it will listen for + * socket connections. This must be called before the connection manager is started (via + * {@via #start}) as the sockets will be bound at that time. + * + * @param socketHostname the hostname to which we bind our sockets or null to bind to all + * interfaces. + * @param socketPorts the ports on which to listen for TCP connection. + */ + public ServerSocketChannelAcceptor (String socketHostname, int[] socketPorts, + ConnectionManager mgr) + { + Preconditions.checkNotNull(socketPorts, "Socket ports must be non-null."); + _bindHostname = socketHostname; + _ports = socketPorts; + _conMan = mgr; + } + + /** + * Bind to the socket ports and return true if any of the binds succeeded. + */ + public boolean bind () + { + // open our TCP listening ports + int successes = 0; + for (int port : _ports) { + try { + acceptConnections(port); + successes++; + } catch (IOException ioe) { + log.warning("Failure listening to socket", "hostname", _bindHostname, + "port", port, ioe); + } + } + return successes > 0; + } + + /** + * Unbind our listening sockets. + */ + public void shutdown () + { + for (ServerSocketChannel ssocket : _ssockets) { + try { + ssocket.socket().close(); + } catch (IOException ioe) { + log.warning("Failed to close listening socket: " + ssocket, ioe); + } + } + } + + + protected void acceptConnections (int port) + throws IOException + { + // create a listening socket + final ServerSocketChannel ssocket = ServerSocketChannel.open(); + ssocket.configureBlocking(false); + InetSocketAddress isa = AddressUtil.getAddress(_bindHostname, port); + ssocket.socket().bind(isa); + + _conMan.register(ssocket, SelectionKey.OP_ACCEPT, new NetEventHandler() { + public int handleEvent (long when) { + try { + _conMan.handleAcceptedSocket(ssocket.accept()); + } catch (IOException ioe) { + log.info("Failure accepting connected socket: " +ioe); + } + // there's no easy way to measure bytes read when accepting a connection, so we + // claim nothing + return 0; + } + public boolean checkIdle (long now) { + return false; // we're never idle + } + public void becameIdle () { + // we're never idle + } + }); + _ssockets.add(ssocket); + log.info("Server listening on " + isa + "."); + } + + protected final int[] _ports; + protected final String _bindHostname; + protected final ConnectionManager _conMan; + + protected List _ssockets = Lists.newArrayList(); +} diff --git a/src/main/java/com/threerings/presents/peer/server/PeerManager.java b/src/main/java/com/threerings/presents/peer/server/PeerManager.java index d594e828b..cd0f35903 100644 --- a/src/main/java/com/threerings/presents/peer/server/PeerManager.java +++ b/src/main/java/com/threerings/presents/peer/server/PeerManager.java @@ -57,6 +57,10 @@ import com.threerings.io.Streamable; import com.threerings.util.Name; +import com.threerings.presents.annotation.MainInvoker; +import com.threerings.presents.client.Client; +import com.threerings.presents.client.InvocationService; +import com.threerings.presents.data.ClientObject; import com.threerings.presents.dobj.DObject; import com.threerings.presents.dobj.EntryAddedEvent; import com.threerings.presents.dobj.EntryRemovedEvent; @@ -64,11 +68,13 @@ import com.threerings.presents.dobj.EntryUpdatedEvent; import com.threerings.presents.dobj.ObjectAccessException; import com.threerings.presents.dobj.SetListener; import com.threerings.presents.dobj.Subscriber; - -import com.threerings.presents.annotation.MainInvoker; -import com.threerings.presents.client.Client; -import com.threerings.presents.client.InvocationService; -import com.threerings.presents.data.ClientObject; +import com.threerings.presents.peer.client.PeerService; +import com.threerings.presents.peer.data.ClientInfo; +import com.threerings.presents.peer.data.NodeObject; +import com.threerings.presents.peer.data.PeerAuthName; +import com.threerings.presents.peer.net.PeerCreds; +import com.threerings.presents.peer.server.persist.NodeRecord; +import com.threerings.presents.peer.server.persist.NodeRepository; import com.threerings.presents.server.ClientManager; import com.threerings.presents.server.InvocationException; import com.threerings.presents.server.InvocationManager; @@ -77,15 +83,7 @@ import com.threerings.presents.server.PresentsSession; import com.threerings.presents.server.ReportManager; import com.threerings.presents.server.ServiceAuthenticator; import com.threerings.presents.server.SessionFactory; -import com.threerings.presents.server.net.ConnectionManager; - -import com.threerings.presents.peer.client.PeerService; -import com.threerings.presents.peer.data.ClientInfo; -import com.threerings.presents.peer.data.NodeObject; -import com.threerings.presents.peer.data.PeerAuthName; -import com.threerings.presents.peer.net.PeerCreds; -import com.threerings.presents.peer.server.persist.NodeRecord; -import com.threerings.presents.peer.server.persist.NodeRepository; +import com.threerings.presents.server.net.PresentsConnectionManager; import static com.threerings.presents.Log.log; @@ -1599,7 +1597,7 @@ public abstract class PeerManager // our service dependencies @Inject protected @MainInvoker Invoker _invoker; @Inject protected ClientManager _clmgr; - @Inject protected ConnectionManager _conmgr; + @Inject protected PresentsConnectionManager _conmgr; @Inject protected Injector _injector; @Inject protected InvocationManager _invmgr; @Inject protected NodeRepository _noderepo; diff --git a/src/main/java/com/threerings/presents/peer/server/PeerNode.java b/src/main/java/com/threerings/presents/peer/server/PeerNode.java index b7f71ea33..0cc6ec062 100644 --- a/src/main/java/com/threerings/presents/peer/server/PeerNode.java +++ b/src/main/java/com/threerings/presents/peer/server/PeerNode.java @@ -48,7 +48,7 @@ import com.threerings.presents.peer.data.NodeObject; import com.threerings.presents.peer.net.PeerBootstrapData; import com.threerings.presents.peer.server.persist.NodeRecord; import com.threerings.presents.server.PresentsDObjectMgr; -import com.threerings.presents.server.net.ConnectionManager; +import com.threerings.presents.server.net.PresentsConnectionManager; import com.threerings.presents.server.net.ServerCommunicator; import static com.threerings.presents.Log.log; @@ -354,7 +354,7 @@ public class PeerNode @Inject protected PeerManager _peermgr; @Inject protected PresentsDObjectMgr _omgr; - @Inject protected ConnectionManager _conmgr; + @Inject protected PresentsConnectionManager _conmgr; /** The amount of time after which a node record can be considered out of date and invalid. */ protected static final long STALE_INTERVAL = 5L * 60L * 1000L; diff --git a/src/main/java/com/threerings/presents/server/ClientManager.java b/src/main/java/com/threerings/presents/server/ClientManager.java index c1de49a24..3adf71c6c 100644 --- a/src/main/java/com/threerings/presents/server/ClientManager.java +++ b/src/main/java/com/threerings/presents/server/ClientManager.java @@ -46,7 +46,9 @@ import com.threerings.presents.data.ClientObject; import com.threerings.presents.net.AuthRequest; import com.threerings.presents.net.AuthResponse; import com.threerings.presents.server.net.AuthingConnection; -import com.threerings.presents.server.net.Connection; +import com.threerings.presents.server.net.PresentsConnection; + +import com.threerings.nio.conman.Connection; import static com.threerings.presents.Log.log; @@ -407,7 +409,7 @@ public class ClientManager * Called by the connection manager to let us know when a new connection has been established. */ public synchronized void connectionEstablished ( - Connection conn, Name authname, AuthRequest req, AuthResponse rsp) + PresentsConnection conn, Name authname, AuthRequest req, AuthResponse rsp) { String type = authname.getClass().getSimpleName(); diff --git a/src/main/java/com/threerings/presents/server/PresentsServer.java b/src/main/java/com/threerings/presents/server/PresentsServer.java index ee348e035..93450a692 100644 --- a/src/main/java/com/threerings/presents/server/PresentsServer.java +++ b/src/main/java/com/threerings/presents/server/PresentsServer.java @@ -41,11 +41,14 @@ import com.threerings.presents.dobj.DObject; import com.threerings.presents.dobj.DObjectManager; import com.threerings.presents.dobj.RootDObjectManager; import com.threerings.presents.server.PresentsDObjectMgr.LongRunnable; -import com.threerings.presents.server.net.BindingConnectionManager; -import com.threerings.presents.server.net.ConnectionManager; +import com.threerings.presents.server.net.DatagramChannelReader; +import com.threerings.presents.server.net.PresentsConnectionManager; import com.threerings.crowd.server.PlaceManager; +import com.threerings.nio.conman.ConnectionManager; +import com.threerings.nio.conman.ServerSocketChannelAcceptor; + import static com.threerings.presents.Log.log; /** @@ -64,21 +67,13 @@ public class PresentsServer { @Override protected void configure () { bindInvokers(); - bindConnectionManager(); + bind(ConnectionManager.class).to(PresentsConnectionManager.class); bind(RunQueue.class).annotatedWith(EventQueue.class).to(PresentsDObjectMgr.class); bind(DObjectManager.class).to(PresentsDObjectMgr.class); bind(RootDObjectManager.class).to(PresentsDObjectMgr.class); bind(Lifecycle.class).toInstance(new Lifecycle()); } - /** - * Binds just the connection manager so this can be overridden if desired. - */ - protected void bindConnectionManager() - { - bind(ConnectionManager.class).to(BindingConnectionManager.class); - } - /** * Binds just the invokers so this can be overridden if desired. */ @@ -157,12 +152,13 @@ public class PresentsServer // provide our client manager with the injector it needs _clmgr.setInjector(injector); - // if we have a connection manager that handles its own socket accepting, initialize it - // with the hostname and ports on which to listen - if (_conmgr instanceof BindingConnectionManager) { - ((BindingConnectionManager)_conmgr).init( - getBindHostname(), getDatagramHostname(), getListenPorts(), getDatagramPorts()); - } + // Create our socket opening objects now that we know what ports we're using. + _socketAcceptor = + new ServerSocketChannelAcceptor(getBindHostname(), getListenPorts(), _conmgr); + _lifecycle.addComponent(_socketAcceptor); + _datagramReader = + new DatagramChannelReader(getDatagramHostname(), getDatagramPorts(), _conmgr); + _lifecycle.addComponent(_datagramReader); // initialize the time base services TimeBaseProvider.init(_invmgr, _omgr); @@ -213,7 +209,12 @@ public class PresentsServer */ protected void openToThePublic () { - _conmgr.start(); + if (getListenPorts().length != 0 && !_socketAcceptor.bind()) { + queueShutdown(); + } else { + _datagramReader.bind(); + _conmgr.start(); + } } /** @@ -257,7 +258,9 @@ public class PresentsServer } /** - * Returns the port on which the connection manager will listen for client connections. + * Returns the port on which the connection manager will listen for client connections or an + * empty array to skip binding and have an external entity transfer accepted sockets into + * the connection manager. */ protected int[] getListenPorts () { @@ -265,7 +268,8 @@ public class PresentsServer } /** - * Returns the ports on which the connection manager will listen for datagrams. + * Returns the ports on which the connection manager will listen for datagrams or an empty + * array if no datagrams are desired. */ protected int[] getDatagramPorts () { @@ -282,11 +286,15 @@ public class PresentsServer { } + protected ServerSocketChannelAcceptor _socketAcceptor; + + protected DatagramChannelReader _datagramReader; + /** The manager of distributed objects. */ @Inject protected PresentsDObjectMgr _omgr; /** The manager of network connections. */ - @Inject protected ConnectionManager _conmgr; + @Inject protected PresentsConnectionManager _conmgr; /** The manager of clients. */ @Inject protected ClientManager _clmgr; diff --git a/src/main/java/com/threerings/presents/server/PresentsSession.java b/src/main/java/com/threerings/presents/server/PresentsSession.java index 342c1ad61..16e596662 100644 --- a/src/main/java/com/threerings/presents/server/PresentsSession.java +++ b/src/main/java/com/threerings/presents/server/PresentsSession.java @@ -67,8 +67,10 @@ import com.threerings.presents.net.TransmitDatagramsRequest; import com.threerings.presents.net.UnsubscribeRequest; import com.threerings.presents.net.UnsubscribeResponse; import com.threerings.presents.net.UpdateThrottleMessage; -import com.threerings.presents.server.net.Connection; -import com.threerings.presents.server.net.ConnectionManager; +import com.threerings.presents.server.net.PresentsConnection; + +import com.threerings.nio.conman.Connection; +import com.threerings.nio.conman.ConnectionManager; import static com.threerings.presents.Log.log; @@ -84,7 +86,7 @@ import static com.threerings.presents.Log.log; * from the conmgr thread and therefore also need not be synchronized. */ public class PresentsSession - implements Connection.MessageHandler, ClientResolutionListener + implements PresentsConnection.MessageHandler, ClientResolutionListener { /** Used by {@link PresentsSession#setUsername} to report success or failure. */ public static interface UserChangeListener @@ -169,7 +171,7 @@ public class PresentsSession */ public boolean getTransmitDatagrams () { - Connection conn = getConnection(); + PresentsConnection conn = getConnection(); return conn != null && conn.getTransmitDatagrams(); } @@ -180,7 +182,7 @@ public class PresentsSession public void setClassLoader (ClassLoader loader) { _loader = loader; - Connection conn = getConnection(); + PresentsConnection conn = getConnection(); if (conn != null) { conn.setClassLoader(loader); } @@ -459,10 +461,11 @@ public class PresentsSession } /** - * Initializes this client instance with the specified username, connection instance and client - * object and begins a client session. + * Initializes this client instance with the specified username, connection instance and + * client object and begins a client session. */ - protected void startSession (Name authname, AuthRequest req, Connection conn, Object authdata) + protected void startSession (Name authname, AuthRequest req, PresentsConnection conn, + Object authdata) { _authname = authname; _areq = req; @@ -480,7 +483,7 @@ public class PresentsSession * Called by the client manager when a new connection arrives that authenticates as this * already established client. This must only be called from the congmr thread. */ - protected void resumeSession (AuthRequest req, Connection conn) + protected void resumeSession (AuthRequest req, PresentsConnection conn) { // check to see if we've already got a connection object, in which case it's probably stale Connection oldconn = getConnection(); @@ -804,7 +807,7 @@ public class PresentsSession * Sets our connection reference in a thread safe way. Also establishes the back reference to * us as the connection's message handler. */ - protected synchronized void setConnection (Connection conn) + protected synchronized void setConnection (PresentsConnection conn) { // if our connection is being cleared out, record the amount of time we were connected to // our total connected time @@ -839,7 +842,7 @@ public class PresentsSession * @return The connection instance associated with this client or null if the client is not * currently connected. */ - protected synchronized Connection getConnection () + protected synchronized PresentsConnection getConnection () { return _conn; } @@ -860,7 +863,7 @@ public class PresentsSession /** Queues a message for delivery to the client. */ protected boolean postMessage (DownstreamMessage msg) { - Connection conn = getConnection(); + PresentsConnection conn = getConnection(); if (conn != null) { conn.postMessage(msg); _messagesOut++; // count 'em up! @@ -1147,7 +1150,7 @@ public class PresentsSession protected AuthRequest _areq; protected Object _authdata; protected Name _authname; - protected Connection _conn; + protected PresentsConnection _conn; protected ClientObject _clobj; protected IntMap _subscrips = IntMaps.newHashIntMap(); protected ClassLoader _loader; diff --git a/src/main/java/com/threerings/presents/server/net/AuthingConnection.java b/src/main/java/com/threerings/presents/server/net/AuthingConnection.java index 119f989b3..f2b1b9150 100644 --- a/src/main/java/com/threerings/presents/server/net/AuthingConnection.java +++ b/src/main/java/com/threerings/presents/server/net/AuthingConnection.java @@ -33,7 +33,7 @@ import static com.threerings.presents.Log.log; * The authing connection manages the client connection until authentication has completed (for * better or for worse). */ -public class AuthingConnection extends Connection +public class AuthingConnection extends PresentsConnection { public AuthingConnection () { @@ -43,7 +43,7 @@ public class AuthingConnection extends Connection // keep a handle on our auth request _authreq = (AuthRequest)msg; // post ourselves for processing by the authmgr - _cmgr.authenticateConnection(AuthingConnection.this); + _pcmgr.authenticateConnection(AuthingConnection.this); } catch (ClassCastException cce) { log.warning("Received non-authreq message during authentication process", "conn", AuthingConnection.this, "msg", msg); diff --git a/src/main/java/com/threerings/presents/server/net/BindingConnectionManager.java b/src/main/java/com/threerings/presents/server/net/BindingConnectionManager.java deleted file mode 100644 index 8e9d9b79e..000000000 --- a/src/main/java/com/threerings/presents/server/net/BindingConnectionManager.java +++ /dev/null @@ -1,241 +0,0 @@ -// -// $Id$ -// -// Narya library - tools for developing networked games -// Copyright (C) 2002-2010 Three Rings Design, Inc., All Rights Reserved -// http://code.google.com/p/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.server.net; - -import java.util.List; - -import java.io.IOException; -import java.net.InetSocketAddress; -import java.nio.channels.DatagramChannel; -import java.nio.channels.SelectionKey; -import java.nio.channels.ServerSocketChannel; - -import com.google.common.base.Preconditions; -import com.google.common.collect.Lists; -import com.google.inject.Inject; -import com.google.inject.Singleton; - -import com.samskivert.net.AddressUtil; -import com.samskivert.util.Lifecycle; - -import com.threerings.presents.server.ReportManager; - -import static com.threerings.presents.Log.log; - -/** - * Binds tcp sockets and listens for datagrams in addition to ConnectionManager's normal duties. - */ -@Singleton -public class BindingConnectionManager extends ConnectionManager -{ - @Inject public BindingConnectionManager (Lifecycle cycle, ReportManager repmgr) - throws IOException - { - super(cycle, repmgr); - } - - /** - * Configures the connection manager with the hostname and ports on which it will listen for - * socket connections and datagram packets. This must be called before the connection manager - * is started (via {@via #start}) as the sockets will be bound at that time. - * - * @param socketHostname the hostname to which we bind our sockets or null to bind to all - * interfaces. - * @param datagramHostname the hostname to which we bind our datagram socket or null to bind - * to all interfaces. - * @param socketPorts the ports on which to listen for TCP connection. - * @param datagramPorts the ports on which to listen for datagram packets. - */ - public void init (String socketHostname, String datagramHostname, - int[] socketPorts, int[] datagramPorts) - throws IOException - { - Preconditions.checkNotNull(socketPorts, "Socket ports must be non-null."); - Preconditions.checkNotNull(datagramPorts, "Datagram ports must be non-null. " + - "Pass a zero-length array to bind no datagram ports."); - - _bindHostname = socketHostname; - _ports = socketPorts; - _datagramHostname = datagramHostname; - _datagramPorts = datagramPorts; - } - - @Override - protected void willStart () - { - super.willStart(); - - // open our TCP listening ports - int successes = 0; - for (int port : _ports) { - try { - acceptConnections(port); - successes++; - } catch (IOException ioe) { - log.warning("Failure listening to socket", "hostname", _bindHostname, - "port", port, ioe); - } - } - if (successes == 0) { - log.warning("ConnectionManager failed to bind to any ports. Shutting down."); - _server.queueShutdown(); - return; - } - - // open up the datagram ports as well - for (int port : _datagramPorts) { - try { - acceptDatagrams(port); - } catch (IOException ioe) { - log.warning("Failure opening datagram channel", "hostname", _datagramHostname, - "port", port, ioe); - } - } - } - - @Override - protected void didShutdown () - { - super.didShutdown(); - // TODO: consider closing the listen sockets earlier, like in the shutdown method - - // unbind our listening sockets; note: because we wait for the objmgr to exit before we do, - // we will still be accepting connections as long as there are events pending. - for (ServerSocketChannel ssocket : _ssockets) { - try { - ssocket.socket().close(); - } catch (IOException ioe) { - log.warning("Failed to close listening socket: " + ssocket, ioe); - } - } - - // unbind datagram channels, if any - for (DatagramChannel datagramChannel : _datagramChannels) { - datagramChannel.socket().close(); - } - } - - protected void acceptConnections (int port) - throws IOException - { - // create a listening socket - final ServerSocketChannel ssocket = ServerSocketChannel.open(); - ssocket.configureBlocking(false); - InetSocketAddress isa = AddressUtil.getAddress(_bindHostname, port); - ssocket.socket().bind(isa); - - // and add it to the select set - SelectionKey sk = ssocket.register(_selector, SelectionKey.OP_ACCEPT); - _handlers.put(sk, new NetEventHandler() { - public int handleEvent (long when) { - try { - handleAcceptedSocket(ssocket.accept()); - } catch (IOException ioe) { - log.info("Failure accepting connected socket: " +ioe); - } - // there's no easy way to measure bytes read when accepting a connection, so we - // claim nothing - return 0; - } - public boolean checkIdle (long now) { - return false; // we're never idle - } - public void becameIdle () { - // we're never idle - } - }); - _ssockets.add(ssocket); - log.info("Server listening on " + isa + "."); - } - - protected void acceptDatagrams (int port) - throws IOException - { - // create a channel - final DatagramChannel channel = DatagramChannel.open(); - channel.socket().setTrafficClass(0x10); // IPTOS_LOWDELAY - channel.configureBlocking(false); - InetSocketAddress isa = AddressUtil.getAddress(_datagramHostname, port); - channel.socket().bind(isa); - - // and add it to the select set - SelectionKey sk = channel.register(_selector, SelectionKey.OP_READ); - _handlers.put(sk, new NetEventHandler() { - public int handleEvent (long when) { - return handleDatagram(channel, when); - } - public boolean checkIdle (long now) { - return false;// Can't be idle - } - public void becameIdle () {} - }); - _datagramChannels.add(channel); - log.info("Server accepting datagrams on " + isa + "."); - } - - /** - * Called when a datagram message is ready to be read off its channel. - */ - protected int handleDatagram (DatagramChannel listener, long when) - { - InetSocketAddress source; - _databuf.clear(); - try { - source = (InetSocketAddress)listener.receive(_databuf); - } catch (IOException ioe) { - log.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, listener, _databuf, when); - } else { - log.debug("Received datagram for unknown connection", "id", connectionId, - "source", source); - } - - return size; - } - - protected int[] _ports, _datagramPorts; - protected String _bindHostname, _datagramHostname; - - protected List _ssockets = Lists.newArrayList(); - protected List _datagramChannels = Lists.newArrayList(); -} diff --git a/src/main/java/com/threerings/presents/server/net/DatagramChannelReader.java b/src/main/java/com/threerings/presents/server/net/DatagramChannelReader.java new file mode 100644 index 000000000..ac52bec77 --- /dev/null +++ b/src/main/java/com/threerings/presents/server/net/DatagramChannelReader.java @@ -0,0 +1,106 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2010 Three Rings Design, Inc., All Rights Reserved +// http://code.google.com/p/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.server.net; + +import java.util.List; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.channels.DatagramChannel; +import java.nio.channels.SelectionKey; + +import com.google.common.base.Preconditions; +import com.google.common.collect.Lists; + +import com.samskivert.net.AddressUtil; +import com.samskivert.util.Lifecycle; + +import com.threerings.nio.conman.NetEventHandler; + +import static com.threerings.presents.Log.log; + +/** + * Binds datagram connections on a given hostname for a set of ports, and passes datagrams read + * off those ports into a connection manager. + */ +public class DatagramChannelReader + implements Lifecycle.ShutdownComponent +{ + public DatagramChannelReader (String datagramHostname, + int[] datagramPorts, PresentsConnectionManager conMgr) + { + Preconditions.checkNotNull(datagramPorts, "Datagram ports must be non-null. " + + "Pass a zero-length array to bind no datagram ports."); + + _datagramHostname = datagramHostname; + _datagramPorts = datagramPorts; + _conMan = conMgr; + } + + public void bind() + { + for (int port : _datagramPorts) { + try { + acceptDatagrams(port); + } catch (IOException ioe) { + log.warning("Failure opening datagram channel", "hostname", _datagramHostname, + "port", port, ioe); + } + } + } + + public void shutdown () + { + for (DatagramChannel datagramChannel : _datagramChannels) { + datagramChannel.socket().close(); + } + } + + protected void acceptDatagrams (int port) + throws IOException + { + // create a channel + final DatagramChannel channel = DatagramChannel.open(); + channel.socket().setTrafficClass(0x10); // IPTOS_LOWDELAY + channel.configureBlocking(false); + InetSocketAddress isa = AddressUtil.getAddress(_datagramHostname, port); + channel.socket().bind(isa); + + // and add it to the select set + _conMan.register(channel, SelectionKey.OP_READ, new NetEventHandler() { + public int handleEvent (long when) { + return _conMan.handleDatagram(channel, when); + } + public boolean checkIdle (long now) { + return false;// Can't be idle + } + public void becameIdle () {} + }); + _datagramChannels.add(channel); + log.info("Server accepting datagrams on " + isa + "."); + } + + protected final int[] _datagramPorts; + protected final String _datagramHostname; + protected final List _datagramChannels = Lists.newArrayList(); + protected final PresentsConnectionManager _conMan; +} diff --git a/src/main/java/com/threerings/presents/server/net/Connection.java b/src/main/java/com/threerings/presents/server/net/PresentsConnection.java similarity index 64% rename from src/main/java/com/threerings/presents/server/net/Connection.java rename to src/main/java/com/threerings/presents/server/net/PresentsConnection.java index 5abfa6c95..e5fedfe85 100644 --- a/src/main/java/com/threerings/presents/server/net/Connection.java +++ b/src/main/java/com/threerings/presents/server/net/PresentsConnection.java @@ -23,11 +23,9 @@ package com.threerings.presents.server.net; import java.io.EOFException; import java.io.IOException; -import java.net.InetAddress; import java.net.InetSocketAddress; import java.nio.ByteBuffer; import java.nio.channels.DatagramChannel; -import java.nio.channels.SelectionKey; import java.nio.channels.SocketChannel; import java.security.MessageDigest; import java.security.NoSuchAlgorithmException; @@ -38,17 +36,18 @@ import com.threerings.io.ObjectInputStream; import com.threerings.io.ObjectOutputStream; import com.threerings.presents.net.Message; -import com.threerings.presents.net.PingRequest; import com.threerings.presents.util.DatagramSequencer; +import com.threerings.nio.conman.Connection; +import com.threerings.nio.conman.ConnectionManager; + import static com.threerings.presents.Log.log; /** - * The base connection class implements the net event handler interface and processes raw incoming - * network data into a stream of parsed {@link Message} objects. It also provides the means to send - * messages to the client and facilities for checking delinquency. + * Parses incoming network data into a stream of {@link Message} objects, sends messages to the + * client and adds datagram support to a connection. */ -public class Connection implements NetEventHandler +public class PresentsConnection extends Connection { /** Used with {@link Connection#setMessageHandler}. */ public static interface MessageHandler { @@ -56,23 +55,16 @@ public class Connection implements NetEventHandler void handleMessage (Message message); } - /** The key used by the NIO code to track this connection. */ - public SelectionKey selkey; - /** - * Initializes a connection object with a socket and related info. - * - * @param cmgr The connection manager with which this connection is associated. - * @param channel The socket channel from which we'll be reading messages. - * @param createStamp The time at which this connection was created. + * Initializes the connection with its channel. Must be called with a + * {@link PresentsConnectionManager} as cmgr. */ + @Override public void init (ConnectionManager cmgr, SocketChannel channel, long createStamp) throws IOException { - _cmgr = cmgr; - _channel = channel; - _lastEvent = createStamp; - _connectionId = ++_lastConnectionId; + super.init(cmgr, channel, createStamp); + _pcmgr = (PresentsConnectionManager)cmgr; } /** @@ -95,31 +87,6 @@ public class Connection implements NetEventHandler } } - /** - * Returns the connection's unique identifier. - */ - public int getConnectionId () - { - return _connectionId; - } - - /** - * Returns the non-blocking socket object used to construct this connection. - */ - public SocketChannel getChannel () - { - return _channel; - } - - /** - * Returns the address associated with this connection or null if it has no underlying socket - * channel. - */ - public InetAddress getInetAddress () - { - return (_channel == null) ? null : _channel.socket().getInetAddress(); - } - /** * Sets whether we should transmit datagrams. */ @@ -166,42 +133,6 @@ public class Connection implements NetEventHandler } } - /** - * Returns true if this connection is closed. - */ - public boolean isClosed () - { - return (_channel == null); - } - - /** - * Closes this connection and unregisters it from the connection manager. This should only be - * called from the conmgr thread. - */ - public void close () - { - // we shouldn't be closed twice - if (isClosed()) { - log.warning("Attempted to re-close connection " + this + ".", new Exception()); - return; - } - - // unregister from the select set - _cmgr.connectionClosed(this); - - // close our socket - closeSocket(); - } - - /** - * Queues up a request to have this connection closed by the connection manager once all - * messages in its queue have been written to its target. - */ - public void asyncClose () - { - _cmgr.postAsyncClose(this); - } - /** * Posts a message for delivery to this connection. The message will be delivered by the conmgr * thread as soon as it gets to it. @@ -209,36 +140,7 @@ public class Connection implements NetEventHandler public void postMessage (Message msg) { // pass this along to the connection manager - _cmgr.postMessage(this, msg); - } - - /** - * Called when an outgoing socket experiences a connect failure. The connection manager will - * have cleaned up the partial registration needed during the connect process, so we are only - * responsible for closing our socket. - */ - public void connectFailure (IOException ioe) - { - closeSocket(); - } - - /** - * Called when there is a failure reading or writing to this connection. We notify the - * connection manager and close ourselves down. - */ - public void networkFailure (IOException ioe) - { - // if we're already closed, then something is seriously funny - if (isClosed()) { - log.warning("Failure reported on closed connection " + this + ".", new Exception()); - return; - } - - // let the connection manager know we're hosed - _cmgr.connectionFailed(this, ioe); - - // and close our socket - closeSocket(); + _pcmgr.postMessage(this, msg); } /** @@ -255,7 +157,7 @@ public class Connection implements NetEventHandler log.warning("Missing MD5 algorithm."); return; } - _sequencer = _cmgr.createDatagramSequencer(); + _sequencer = _pcmgr.createDatagramSequencer(); } // verify the hash @@ -291,7 +193,6 @@ public class Connection implements NetEventHandler } } - // from interface NetEventHandler public int handleEvent (long when) { // make a note that we received an event as of this time @@ -345,32 +246,6 @@ public class Connection implements NetEventHandler return bytesIn; } - // from interface NetEventHandler - public boolean checkIdle (long now) - { - long idleMillis = now - _lastEvent; - if (idleMillis < PingRequest.PING_INTERVAL + LATENCY_GRACE) { - return false; - } - if (isClosed()) { - return true; - } - log.info("Disconnecting non-communicative client", "conn", this, "idle", idleMillis + "ms"); - return true; - } - - // from interface NetEventHandler - public void becameIdle () - { - _cmgr.closeConnection(this); - } - - @Override // from Object - public String toString () - { - return "[id=" + (hashCode() % 1000) + ", addr=" + getInetAddress() + "]"; - } - /** * Returns the object input stream associated with this connection. This should only be used * by the connection manager. @@ -385,7 +260,7 @@ public class Connection implements NetEventHandler * is called by the connection manager when the time comes to pass streams from the authing * connection to the running connection. */ - protected void inheritStreams (Connection other) + protected void inheritStreams (PresentsConnection other) { _fin = other._fin; _oin = other._oin; @@ -427,34 +302,6 @@ public class Connection implements NetEventHandler 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. - */ - protected void closeSocket () - { - if (_channel == null) { - return; - } - - log.debug("Closing channel " + this + "."); - try { - _channel.close(); - } catch (IOException ioe) { - log.warning("Error closing connection", "conn", this, "error", ioe); - } - - // clear out our references to prevent repeat closings - _channel = null; - } - - protected ConnectionManager _cmgr; - protected SocketChannel _channel; - - protected long _lastEvent; - - protected int _connectionId; - protected FramedInputStream _fin; protected ObjectInputStream _oin; protected ObjectOutputStream _oout; @@ -470,10 +317,5 @@ public class Connection implements NetEventHandler 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; + protected PresentsConnectionManager _pcmgr; } diff --git a/src/main/java/com/threerings/presents/server/net/PresentsConnectionManager.java b/src/main/java/com/threerings/presents/server/net/PresentsConnectionManager.java new file mode 100644 index 000000000..aeca0f025 --- /dev/null +++ b/src/main/java/com/threerings/presents/server/net/PresentsConnectionManager.java @@ -0,0 +1,490 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2010 Three Rings Design, Inc., All Rights Reserved +// http://code.google.com/p/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.server.net; + +import java.util.List; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.net.InetSocketAddress; +import java.nio.ByteBuffer; +import java.nio.channels.DatagramChannel; +import java.nio.channels.SelectionKey; +import java.nio.channels.SocketChannel; + +import com.google.common.collect.Lists; +import com.google.inject.Inject; +import com.google.inject.Singleton; + +import com.samskivert.util.Invoker; +import com.samskivert.util.Lifecycle; +import com.samskivert.util.Queue; +import com.samskivert.util.ResultListener; +import com.samskivert.util.Tuple; + +import com.threerings.io.ByteBufferInputStream; +import com.threerings.io.FramingOutputStream; +import com.threerings.io.ObjectOutputStream; +import com.threerings.io.UnreliableObjectInputStream; +import com.threerings.io.UnreliableObjectOutputStream; + +import com.threerings.presents.annotation.AuthInvoker; +import com.threerings.presents.client.Client; +import com.threerings.presents.data.ConMgrStats; +import com.threerings.presents.net.Message; +import com.threerings.presents.net.PongResponse; +import com.threerings.presents.net.Transport; +import com.threerings.presents.server.Authenticator; +import com.threerings.presents.server.ChainedAuthenticator; +import com.threerings.presents.server.ClientManager; +import com.threerings.presents.server.DummyAuthenticator; +import com.threerings.presents.server.PresentsDObjectMgr; +import com.threerings.presents.server.ReportManager; +import com.threerings.presents.util.DatagramSequencer; + +import com.threerings.nio.conman.Connection; +import com.threerings.nio.conman.ConnectionManager; +import com.threerings.nio.conman.NetEventHandler; + +import static com.threerings.presents.Log.log; + +@Singleton +public class PresentsConnectionManager extends ConnectionManager +{ + + @Inject + public PresentsConnectionManager (Lifecycle cycle, ReportManager repmgr) + throws IOException + { + super(cycle, repmgr); + } + + @Override + public synchronized ConMgrStats getStats () + { + _stats.authQueueSize = _authq.size(); + return super.getStats(); + } + + /** + * Adds an authenticator to the authentication chain. This authenticator will be offered a + * chance to authenticate incoming connections before falling back to the main authenticator. + */ + public void addChainedAuthenticator (ChainedAuthenticator author) + { + _authors.add(author); + } + + /** + * Called when a datagram message is ready to be read off its channel. + */ + protected int handleDatagram (DatagramChannel listener, long when) + { + InetSocketAddress source; + _databuf.clear(); + try { + source = (InetSocketAddress)listener.receive(_databuf); + } catch (IOException ioe) { + log.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) { + ((PresentsConnection)conn).handleDatagram(source, listener, _databuf, when); + } else { + log.debug("Received datagram for unknown connection", "id", connectionId, + "source", source); + } + + 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 PresentsConnection#postMessage} which happens when forwarding an event to a client + * and at the completion of authentication, both of which must happen only on the + * distributed object thread. + */ + protected void postMessage (PresentsConnection conn, Message msg) + { + if (!isRunning()) { + log.warning("Posting message to inactive connection manager", + "msg", msg, new Exception()); + } + + // sanity check + if (conn == null || msg == null) { + log.warning("postMessage() bogosity", "conn", conn, "msg", msg, new Exception()); + return; + } + + // more sanity check; messages must only be posted from the dobjmgr thread + if (!_omgr.isDispatchThread()) { + log.warning("Message posted on non-distributed object thread", "conn", conn, + "msg", msg, "thread", Thread.currentThread(), new Exception()); + // let it through though as we don't want to break things unnecessarily + } + + try { + // send it as a datagram if hinted and possible (pongs must be sent as part of the + // negotation process) + if (!msg.getTransport().isReliable() && + (conn.getTransmitDatagrams() || msg instanceof PongResponse) && + postDatagram(conn, msg)) { + return; + } + + // note the actual transport + msg.noteActualTransport(Transport.RELIABLE_ORDERED); + + _framer.resetFrame(); + + // flatten this message using the connection's output stream + 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); + // log.info("Flattened " + msg + " into " + data.length + " bytes."); + + // and slap both on the queue + _outq.append(Tuple.newTuple(conn, data)); + + } catch (Exception e) { + log.warning("Failure flattening message", "conn", conn, "msg", msg, e); + } + } + + /** + * Helper function for {@link #postMessage}; handles posting the message as a datagram. + * + * @return true if the datagram was successfully posted, false if it was too big. + */ + protected boolean postDatagram (PresentsConnection conn, Message msg) + throws Exception + { + _flattener.reset(); + + // flatten the message using the connection's sequencer + DatagramSequencer sequencer = conn.getDatagramSequencer(); + sequencer.writeDatagram(msg); + + // if the message is too big, we must fall back to sending it through the stream channel + if (_flattener.size() > Client.MAX_DATAGRAM_SIZE) { + return false; + } + + // note the actual transport + msg.noteActualTransport(Transport.UNRELIABLE_UNORDERED); + + // extract as a byte array + byte[] data = _flattener.toByteArray(); + + // slap it on the queue + _dataq.append(Tuple.newTuple(conn, data)); + + return true; + } + + /** + * Creates a datagram sequencer for use by a {@link Connection}. + */ + protected DatagramSequencer createDatagramSequencer () + { + return new DatagramSequencer( + new UnreliableObjectInputStream(new ByteBufferInputStream(_databuf)), + new UnreliableObjectOutputStream(_flattener)); + } + + /** + * Opens an outgoing connection to the supplied address. The connection will be opened in a + * non-blocking manner and added to the connection manager's select set. Messages posted to the + * connection prior to it being actually connected to its destination will remain in the queue. + * If the connection fails those messages will be dropped. + * + * @param conn the connection to be initialized and opened. Callers may want to provide a + * {@link Connection} derived class so that they may intercept calldown methods. + * @param hostname the hostname of the server to which to connect. + * @param port the port on which to connect to the server. + * + * @exception IOException thrown if an error occurs creating our socket. Everything else + * happens asynchronously. If the connection attempt fails, the Connection will be notified via + * {@link Connection#networkFailure}. + */ + public void openOutgoingConnection (Connection conn, String hostname, int port) + throws IOException + { + // create a socket channel to use for this connection, initialize it and queue it up to + // have the non-blocking connect process started + SocketChannel sockchan = SocketChannel.open(); + sockchan.configureBlocking(false); + conn.init(this, sockchan, System.currentTimeMillis()); + _connectq.append(Tuple.newTuple(conn, new InetSocketAddress(hostname, port))); + } + + /** + * Starts the connection process for an outgoing connection. This is called as part of the + * conmgr tick for any pending outgoing connections. + */ + protected void startOutgoingConnection (final Connection conn, InetSocketAddress addr) + { + final SocketChannel sockchan = conn.getChannel(); + try { + // register our channel with the selector (if this fails, we abandon ship immediately) + conn.selkey = sockchan.register(_selector, SelectionKey.OP_CONNECT); + + // start our connection process (now if we fail we need to clean things up) + NetEventHandler handler; + if (sockchan.connect(addr)) { + // it is possible even for a non-blocking socket to connect immediately, in which + // case we stick the connection in as its event handler immediately + handler = conn; + + } 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 now) { + return conn.checkIdle(now); + } + 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); + } + }; + } + _handlers.put(conn.selkey, handler); + + } catch (IOException ioe) { + log.warning("Failed to initiate connection for " + sockchan + ".", ioe); + conn.connectFailure(ioe); // nothing else to clean up + } + } + + @Override // from LoopingThread + public boolean isRunning () + { + // Prevent exiting our thread until the object manager is done. + return super.isRunning() || _omgr.isRunning(); + } + + @Override + protected void handleIncoming (long iterStamp) + { + super.handleIncoming(iterStamp); + + // start up any outgoing connections that need to be connected + Tuple pconn; + while ((pconn = _connectq.getNonBlocking()) != null) { + startOutgoingConnection(pconn.left, pconn.right); + } + + // check for connections that have completed authentication + processAuthedConnections(iterStamp); + } + + @Override + protected void connectionFailed (Connection conn, IOException ioe) + { + super.connectionFailed(conn, ioe); + + // let the client manager know what's up + _clmgr.connectionFailed(conn, ioe); + } + + @Override + protected void connectionClosed (Connection conn) + { + super.connectionClosed(conn); + + // let the client manager know what's up + _clmgr.connectionClosed(conn); + } + + /** + * Performs the authentication process on the specified connection. This is called by {@link + * AuthingConnection} itself once it receives its auth request. + */ + protected void authenticateConnection (AuthingConnection conn) + { + Authenticator author = _author; + for (ChainedAuthenticator cauthor : _authors) { + if (cauthor.shouldHandleConnection(conn)) { + author = cauthor; + break; + } + } + + author.authenticateConnection(_authInvoker, conn, new ResultListener() { + public void requestCompleted (AuthingConnection conn) { + _authq.append(conn); + } + public void requestFailed (Exception cause) { + // this never happens + } + }); + } + + /** + * Starts an accepted socket down the path to authorization. + */ + @Override + protected void handleAcceptedSocket (SocketChannel channel) + { + handleAcceptedSocket(channel, new AuthingConnection()); + } + + /** + * Converts connections that have completed the authentication process into full running + * connections and notifies the client manager that new connections have been established. + */ + protected void processAuthedConnections (long iterStamp) + { + AuthingConnection conn; + while ((conn = _authq.getNonBlocking()) != null) { + try { + // construct a new running connection to handle this connections network traffic + // from here on out + PresentsConnection rconn = new PresentsConnection(); + rconn.init(this, conn.getChannel(), iterStamp); + rconn.selkey = conn.selkey; + + // we need to keep using the same object input and output streams from the + // beginning of the session because they have context that needs to be preserved + rconn.inheritStreams(conn); + + // replace the mapping in the handlers table from the old conn with the new one + _handlers.put(rconn.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) { + _oflowqs.put(rconn, oflowHandler); + } + + // and let the client manager know about our new connection + _clmgr.connectionEstablished(rconn, conn.getAuthName(), conn.getAuthRequest(), + conn.getAuthResponse()); + + } catch (IOException ioe) { + log.warning("Failure upgrading authing connection to running.", ioe); + } + } + } + + @Override + protected void sendOutgoingMessages (long iterStamp) + { + super.sendOutgoingMessages(iterStamp); + + // send any datagrams + Tuple tup; + while ((tup = _dataq.getNonBlocking()) != null) { + writeDatagram(tup.left, tup.right); + } + } + + /** + * 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 (PresentsConnection 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 conn.getDatagramChannel().send(_databuf, target) > 0; + } catch (IOException ioe) { + log.warning("Failed to send datagram.", ioe); + return false; + } + } + + /** 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. */ + @Inject(optional=true) protected Authenticator _author = new DummyAuthenticator(); + protected List _authors = Lists.newArrayList(); + + protected Queue _authq = Queue.newQueue(); + protected Queue> _connectq = Queue.newQueue(); + + protected FramingOutputStream _framer = new FramingOutputStream(); + protected ByteArrayOutputStream _flattener = new ByteArrayOutputStream(); + + // some dependencies + @Inject @AuthInvoker protected Invoker _authInvoker; + @Inject protected ClientManager _clmgr; + @Inject protected PresentsDObjectMgr _omgr; + + protected Queue> _dataq = Queue.newQueue(); + protected ByteBuffer _databuf = ByteBuffer.allocateDirect(Client.MAX_DATAGRAM_SIZE); +} diff --git a/src/main/java/com/threerings/presents/server/net/ServerCommunicator.java b/src/main/java/com/threerings/presents/server/net/ServerCommunicator.java index 5cba1d1e0..09bb7b45c 100644 --- a/src/main/java/com/threerings/presents/server/net/ServerCommunicator.java +++ b/src/main/java/com/threerings/presents/server/net/ServerCommunicator.java @@ -45,7 +45,8 @@ import static com.threerings.presents.Log.log; */ public class ServerCommunicator extends Communicator { - public ServerCommunicator (Client client, ConnectionManager conmgr, RootDObjectManager rootmgr) + public ServerCommunicator (Client client, PresentsConnectionManager conmgr, + RootDObjectManager rootmgr) { super(client); _conmgr = conmgr; @@ -62,7 +63,7 @@ public class ServerCommunicator extends Communicator // we assume server entities have no firewall issues and can connect on the first port try { - Connection conn = new Connection() { + PresentsConnection conn = new PresentsConnection() { @Override public void postMessage (Message msg) { super.postMessage(msg); // outgoing traffic on this connection is used to prevent idleness @@ -92,7 +93,7 @@ public class ServerCommunicator extends Communicator shutdown(); } }; - conn.setMessageHandler(new Connection.MessageHandler() { + conn.setMessageHandler(new PresentsConnection.MessageHandler() { public void handleMessage (Message message) { try { // our first message will always be an auth response @@ -177,7 +178,7 @@ public class ServerCommunicator extends Communicator super.logonSucceeded(data); // now we can route all messages to the ClientDObjectMgr - _conn.setMessageHandler(new Connection.MessageHandler() { + _conn.setMessageHandler(new PresentsConnection.MessageHandler() { public void handleMessage (Message message) { if (PING_DEBUG && message instanceof PongResponse) { log.info("Got pong from server " + message); @@ -200,9 +201,9 @@ public class ServerCommunicator extends Communicator clientCleanup(_logonError); } - protected ConnectionManager _conmgr; + protected PresentsConnectionManager _conmgr; protected RootDObjectManager _rootmgr; - protected Connection _conn; + protected PresentsConnection _conn; protected ClassLoader _loader; protected Exception _logonError;