From 421272f4f2c899b6005e2ef6792c6d467c157d94 Mon Sep 17 00:00:00 2001 From: Michael Bayne Date: Sun, 10 Aug 2008 02:40:19 +0000 Subject: [PATCH] Basic framework for distributed chat channel system that does not use chat channel distributed objects but rather resolves the location of all channel participants on every message delivery so that we don't have to attempt the very fragile process of having every server that hosts a channel participant maintain a subscription to the chat channel object on another peer. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@5305 542714f4-19e9-0310-aa3c-eee0fc999fb1 --- .../crowd/chat/client/ChannelSpeakService.as | 36 ++ .../crowd/chat/data/ChannelSpeakMarshaller.as | 50 +++ .../threerings/crowd/chat/data/ChatChannel.as | 54 +++ .../threerings/crowd/chat/data/ChatCodes.as | 5 +- .../chat/client/ChannelSpeakService.java | 38 ++ .../chat/data/ChannelSpeakMarshaller.java | 48 +++ .../crowd/chat/data/ChatChannel.java | 57 +++ .../threerings/crowd/chat/data/ChatCodes.java | 5 +- .../crowd/chat/data/ChatMarshaller.java | 2 +- .../chat/server/ChannelSpeakDispatcher.java | 67 ++++ .../chat/server/ChannelSpeakProvider.java | 38 ++ .../crowd/chat/server/ChatChannelManager.java | 357 ++++++++++++++++++ .../crowd/chat/server/ChatProvider.java | 2 +- .../crowd/data/LocationMarshaller.java | 6 +- .../crowd/peer/data/CrowdNodeObject.java | 56 +++ 15 files changed, 813 insertions(+), 8 deletions(-) create mode 100644 src/as/com/threerings/crowd/chat/client/ChannelSpeakService.as create mode 100644 src/as/com/threerings/crowd/chat/data/ChannelSpeakMarshaller.as create mode 100644 src/as/com/threerings/crowd/chat/data/ChatChannel.as create mode 100644 src/java/com/threerings/crowd/chat/client/ChannelSpeakService.java create mode 100644 src/java/com/threerings/crowd/chat/data/ChannelSpeakMarshaller.java create mode 100644 src/java/com/threerings/crowd/chat/data/ChatChannel.java create mode 100644 src/java/com/threerings/crowd/chat/server/ChannelSpeakDispatcher.java create mode 100644 src/java/com/threerings/crowd/chat/server/ChannelSpeakProvider.java create mode 100644 src/java/com/threerings/crowd/chat/server/ChatChannelManager.java diff --git a/src/as/com/threerings/crowd/chat/client/ChannelSpeakService.as b/src/as/com/threerings/crowd/chat/client/ChannelSpeakService.as new file mode 100644 index 000000000..9b4bba952 --- /dev/null +++ b/src/as/com/threerings/crowd/chat/client/ChannelSpeakService.as @@ -0,0 +1,36 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved +// http://www.threerings.net/code/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.crowd.chat.client { + +import com.threerings.crowd.chat.data.ChatChannel; +import com.threerings.presents.client.Client; +import com.threerings.presents.client.InvocationService; + +/** + * An ActionScript version of the Java ChannelSpeakService interface. + */ +public interface ChannelSpeakService extends InvocationService +{ + // from Java interface ChannelSpeakService + function speak (arg1 :Client, arg2 :ChatChannel, arg3 :String, arg4 :int) :void; +} +} diff --git a/src/as/com/threerings/crowd/chat/data/ChannelSpeakMarshaller.as b/src/as/com/threerings/crowd/chat/data/ChannelSpeakMarshaller.as new file mode 100644 index 000000000..42ee12dc4 --- /dev/null +++ b/src/as/com/threerings/crowd/chat/data/ChannelSpeakMarshaller.as @@ -0,0 +1,50 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved +// http://www.threerings.net/code/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.crowd.chat.data { + +import com.threerings.crowd.chat.client.ChannelSpeakService; +import com.threerings.presents.client.Client; +import com.threerings.presents.data.InvocationMarshaller; +import com.threerings.util.Byte; + +/** + * Provides the implementation of the ChannelSpeakService interface + * that marshalls the arguments and delivers the request to the provider + * on the server. Also provides an implementation of the response listener + * interfaces that marshall the response arguments and deliver them back + * to the requesting client. + */ +public class ChannelSpeakMarshaller extends InvocationMarshaller + implements ChannelSpeakService +{ + /** The method id used to dispatch speak requests. */ + public static const SPEAK :int = 1; + + // from interface ChannelSpeakService + public function speak (arg1 :Client, arg2 :ChatChannel, arg3 :String, arg4 :int) :void + { + sendRequest(arg1, SPEAK, [ + arg2, arg3, Byte.valueOf(arg4) + ]); + } +} +} diff --git a/src/as/com/threerings/crowd/chat/data/ChatChannel.as b/src/as/com/threerings/crowd/chat/data/ChatChannel.as new file mode 100644 index 000000000..1e82ecfaa --- /dev/null +++ b/src/as/com/threerings/crowd/chat/data/ChatChannel.as @@ -0,0 +1,54 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved +// http://www.threerings.net/code/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.crowd.chat.data { + +import com.threerings.io.SimpleStreamableObject; +import com.threerings.util.Comparable; +import com.threerings.util.Equalable; + +import com.threerings.presents.dobj.DSet_Entry; + +/** + * Represents a chat channel. + */ +public /*abstract*/ class ChatChannel extends SimpleStreamableObject + implements Comparable, Equalable, DSet_Entry +{ + // from interface Comparable + public function compareTo (other :Object) :int + { + throw new Error("abstract"); + } + + // from interface Equalable + public function equals (other :Object) :Boolean + { + throw new Error("abstract"); + } + + // from interface DSet_Entry + public function getKey () :Object + { + return this; + } +} +} diff --git a/src/as/com/threerings/crowd/chat/data/ChatCodes.as b/src/as/com/threerings/crowd/chat/data/ChatCodes.as index 6c40b0d8b..c0b70ae05 100644 --- a/src/as/com/threerings/crowd/chat/data/ChatCodes.as +++ b/src/as/com/threerings/crowd/chat/data/ChatCodes.as @@ -36,7 +36,10 @@ public class ChatCodes extends InvocationCodes public static const SUCCESS :String = "success"; /** The message identifier for a chat notification message. */ - public static const CHAT_NOTIFICATION :String = "chat"; + public static const CHAT_NOTIFICATION :String = "crowd.chat"; + + /** The message identifier for a chat channel notification message. */ + public static const CHAT_CHANENL_NOTIFICATION :String = "crowd.chat.channel"; /** The access control identifier for normal chat privileges. */ public static const CHAT_ACCESS :Permission = new Permission(); diff --git a/src/java/com/threerings/crowd/chat/client/ChannelSpeakService.java b/src/java/com/threerings/crowd/chat/client/ChannelSpeakService.java new file mode 100644 index 000000000..6da50ba0f --- /dev/null +++ b/src/java/com/threerings/crowd/chat/client/ChannelSpeakService.java @@ -0,0 +1,38 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved +// http://www.threerings.net/code/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.crowd.chat.client; + +import com.threerings.presents.client.Client; +import com.threerings.presents.client.InvocationService; + +import com.threerings.crowd.chat.data.ChatChannel; + +/** + * Provides a way for clients to speak on chat channels. + */ +public interface ChannelSpeakService extends InvocationService +{ + /** + * Requests to speak the supplied message on the specified channel. + */ + public void speak (Client client, ChatChannel channel, String message, byte mode); +} diff --git a/src/java/com/threerings/crowd/chat/data/ChannelSpeakMarshaller.java b/src/java/com/threerings/crowd/chat/data/ChannelSpeakMarshaller.java new file mode 100644 index 000000000..4e596aa6a --- /dev/null +++ b/src/java/com/threerings/crowd/chat/data/ChannelSpeakMarshaller.java @@ -0,0 +1,48 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved +// http://www.threerings.net/code/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.crowd.chat.data; + +import com.threerings.crowd.chat.client.ChannelSpeakService; +import com.threerings.presents.client.Client; +import com.threerings.presents.data.InvocationMarshaller; + +/** + * Provides the implementation of the {@link ChannelSpeakService} interface + * that marshalls the arguments and delivers the request to the provider + * on the server. Also provides an implementation of the response listener + * interfaces that marshall the response arguments and deliver them back + * to the requesting client. + */ +public class ChannelSpeakMarshaller extends InvocationMarshaller + implements ChannelSpeakService +{ + /** The method id used to dispatch {@link #speak} requests. */ + public static final int SPEAK = 1; + + // from interface ChannelSpeakService + public void speak (Client arg1, ChatChannel arg2, String arg3, byte arg4) + { + sendRequest(arg1, SPEAK, new Object[] { + arg2, arg3, Byte.valueOf(arg4) + }); + } +} diff --git a/src/java/com/threerings/crowd/chat/data/ChatChannel.java b/src/java/com/threerings/crowd/chat/data/ChatChannel.java new file mode 100644 index 000000000..ce072f5ca --- /dev/null +++ b/src/java/com/threerings/crowd/chat/data/ChatChannel.java @@ -0,0 +1,57 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved +// http://www.threerings.net/code/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.crowd.chat.data; + +import com.threerings.io.SimpleStreamableObject; + +import com.threerings.presents.dobj.DSet; + +/** + * Represents a chat channel. + */ +public abstract class ChatChannel extends SimpleStreamableObject + implements Comparable, DSet.Entry +{ + // from interface Comparable + public abstract int compareTo (ChatChannel other); + + /** + * Converts this channel into a unique name that can be used as the name of the distributed + * lock used when resolving the channel. + */ + public abstract String getLockName (); + + // from interface DSet.Entry + public Comparable getKey () + { + return this; + } + + @Override + public boolean equals (Object other) + { + return compareTo((ChatChannel)other) == 0; + } + + @Override + public abstract int hashCode (); +} diff --git a/src/java/com/threerings/crowd/chat/data/ChatCodes.java b/src/java/com/threerings/crowd/chat/data/ChatCodes.java index 48a170681..2f021285b 100644 --- a/src/java/com/threerings/crowd/chat/data/ChatCodes.java +++ b/src/java/com/threerings/crowd/chat/data/ChatCodes.java @@ -37,7 +37,10 @@ public interface ChatCodes extends InvocationCodes public static final String SUCCESS = "success"; /** The message identifier for a chat notification message. */ - public static final String CHAT_NOTIFICATION = "chat"; + public static final String CHAT_NOTIFICATION = "crowd.chat"; + + /** The message identifier for a chat channel notification message. */ + public static final String CHAT_CHANNEL_NOTIFICATION = "crowd.chat.channel"; /** The access control identifier for normal chat privileges. */ public static final Permission CHAT_ACCESS = new Permission(); diff --git a/src/java/com/threerings/crowd/chat/data/ChatMarshaller.java b/src/java/com/threerings/crowd/chat/data/ChatMarshaller.java index de4c7036a..949283a1d 100644 --- a/src/java/com/threerings/crowd/chat/data/ChatMarshaller.java +++ b/src/java/com/threerings/crowd/chat/data/ChatMarshaller.java @@ -39,7 +39,7 @@ public class ChatMarshaller extends InvocationMarshaller implements ChatService { /** - * Marshalls results to implementations of {@link TellListener}. + * Marshalls results to implementations of {@link ChatService.TellListener}. */ public static class TellMarshaller extends ListenerMarshaller implements TellListener diff --git a/src/java/com/threerings/crowd/chat/server/ChannelSpeakDispatcher.java b/src/java/com/threerings/crowd/chat/server/ChannelSpeakDispatcher.java new file mode 100644 index 000000000..c3487deb9 --- /dev/null +++ b/src/java/com/threerings/crowd/chat/server/ChannelSpeakDispatcher.java @@ -0,0 +1,67 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved +// http://www.threerings.net/code/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.crowd.chat.server; + +import com.threerings.crowd.chat.data.ChannelSpeakMarshaller; +import com.threerings.crowd.chat.data.ChatChannel; +import com.threerings.presents.data.ClientObject; +import com.threerings.presents.server.InvocationDispatcher; +import com.threerings.presents.server.InvocationException; + +/** + * Dispatches requests to the {@link ChannelSpeakProvider}. + */ +public class ChannelSpeakDispatcher extends InvocationDispatcher +{ + /** + * Creates a dispatcher that may be registered to dispatch invocation + * service requests for the specified provider. + */ + public ChannelSpeakDispatcher (ChannelSpeakProvider provider) + { + this.provider = provider; + } + + @Override // documentation inherited + public ChannelSpeakMarshaller createMarshaller () + { + return new ChannelSpeakMarshaller(); + } + + @Override // documentation inherited + public void dispatchRequest ( + ClientObject source, int methodId, Object[] args) + throws InvocationException + { + switch (methodId) { + case ChannelSpeakMarshaller.SPEAK: + ((ChannelSpeakProvider)provider).speak( + source, (ChatChannel)args[0], (String)args[1], ((Byte)args[2]).byteValue() + ); + return; + + default: + super.dispatchRequest(source, methodId, args); + return; + } + } +} diff --git a/src/java/com/threerings/crowd/chat/server/ChannelSpeakProvider.java b/src/java/com/threerings/crowd/chat/server/ChannelSpeakProvider.java new file mode 100644 index 000000000..59bea767d --- /dev/null +++ b/src/java/com/threerings/crowd/chat/server/ChannelSpeakProvider.java @@ -0,0 +1,38 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved +// http://www.threerings.net/code/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.crowd.chat.server; + +import com.threerings.crowd.chat.client.ChannelSpeakService; +import com.threerings.crowd.chat.data.ChatChannel; +import com.threerings.presents.data.ClientObject; +import com.threerings.presents.server.InvocationProvider; + +/** + * Defines the server-side of the {@link ChannelSpeakService}. + */ +public interface ChannelSpeakProvider extends InvocationProvider +{ + /** + * Handles a {@link ChannelSpeakService#speak} request. + */ + void speak (ClientObject caller, ChatChannel arg1, String arg2, byte arg3); +} diff --git a/src/java/com/threerings/crowd/chat/server/ChatChannelManager.java b/src/java/com/threerings/crowd/chat/server/ChatChannelManager.java new file mode 100644 index 000000000..06841a81c --- /dev/null +++ b/src/java/com/threerings/crowd/chat/server/ChatChannelManager.java @@ -0,0 +1,357 @@ +// +// $Id$ +// +// Narya library - tools for developing networked games +// Copyright (C) 2002-2008 Three Rings Design, Inc., All Rights Reserved +// http://www.threerings.net/code/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.crowd.chat.server; + +import java.util.List; +import java.util.Map; + +import com.google.common.base.Function; +import com.google.common.collect.Lists; +import com.google.common.collect.Maps; +import com.google.inject.Inject; +import com.google.inject.Singleton; + +import com.samskivert.io.PersistenceException; +import com.samskivert.jdbc.RepositoryUnit; +import com.samskivert.util.ArrayIntSet; +import com.samskivert.util.IntSet; +import com.threerings.util.Name; + +import com.threerings.presents.annotation.AnyThread; +import com.threerings.presents.data.ClientObject; +import com.threerings.presents.peer.data.ClientInfo; +import com.threerings.presents.peer.data.NodeObject; +import com.threerings.presents.peer.server.PeerManager; +import com.threerings.presents.server.InvocationManager; +import com.threerings.presents.server.PresentsInvoker; + +import com.threerings.crowd.data.BodyObject; +import com.threerings.crowd.data.CrowdCodes; +import com.threerings.crowd.peer.data.CrowdClientInfo; +import com.threerings.crowd.peer.data.CrowdNodeObject; +import com.threerings.crowd.peer.server.CrowdPeerManager; + +import com.threerings.crowd.chat.data.ChatChannel; +import com.threerings.crowd.chat.data.ChatCodes; +import com.threerings.crowd.chat.data.UserMessage; + +import static com.threerings.crowd.Log.log; + +/** + * Handles chat channel services. + */ +@Singleton +public abstract class ChatChannelManager + implements ChannelSpeakProvider +{ + /** + * When a body becomes a member of a channel, this method should be called so that any server + * that happens to be hosting that channel can be told that the body in question is now a + * participant. + */ + @AnyThread + public void bodyAddedToChannel (ChatChannel channel, int bodyId) + { + _peerMan.invokeNodeAction(new ParticipantChanged(channel, bodyId, true)); + } + + /** + * When a body loses channel membership, this method should be called so that any server that + * happens to be hosting that channel can be told that the body in question is now a + * participant. + */ + @AnyThread + public void bodyRemovedFromChannel (ChatChannel channel, int bodyId) + { + _peerMan.invokeNodeAction(new ParticipantChanged(channel, bodyId, false)); + } + + // from interface ChannelSpeakProvider + public void speak (ClientObject caller, final ChatChannel channel, String message, byte mode) + { + final UserMessage umsg = new UserMessage( + ((BodyObject)caller).getVisibleName(), null, message, mode); + + // if we're hosting this channel, dispatch it directly + if (_channels.containsKey(channel)) { + dispatchSpeak(channel, umsg); + return; + } + + // if we're resolving this channel, queue up our message for momentary deliver + List msgs = _resolving.get(channel); + if (msgs != null) { + msgs.add(umsg); + return; + } + + // forward the speak request to the server that hosts the channel in question + _peerMan.invokeNodeAction(new ForwardChannelSpeak(channel, umsg), new Runnable() { + public void run () { + _resolving.put(channel, Lists.newArrayList(umsg)); + resolveAndDispatch(channel); + } + }); + } + + /** + * Creates our singleton manager and registers our invocation service. + */ + @Inject protected ChatChannelManager (InvocationManager invmgr) + { + invmgr.registerDispatcher(new ChannelSpeakDispatcher(this), CrowdCodes.CROWD_GROUP); + } + + /** + * Resolves the channel specified in the supplied action and then dispatches it. + */ + protected void resolveAndDispatch (final ChatChannel channel) + { + NodeObject.Lock lock = new NodeObject.Lock("ChatChannel", channel.getLockName()); + _peerMan.performWithLock(lock, new PeerManager.LockedOperation() { + public void run () { + ((CrowdNodeObject)_peerMan.getNodeObject()).addToHostedChannels(channel); + finishResolveAndDispatch(channel); + } + public void fail (String peerName) { + List msgs = _resolving.remove(channel); + if (peerName == null) { + log.warning("Failed to resolve chat channel due to lock failure", + "channel", channel); + } else { + // some other peer resolved this channel first, so forward any queued messages + // directly to that node + for (UserMessage msg : msgs) { + _peerMan.invokeNodeAction(peerName, new ForwardChannelSpeak(channel, msg)); + } + } + } + }); + } + + /** + * Resolves the participant set for the specified chat channel and dispatches all pending + * messages to the channel. End users of the chat channel system should override this method + * and do what is necessary to resolve the channel's participant set and call {@link + * #resolutionComplete} or {@link #resolutionFailed}. + */ + protected void finishResolveAndDispatch (ChatChannel channel) + { + resolutionComplete(channel, new ArrayIntSet()); + } + + /** + * This should be called when a channel's participant set has been resolved. + */ + protected void resolutionComplete (ChatChannel channel, IntSet parts) + { + // map the participants of our now resolved channel + _channels.put(channel, parts); + + // dispatch any pending messages now that we know where they go + for (UserMessage msg : _resolving.remove(channel)) { + dispatchSpeak(channel, msg); + } + } + + /** + * This should be called if channel resolution fails. + */ + protected void resolutionFailed (ChatChannel channel, Exception cause) + { + log.warning("Failed to resolve chat channel", "channel", channel, cause); + + // alas, we just drop all pending messages because we're hosed + _resolving.remove(channel); + } + + /** + * Requests that we dispatch the supplied message to all participants of the specified chat + * channel. The speaker will be validated prior to dispatching the message as the originating + * server does not have the information it needs to validate the speaker and must leave that to + * us, the channel hosting server. + */ + protected void dispatchSpeak (ChatChannel channel, UserMessage message) + { + final IntSet parts = _channels.get(channel); + if (parts == null) { + // TODO: maybe we should just reresolve the channel... + log.warning("Requested to dispatch speak on unhosted channel", "channel", channel, + "msg", message); + return; + } + + // validate the speaker + if (!parts.contains(getBodyId(message.speaker))) { + log.warning("Dropping channel chat message from non-speaker", "channel", channel, + "message", message); + return; + } + + // generate a mapping from node name to an array of body ids for the participants that are + // currently on the node in question + final Map partMap = Maps.newHashMap(); + _peerMan.applyToNodes(new Function() { + public Void apply (NodeObject nodeobj) { + ArrayIntSet nodeBodyIds = new ArrayIntSet(); + for (ClientInfo info : nodeobj.clients) { + int bodyId = getBodyId(((CrowdClientInfo)info).visibleName); + if (parts.contains(bodyId)) { + nodeBodyIds.add(bodyId); + } + } + partMap.put(nodeobj.nodeName, nodeBodyIds.toIntArray()); + return null; + } + }); + + for (Map.Entry entry : partMap.entrySet()) { + _peerMan.invokeNodeAction( + entry.getKey(), new DispatchChannelSpeak(channel, message, entry.getValue())); + } + } + + /** + * Delivers the supplied chat channel message to the specified bodies. + */ + protected void deliverSpeak (ChatChannel channel, UserMessage message, int[] bodyIds) + { + for (int bodyId : bodyIds) { + BodyObject bobj = getBodyObject(bodyId); + if (bobj != null && shouldDeliverSpeak(channel, message, bobj)) { + bobj.postMessage(ChatCodes.CHAT_CHANNEL_NOTIFICATION, channel, message); + } + } + } + + /** + * Ratifies the delivery of the supplied chat channel message to the specified body. Derived + * classes can override this method to implement channel disabling, mute lists or any other + * suppression they might need. + */ + protected boolean shouldDeliverSpeak (ChatChannel channel, UserMessage message, BodyObject body) + { + return true; + } + + /** + * Converts a speaker's visible name into a unique integer id. This is not the oid for this + * speaker but rather a persistent integer identifier that can be passed between servers and + * used to look up the speaker on the target server via a call to {@link #getBodyObject}. We + * use this rather than names to avoid having to send (large) {@link Name} objects for every + * channel participant to each individual peer that will be forwarding messages. + */ + protected abstract int getBodyId (Name speaker); + + /** + * Locates a body object from the given unique id. May return null. + */ + protected abstract BodyObject getBodyObject (int bodyId); + + /** Forwards a channel speak request from the server hosting the message originator to the + * server that is hosting the channel. */ + protected abstract static class ChannelAction extends PeerManager.NodeAction + { + public ChannelAction (ChatChannel channel) { + _channel = channel; + } + public ChannelAction () { + } + @Override public boolean isApplicable (NodeObject nodeobj) { + return ((CrowdNodeObject)nodeobj).hostedChannels.contains(_channel); + } + protected ChatChannel _channel; + @Inject protected ChatChannelManager _channelMan; + } + + /** Informs the server hosting a channel that a body has been added to or removed from the + * channel's participants set. */ + protected static class ParticipantChanged extends ChannelAction + { + public ParticipantChanged (ChatChannel channel, int bodyId, boolean added) { + super(channel); + _bodyId = bodyId; + _added = added; + } + public ParticipantChanged () { + } + @Override protected void execute () { + IntSet partSet = _channelMan._channels.get(_channel); + if (partSet != null) { + if (_added) { + partSet.add(_bodyId); + } else { + partSet.remove(_bodyId); + } + } else if (_channelMan._resolving.containsKey(_channel)) { + log.warning("Oh for fuck's sake, distributed systems are complicated", + "channel", _channel); + } + } + protected int _bodyId; + protected boolean _added; + } + + /** Forwards a channel speak request from the server hosting the message originator to the + * server that is hosting the channel. */ + protected static class ForwardChannelSpeak extends ChannelAction + { + public ForwardChannelSpeak (ChatChannel channel, UserMessage message) { + super(channel); + _message = message; + } + public ForwardChannelSpeak () { + } + @Override protected void execute () { + _channelMan.dispatchSpeak(_channel, _message); + } + protected UserMessage _message; + } + + /** Forwards a chat channel message to the server to which some subset of the channel + * participants are connected so that it can dispatch the message on their body objects. */ + protected static class DispatchChannelSpeak extends ForwardChannelSpeak + { + public DispatchChannelSpeak (ChatChannel channel, UserMessage message, int[] bodyIds) { + super(channel, message); + _bodyIds = bodyIds; + } + public DispatchChannelSpeak () { + } + @Override public boolean isApplicable (NodeObject nodeobj) { + return true; // not used + } + @Override protected void execute () { + _channelMan.deliverSpeak(_channel, _message, _bodyIds); + } + protected int[] _bodyIds; + } + + /** Contains pending messages for all channels currently being resolved. */ + protected Map> _resolving = Maps.newHashMap(); + + /** A map of resolved channels to the body ids of their participants. */ + protected Map _channels = Maps.newHashMap(); + + @Inject protected CrowdPeerManager _peerMan; + @Inject protected PresentsInvoker _invoker; +} diff --git a/src/java/com/threerings/crowd/chat/server/ChatProvider.java b/src/java/com/threerings/crowd/chat/server/ChatProvider.java index 842fbe24b..776207f41 100644 --- a/src/java/com/threerings/crowd/chat/server/ChatProvider.java +++ b/src/java/com/threerings/crowd/chat/server/ChatProvider.java @@ -126,7 +126,7 @@ public class ChatProvider } /** - * Processes a request from a client to deliver a tell message to another client. + * Processes a {@link ChatService#tell} request. */ public void tell (ClientObject caller, Name target, String message, TellListener listener) throws InvocationException diff --git a/src/java/com/threerings/crowd/data/LocationMarshaller.java b/src/java/com/threerings/crowd/data/LocationMarshaller.java index 114bdb1b5..7f4f5d274 100644 --- a/src/java/com/threerings/crowd/data/LocationMarshaller.java +++ b/src/java/com/threerings/crowd/data/LocationMarshaller.java @@ -37,7 +37,7 @@ public class LocationMarshaller extends InvocationMarshaller implements LocationService { /** - * Marshalls results to implementations of {@link MoveListener}. + * Marshalls results to implementations of {@link LocationService.MoveListener}. */ public static class MoveMarshaller extends ListenerMarshaller implements MoveListener @@ -77,9 +77,7 @@ public class LocationMarshaller extends InvocationMarshaller // from interface LocationService public void leavePlace (Client arg1) { - sendRequest(arg1, LEAVE_PLACE, new Object[] { - - }); + sendRequest(arg1, LEAVE_PLACE, new Object[] {}); } /** The method id used to dispatch {@link #moveTo} requests. */ diff --git a/src/java/com/threerings/crowd/peer/data/CrowdNodeObject.java b/src/java/com/threerings/crowd/peer/data/CrowdNodeObject.java index b18229cad..2b43ab6a5 100644 --- a/src/java/com/threerings/crowd/peer/data/CrowdNodeObject.java +++ b/src/java/com/threerings/crowd/peer/data/CrowdNodeObject.java @@ -21,8 +21,11 @@ package com.threerings.crowd.peer.data; +import com.threerings.presents.dobj.DSet; import com.threerings.presents.peer.data.NodeObject; +import com.threerings.crowd.chat.data.ChatChannel; + /** * Extends the basic {@link NodeObject} with Crowd bits. */ @@ -31,11 +34,17 @@ public class CrowdNodeObject extends NodeObject // AUTO-GENERATED: FIELDS START /** The field name of the crowdPeerService field. */ public static final String CROWD_PEER_SERVICE = "crowdPeerService"; + + /** The field name of the hostedChannels field. */ + public static final String HOSTED_CHANNELS = "hostedChannels"; // AUTO-GENERATED: FIELDS END /** Used to coordinate tells between servers. */ public CrowdPeerMarshaller crowdPeerService; + /** The chat channels hosted on this server. */ + public DSet hostedChannels = new DSet(); + // AUTO-GENERATED: METHODS START /** * Requests that the crowdPeerService field be set to the @@ -52,5 +61,52 @@ public class CrowdNodeObject extends NodeObject CROWD_PEER_SERVICE, value, ovalue); this.crowdPeerService = value; } + + /** + * Requests that the specified entry be added to the + * hostedChannels set. The set will not change until the event is + * actually propagated through the system. + */ + public void addToHostedChannels (ChatChannel elem) + { + requestEntryAdd(HOSTED_CHANNELS, hostedChannels, elem); + } + + /** + * Requests that the entry matching the supplied key be removed from + * the hostedChannels set. The set will not change until the + * event is actually propagated through the system. + */ + public void removeFromHostedChannels (Comparable key) + { + requestEntryRemove(HOSTED_CHANNELS, hostedChannels, key); + } + + /** + * Requests that the specified entry be updated in the + * hostedChannels set. The set will not change until the event is + * actually propagated through the system. + */ + public void updateHostedChannels (ChatChannel elem) + { + requestEntryUpdate(HOSTED_CHANNELS, hostedChannels, elem); + } + + /** + * Requests that the hostedChannels field be set to the + * specified value. Generally one only adds, updates and removes + * entries of a distributed set, but certain situations call for a + * complete replacement of the set value. The local value will be + * updated immediately and an event will be propagated through the + * system to notify all listeners that the attribute did + * change. Proxied copies of this object (on clients) will apply the + * value change when they received the attribute changed notification. + */ + public void setHostedChannels (DSet value) + { + requestAttributeChange(HOSTED_CHANNELS, value, this.hostedChannels); + DSet clone = (value == null) ? null : value.typedClone(); + this.hostedChannels = clone; + } // AUTO-GENERATED: METHODS END }