diff --git a/src/java/com/threerings/crowd/chat/server/ChatProvider.java b/src/java/com/threerings/crowd/chat/server/ChatProvider.java index abf65edb8..ca27e2c01 100644 --- a/src/java/com/threerings/crowd/chat/server/ChatProvider.java +++ b/src/java/com/threerings/crowd/chat/server/ChatProvider.java @@ -65,15 +65,41 @@ public class ChatProvider String message); } + /** Used to forward tells between servers in a multi-server setup. */ + public static interface TellForwarder + { + /** + * Requests that the supplied tell message be delivered to the + * appropriate destination. + * + * @return true if the tell was delivered, false otherwise. + */ + public boolean forwardTell (Name teller, Name target, String message, + TellListener listener); + } + /** * Set the auto tell responder for the chat provider. Only one auto - * responder is allowed. + * responder is allowed. Note: this only works for same-server + * tells. If the tell is forwarded to another server, no auto-response + * opportunity is provided (because we never have both body objects in the + * same place). */ public static void setTellAutoResponder (TellAutoResponder autoRespond) { _autoRespond = autoRespond; } + /** + * Configures the tell forwarded for the chat provider. This is used by the + * Crowd peer services to forward tells between servers in a multi-server + * cluster. + */ + public static void setTellForwarder (TellForwarder forwarder) + { + _tellForwarder = forwarder; + } + /** * Set an object to which all broadcasts should be sent, rather * than iterating over the place objects and sending to each of them. @@ -114,36 +140,14 @@ public class ChatProvider throw new InvocationException(errmsg); } - // make sure the target user is online - BodyObject tobj = CrowdServer.lookupBody(target); - if (tobj == null) { - throw new InvocationException(USER_NOT_ONLINE); - } + // deliver the tell message to the target + deliverTell(source.getVisibleName(), target, message, listener); - if (tobj.status == OccupantInfo.DISCONNECTED) { - throw new InvocationException(MessageBundle.compose( - USER_DISCONNECTED, TimeUtil.getTimeOrderString( - System.currentTimeMillis() - tobj.statusTime, - TimeUtil.SECOND))); - } - - // deliver a tell notification to the target player - sendTellMessage(tobj, source.getVisibleName(), null, message); - - // let the teller know it went ok - long idle = 0L; - if (tobj.status == OccupantInfo.IDLE) { - idle = System.currentTimeMillis() - tobj.statusTime; - } - String awayMessage = null; - if (!StringUtil.isBlank(tobj.awayMessage)) { - awayMessage = tobj.awayMessage; - } - listener.tellSucceeded(idle, awayMessage); - - // do the autoresponse if needed - if (_autoRespond != null) { - _autoRespond.sentTell(source, tobj, message); + // inform the auto-responder if needed + BodyObject targobj; + if (_autoRespond != null && + (targobj = CrowdServer.lookupBody(target)) != null) { + _autoRespond.sentTell(source, targobj, message); } } @@ -221,6 +225,49 @@ public class ChatProvider body.setAwayMessage(message); } + /** + * Delivers a tell message from the specified teller to the specified + * target. It is assumed that the teller has already been permissions + * checked. + */ + public static void deliverTell (Name teller, Name target, String message, + TellListener listener) + throws InvocationException + { + // make sure the target user is online + BodyObject tobj = CrowdServer.lookupBody(target); + if (tobj == null) { + // if we have a forwarder configured, try forwarding the tell + if (_tellForwarder != null && + _tellForwarder.forwardTell(teller, target, message, listener)) { + return; + } + throw new InvocationException(USER_NOT_ONLINE); + } + + if (tobj.status == OccupantInfo.DISCONNECTED) { + String errmsg = MessageBundle.compose( + USER_DISCONNECTED, TimeUtil.getTimeOrderString( + System.currentTimeMillis() - tobj.statusTime, + TimeUtil.SECOND)); + throw new InvocationException(errmsg); + } + + // deliver a tell notification to the target player + sendTellMessage(tobj, teller, null, message); + + // let the teller know it went ok + long idle = 0L; + if (tobj.status == OccupantInfo.IDLE) { + idle = System.currentTimeMillis() - tobj.statusTime; + } + String awayMessage = null; + if (!StringUtil.isBlank(tobj.awayMessage)) { + awayMessage = tobj.awayMessage; + } + listener.tellSucceeded(idle, awayMessage); + } + /** * Delivers a tell notification to the specified target player, * originating with the specified speaker. @@ -249,9 +296,12 @@ public class ChatProvider /** The distributed object manager used by the chat services. */ protected static DObjectManager _omgr; - /** Reference to our auto responder object. */ + /** Generates auto-responses to tells. May be null. */ protected static TellAutoResponder _autoRespond; + /** Forwards tells between servers. May be null. */ + protected static TellForwarder _tellForwarder; + /** An alternative object to which broadcasts should be sent. */ protected static DObject _broadcastObject; } diff --git a/src/java/com/threerings/crowd/peer/client/CrowdPeerService.java b/src/java/com/threerings/crowd/peer/client/CrowdPeerService.java index 8dc7ea14d..913a5329d 100644 --- a/src/java/com/threerings/crowd/peer/client/CrowdPeerService.java +++ b/src/java/com/threerings/crowd/peer/client/CrowdPeerService.java @@ -21,11 +21,12 @@ package com.threerings.crowd.peer.client; +import com.threerings.util.Name; + import com.threerings.presents.client.Client; import com.threerings.presents.client.InvocationService; import com.threerings.crowd.chat.client.ChatService; -import com.threerings.crowd.chat.data.ChatMessage; /** * Bridges certain Crowd services between peers in a cluster configuration. @@ -33,8 +34,9 @@ import com.threerings.crowd.chat.data.ChatMessage; public interface CrowdPeerService extends InvocationService { /** - * Forwards a tell request to the server on which a user actually resides. + * Used to forward a tell request to the server on which the destination + * user actually occupies. */ - public void deliverTell ( - Client client, ChatMessage message, ChatService.TellListener listener); + public void deliverTell (Client client, Name teller, Name target, + String message, ChatService.TellListener listener); } diff --git a/src/java/com/threerings/crowd/peer/data/CrowdClientInfo.java b/src/java/com/threerings/crowd/peer/data/CrowdClientInfo.java index 4081d5160..95c4b561f 100644 --- a/src/java/com/threerings/crowd/peer/data/CrowdClientInfo.java +++ b/src/java/com/threerings/crowd/peer/data/CrowdClientInfo.java @@ -32,4 +32,14 @@ public class CrowdClientInfo extends ClientInfo { /** The client's visible name, which is used for chatting. */ public Name visibleName; + + @Override // documentation inherited + public Comparable getKey () + { + // the PeerManager works in such a way that we can override our client + // info key and things still work properly; all inter-server + // communication regarding users will be based on visible name so this + // makes lookups much more efficient + return visibleName; + } } diff --git a/src/java/com/threerings/crowd/peer/data/CrowdPeerMarshaller.java b/src/java/com/threerings/crowd/peer/data/CrowdPeerMarshaller.java index d670ef6b2..d871a8abb 100644 --- a/src/java/com/threerings/crowd/peer/data/CrowdPeerMarshaller.java +++ b/src/java/com/threerings/crowd/peer/data/CrowdPeerMarshaller.java @@ -23,11 +23,11 @@ package com.threerings.crowd.peer.data; import com.threerings.crowd.chat.client.ChatService; import com.threerings.crowd.chat.data.ChatMarshaller; -import com.threerings.crowd.chat.data.ChatMessage; import com.threerings.crowd.peer.client.CrowdPeerService; import com.threerings.presents.client.Client; import com.threerings.presents.data.InvocationMarshaller; import com.threerings.presents.dobj.InvocationResponseEvent; +import com.threerings.util.Name; /** * Provides the implementation of the {@link CrowdPeerService} interface @@ -43,12 +43,12 @@ public class CrowdPeerMarshaller extends InvocationMarshaller public static final int DELIVER_TELL = 1; // documentation inherited from interface - public void deliverTell (Client arg1, ChatMessage arg2, ChatService.TellListener arg3) + public void deliverTell (Client arg1, Name arg2, Name arg3, String arg4, ChatService.TellListener arg5) { - ChatMarshaller.TellMarshaller listener3 = new ChatMarshaller.TellMarshaller(); - listener3.listener = arg3; + ChatMarshaller.TellMarshaller listener5 = new ChatMarshaller.TellMarshaller(); + listener5.listener = arg5; sendRequest(arg1, DELIVER_TELL, new Object[] { - arg2, listener3 + arg2, arg3, arg4, listener5 }); } diff --git a/src/java/com/threerings/crowd/peer/server/CrowdPeerDispatcher.java b/src/java/com/threerings/crowd/peer/server/CrowdPeerDispatcher.java index e4d4b9feb..034c89c6b 100644 --- a/src/java/com/threerings/crowd/peer/server/CrowdPeerDispatcher.java +++ b/src/java/com/threerings/crowd/peer/server/CrowdPeerDispatcher.java @@ -23,7 +23,6 @@ package com.threerings.crowd.peer.server; import com.threerings.crowd.chat.client.ChatService; import com.threerings.crowd.chat.data.ChatMarshaller; -import com.threerings.crowd.chat.data.ChatMessage; import com.threerings.crowd.peer.client.CrowdPeerService; import com.threerings.crowd.peer.data.CrowdPeerMarshaller; import com.threerings.presents.client.Client; @@ -31,6 +30,7 @@ import com.threerings.presents.data.ClientObject; import com.threerings.presents.data.InvocationMarshaller; import com.threerings.presents.server.InvocationDispatcher; import com.threerings.presents.server.InvocationException; +import com.threerings.util.Name; /** * Dispatches requests to the {@link CrowdPeerProvider}. @@ -61,7 +61,7 @@ public class CrowdPeerDispatcher extends InvocationDispatcher case CrowdPeerMarshaller.DELIVER_TELL: ((CrowdPeerProvider)provider).deliverTell( source, - (ChatMessage)args[0], (ChatService.TellListener)args[1] + (Name)args[0], (Name)args[1], (String)args[2], (ChatService.TellListener)args[3] ); return; diff --git a/src/java/com/threerings/crowd/peer/server/CrowdPeerManager.java b/src/java/com/threerings/crowd/peer/server/CrowdPeerManager.java index b703fd91d..6aec27d50 100644 --- a/src/java/com/threerings/crowd/peer/server/CrowdPeerManager.java +++ b/src/java/com/threerings/crowd/peer/server/CrowdPeerManager.java @@ -25,15 +25,31 @@ import com.samskivert.io.PersistenceException; import com.samskivert.jdbc.ConnectionProvider; import com.samskivert.util.Invoker; +import com.threerings.util.Name; + +import com.threerings.presents.data.ClientObject; +import com.threerings.presents.server.InvocationException; +import com.threerings.presents.server.PresentsClient; + +import com.threerings.presents.peer.data.ClientInfo; import com.threerings.presents.peer.data.NodeObject; import com.threerings.presents.peer.server.PeerManager; +import com.threerings.crowd.chat.client.ChatService; +import com.threerings.crowd.chat.data.ChatMessage; +import com.threerings.crowd.chat.server.ChatProvider; +import com.threerings.crowd.data.BodyObject; +import com.threerings.crowd.server.CrowdServer; + +import com.threerings.crowd.peer.data.CrowdClientInfo; import com.threerings.crowd.peer.data.CrowdNodeObject; +import com.threerings.crowd.peer.data.CrowdPeerMarshaller; /** * Extends the standard peer manager and bridges certain Crowd services. */ public class CrowdPeerManager extends PeerManager + implements CrowdPeerProvider, ChatProvider.TellForwarder { /** * Creates a peer manager that integrates Crowd services across a cluster @@ -45,9 +61,81 @@ public class CrowdPeerManager extends PeerManager super(conprov, invoker); } + // documentation inherited from interface CrowdPeerProvider + public void deliverTell (ClientObject caller, Name teller, Name target, + String message, ChatService.TellListener listener) + throws InvocationException + { + // the originating server has already permissions checked the teller, + // so we just forward the message as if it originated on this server + ChatProvider.deliverTell(teller, target, message, listener); + } + + // documentation inherited from interface ChatProvider.TellForwarder + public boolean forwardTell (Name teller, Name target, String message, + ChatService.TellListener listener) + { + // look through our peer objects to see if the target user is online on + // one of the other servers + for (PeerNode peer : _peers.values()) { + CrowdNodeObject cnobj = (CrowdNodeObject)peer.nodeobj; + CrowdClientInfo cinfo = (CrowdClientInfo)cnobj.clients.get(target); + if (cinfo != null) { + cnobj.crowdPeerService.deliverTell( + peer.getClient(), teller, target, message, listener); + return true; + } + } + return false; + } + + @Override // documentation inherited + public void shutdown () + { + super.shutdown(); + + // unregister our invocation service + if (_nodeobj != null) { + CrowdServer.invmgr.clearDispatcher( + ((CrowdNodeObject)_nodeobj).crowdPeerService); + } + + // clear our tell forwarder registration + ChatProvider.setTellForwarder(null); + } + + @Override // documentation inherited + protected void finishInit (NodeObject nodeobj) + { + super.finishInit(nodeobj); + + // register and initialize our invocation service + CrowdNodeObject cnobj = (CrowdNodeObject)nodeobj; + cnobj.setCrowdPeerService( + (CrowdPeerMarshaller)CrowdServer.invmgr.registerDispatcher( + new CrowdPeerDispatcher(this), false)); + + // register ourselves as a tell forwarder + ChatProvider.setTellForwarder(this); + } + @Override // documentation inherited protected Class getNodeObjectClass () { return CrowdNodeObject.class; } + + @Override // documentation inherited + protected ClientInfo createClientInfo () + { + return new CrowdClientInfo(); + } + + @Override // documentation inherited + protected void initClientInfo (PresentsClient client, ClientInfo info) + { + super.initClientInfo(client, info); + ((CrowdClientInfo)info).visibleName = + ((BodyObject)client.getClientObject()).getVisibleName(); + } } diff --git a/src/java/com/threerings/crowd/peer/server/CrowdPeerProvider.java b/src/java/com/threerings/crowd/peer/server/CrowdPeerProvider.java index a289067c4..28bf43ebe 100644 --- a/src/java/com/threerings/crowd/peer/server/CrowdPeerProvider.java +++ b/src/java/com/threerings/crowd/peer/server/CrowdPeerProvider.java @@ -23,12 +23,12 @@ package com.threerings.crowd.peer.server; import com.threerings.crowd.chat.client.ChatService; import com.threerings.crowd.chat.data.ChatMarshaller; -import com.threerings.crowd.chat.data.ChatMessage; import com.threerings.crowd.peer.client.CrowdPeerService; import com.threerings.presents.client.Client; import com.threerings.presents.data.ClientObject; import com.threerings.presents.server.InvocationException; import com.threerings.presents.server.InvocationProvider; +import com.threerings.util.Name; /** * Defines the server-side of the {@link CrowdPeerService}. @@ -38,6 +38,6 @@ public interface CrowdPeerProvider extends InvocationProvider /** * Handles a {@link CrowdPeerService#deliverTell} request. */ - public void deliverTell (ClientObject caller, ChatMessage arg1, ChatService.TellListener arg2) + public void deliverTell (ClientObject caller, Name arg1, Name arg2, String arg3, ChatService.TellListener arg4) throws InvocationException; } diff --git a/src/java/com/threerings/presents/peer/server/PeerManager.java b/src/java/com/threerings/presents/peer/server/PeerManager.java index c2654d53d..df8d8ea13 100644 --- a/src/java/com/threerings/presents/peer/server/PeerManager.java +++ b/src/java/com/threerings/presents/peer/server/PeerManager.java @@ -35,8 +35,11 @@ import com.threerings.presents.client.Client; import com.threerings.presents.client.ClientObserver; import com.threerings.presents.dobj.ObjectAccessException; import com.threerings.presents.dobj.Subscriber; +import com.threerings.presents.server.ClientManager; +import com.threerings.presents.server.PresentsClient; import com.threerings.presents.server.PresentsServer; +import com.threerings.presents.peer.data.ClientInfo; import com.threerings.presents.peer.data.NodeObject; import com.threerings.presents.peer.net.PeerBootstrapData; import com.threerings.presents.peer.net.PeerCreds; @@ -52,6 +55,7 @@ import static com.threerings.presents.Log.log; * communicate cross-node information. */ public class PeerManager + implements ClientManager.ClientObserver { /** * Creates a peer manager which will create a {@link NodeRepository} which @@ -124,6 +128,9 @@ public class PeerManager */ public void shutdown () { + // clear out our client observer registration + PresentsServer.clmgr.removeClientObserver(this); + // TODO: clear our record from the node table for (PeerNode peer : _peers.values()) { peer.shutdown(); @@ -139,6 +146,43 @@ public class PeerManager creds.getNodeName(), _sharedSecret).equals(creds.getPassword()); } + // documentation inherited from interface ClientManager.ClientObserver + public void clientSessionDidStart (PresentsClient client) + { + // create and publish a ClientInfo record for this client + ClientInfo clinfo = createClientInfo(); + initClientInfo(client, clinfo); + + // sanity check + if (_nodeobj.clients.contains(clinfo)) { + log.warning("Received clientSessionDidStart() for already " + + "registered client!? " + + "[old=" + _nodeobj.clients.get(clinfo.getKey()) + + ", new=" + clinfo + "]."); + // go ahead and update the record + _nodeobj.updateClients(clinfo); + } else { + _nodeobj.addToClients(clinfo); + } + } + + // documentation inherited from interface ClientManager.ClientObserver + public void clientSessionDidEnd (PresentsClient client) + { + // we create a new client info for this client so that we can support + // derived classes overriding the value we use for the DSet key + ClientInfo clinfo = createClientInfo(); + initClientInfo(client, clinfo); + + // sanity check + if (!_nodeobj.clients.containsKey(clinfo.getKey())) { + log.warning("Session ended for unregistered client " + + "[info=" + clinfo + "]."); + } else { + _nodeobj.removeFromClients(clinfo.getKey()); + } + } + /** * Called once our node object is available. */ @@ -161,6 +205,9 @@ public class PeerManager } }); + // register ourselves as a client observer + PresentsServer.clmgr.addClientObserver(this); + // and start our peer refresh interval (this need not use a runqueue as // all it will do is post an invoker unit) new Interval() { @@ -226,6 +273,23 @@ public class PeerManager return NodeObject.class; } + /** + * Creates a {@link ClientInfo} record which will subsequently be + * initialized by a call to {@link #initClientInfo}. + */ + protected ClientInfo createClientInfo () + { + return new ClientInfo(); + } + + /** + * Initializes the supplied client info for the supplied client. + */ + protected void initClientInfo (PresentsClient client, ClientInfo info) + { + info.username = client.getCredentials().getUsername(); + } + /** * Contains all runtime information for one of our peer nodes. */ @@ -242,6 +306,11 @@ public class PeerManager _client.addClientObserver(this); } + public Client getClient () + { + return _client; + } + public void refresh (NodeRecord record) { // if the hostname of this node changed, kill our existing client diff --git a/src/java/com/threerings/presents/server/ClientManager.java b/src/java/com/threerings/presents/server/ClientManager.java index 26cf52f17..519aff8d0 100644 --- a/src/java/com/threerings/presents/server/ClientManager.java +++ b/src/java/com/threerings/presents/server/ClientManager.java @@ -28,6 +28,7 @@ import java.util.HashSet; import java.util.Iterator; import com.samskivert.util.Interval; +import com.samskivert.util.ObserverList; import com.samskivert.util.StringUtil; import com.threerings.util.Name; @@ -70,6 +71,25 @@ public class ClientManager public void resolutionFailed (Exception e); } + /** + * Used by entites that wish to track when clients initiate and end + * sessions on this server. + */ + public static interface ClientObserver + { + /** + * Called when a client has authenticated and been resolved and has + * started their session. + */ + public void clientSessionDidStart (PresentsClient client); + + /** + * Called when a client has logged off or been forcibly logged off due + * to inactivity and has thus ended their session. + */ + public void clientSessionDidEnd (PresentsClient client); + } + /** * Constructs a client manager that will interact with the supplied * connection manager. @@ -161,6 +181,24 @@ public class ClientManager return _objmap.values().iterator(); } + /** + * Registers an observer that will be notified when clients start and end + * their sessions. + */ + public void addClientObserver (ClientObserver observer) + { + _clobservers.add(observer); + } + + /** + * Removes an observer previously registered with {@link + * #addClientObserver}. + */ + public void removeClientObserver (ClientObserver observer) + { + _clobservers.remove(observer); + } + /** * Returns the client instance that manages the client session for the * specified authentication username or null if that client is not @@ -372,15 +410,31 @@ public class ClientManager report.append("- Mapped users: ").append(_objmap.size()).append("\n"); } + /** + * Called by the client instance when it has started its session. This is + * called from the dobjmgr thread. + */ + protected void clientSessionDidStart (final PresentsClient client) + { + // let the observers know + _clobservers.apply(new ObserverList.ObserverOp() { + public boolean apply (ClientObserver observer) { + observer.clientSessionDidStart(client); + return true; + } + }); + } + /** * Called by the client instance when the client requests a logoff. * This is called from the conmgr thread. */ - synchronized void clientDidEndSession (PresentsClient client) + protected synchronized void clientSessionDidEnd ( + final PresentsClient client) { // remove the client from the username map - Credentials creds = client.getCredentials(); - PresentsClient rc = _usermap.remove(creds.getUsername()); + PresentsClient rc = _usermap.remove( + client.getCredentials().getUsername()); // sanity check just because we can if (rc == null) { @@ -392,6 +446,20 @@ public class ClientManager } else { Log.info("Ending session " + client + "."); } + + // queue up a unit on the dobjmgr thread to notify the client observers + // that the session is ended + PresentsServer.omgr.postRunnable(new Runnable() { + public void run () { + _clobservers.apply( + new ObserverList.ObserverOp() { + public boolean apply (ClientObserver observer) { + observer.clientSessionDidEnd(client); + return true; + } + }); + } + }); } /** @@ -484,6 +552,10 @@ public class ClientManager /** The client class in use. */ protected ClientFactory _factory = ClientFactory.DEFAULT; + /** Tracks registered {@link ClientObserver}s. */ + protected ObserverList _clobservers = + new ObserverList(ObserverList.SAFE_IN_ORDER_NOTIFY); + /** The frequency with which we check for expired clients. */ protected static final long CLIENT_FLUSH_INTERVAL = 60 * 1000L; } diff --git a/src/java/com/threerings/presents/server/PresentsClient.java b/src/java/com/threerings/presents/server/PresentsClient.java index 85b8fa5d7..ab58bcd31 100644 --- a/src/java/com/threerings/presents/server/PresentsClient.java +++ b/src/java/com/threerings/presents/server/PresentsClient.java @@ -340,6 +340,9 @@ public class PresentsClient // finish up our regular business sessionWillStart(); sendBootstrap(); + + // let the client manager know that we're operational + _cmgr.clientSessionDidStart(this); } // documentation inherited from interface @@ -449,7 +452,7 @@ public class PresentsClient } // let the client manager know that we're audi 5000 - _cmgr.clientDidEndSession(this); + _cmgr.clientSessionDidEnd(this); // clear out the client object so that we know the session is over _clobj = null;