From 628e03beed80f35d8e3bd7585bf68916cf44c708 Mon Sep 17 00:00:00 2001 From: Par Winzell Date: Tue, 7 Sep 2010 19:10:00 +0000 Subject: [PATCH] As the first application of the new invokeNodeRequest, rip all the chat collection stuff out of CrowdPeerManager and into the chat package where it belongs. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@6142 542714f4-19e9-0310-aa3c-eee0fc999fb1 --- .../crowd/chat/server/ChatChannelManager.java | 96 +++++++++++- .../crowd/peer/client/CrowdPeerService.java | 6 - .../crowd/peer/data/CrowdPeerMarshaller.java | 14 -- .../peer/server/CrowdPeerDispatcher.java | 7 - .../crowd/peer/server/CrowdPeerManager.java | 144 ------------------ .../crowd/peer/server/CrowdPeerProvider.java | 7 - 6 files changed, 92 insertions(+), 182 deletions(-) diff --git a/src/java/com/threerings/crowd/chat/server/ChatChannelManager.java b/src/java/com/threerings/crowd/chat/server/ChatChannelManager.java index 64297db4e..f7807e686 100644 --- a/src/java/com/threerings/crowd/chat/server/ChatChannelManager.java +++ b/src/java/com/threerings/crowd/chat/server/ChatChannelManager.java @@ -21,16 +21,22 @@ package com.threerings.crowd.chat.server; -import java.util.Iterator; -import java.util.List; -import java.util.Map; -import java.util.Set; +import java.util.*; +import com.google.common.base.Predicate; +import com.google.common.collect.Iterables; import com.google.common.collect.Lists; import com.google.common.collect.Maps; +import com.google.common.primitives.Longs; import com.google.inject.Inject; import com.samskivert.util.ArrayIntSet; +import com.samskivert.util.ResultListener; +import com.threerings.crowd.chat.server.SpeakUtil.ChatHistoryEntry; +import com.threerings.presents.client.InvocationService; +import com.threerings.presents.peer.server.PeerManager.NodeRequest; +import com.threerings.presents.peer.server.NodeRequestsListener; +import com.threerings.presents.server.InvocationException; import com.threerings.util.Name; import com.threerings.presents.annotation.AnyThread; @@ -58,6 +64,19 @@ import static com.threerings.crowd.Log.log; public abstract class ChatChannelManager implements ChannelSpeakProvider { + /** + * Value asynchronously returned by {@link #collectChatHistory} after polling all peer nodes. + */ + public static class ChatHistoryResult + { + /** The set of nodes that either did not reply within the timeout, or had a failure. */ + public Set failedNodes; + + /** The things in the user's chat history, aggregated from all nodes and sorted by + * timestamp. */ + public List history; + } + /** * 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 @@ -80,6 +99,34 @@ public abstract class ChatChannelManager _peerMan.invokeNodeAction(new ParticipantChanged(channel, bodyId, false)); } + /** + * Collects all chat messages heard by the given user on all peers. + */ + @AnyThread + public void collectChatHistory (Name user, final ResultListener lner) + { + _peerMan.invokeNodeRequest ( + new ChatCollectionRequest(user), new NodeRequestsListener>() + { + @Override public void requestsProcessed ( + NodeRequestsResult> requestResult) + { + ChatHistoryResult result = new ChatHistoryResult(); + result.failedNodes = requestResult.getNodeErrors().keySet(); + result.history = Lists.newArrayList(); + for (Map.Entry> entry : requestResult.getNodeResults().entrySet()) { + result.history.addAll(entry.getValue()); + } + lner.requestCompleted(result); + } + + @Override public void requestFailed (String cause) + { + lner.requestFailed(new InvocationException(cause)); + } + }); + } + // from interface ChannelSpeakProvider public void speak (ClientObject caller, final ChatChannel channel, String message, byte mode) { @@ -348,6 +395,47 @@ public abstract class ChatChannelManager protected boolean _added; } + protected static class ChatCollectionRequest extends NodeRequest + { + public ChatCollectionRequest (Name user) + { + _user = user; + } + + public ChatCollectionRequest () + { + } + + @Override public boolean isApplicable (NodeObject nodeobj) + { + // poll all nodes + return true; + } + + @Override protected void execute (InvocationService.ResultListener listener) + { + // find all the UserMessages for the given user and send them back + listener.requestProcessed(Lists.newArrayList(Iterables.filter( + SpeakUtil.getChatHistory(_user), IS_USER_MESSAGE))); + } + + protected Name _user; + } + + protected static final Predicate IS_USER_MESSAGE = + new Predicate() { + public boolean apply (ChatHistoryEntry entry) { + return entry.message instanceof UserMessage; + } + }; + + protected static final Comparator SORT_BY_TIMESTAMP = + new Comparator() { + public int compare (ChatHistoryEntry e1, ChatHistoryEntry e2) { + return Longs.compare(e1.message.timestamp, e2.message.timestamp); + } + }; + /** 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 diff --git a/src/java/com/threerings/crowd/peer/client/CrowdPeerService.java b/src/java/com/threerings/crowd/peer/client/CrowdPeerService.java index d657721d8..601ae6357 100644 --- a/src/java/com/threerings/crowd/peer/client/CrowdPeerService.java +++ b/src/java/com/threerings/crowd/peer/client/CrowdPeerService.java @@ -46,10 +46,4 @@ public interface CrowdPeerService extends InvocationService * Dispatches a broadcast message on this peer. */ void deliverBroadcast (Client client, Name from, byte levelOrMode, String bundle, String msg); - - /** - * Obtains the messages that the user has heard on this peer. On success, the listener will - * receive a {@code List} of {@link SpeakUtil.ChatHistoryEntry} instances. - */ - void getChatHistory (Client caller, Name user, ResultListener lner); } diff --git a/src/java/com/threerings/crowd/peer/data/CrowdPeerMarshaller.java b/src/java/com/threerings/crowd/peer/data/CrowdPeerMarshaller.java index d346f56b8..8d9a9badf 100644 --- a/src/java/com/threerings/crowd/peer/data/CrowdPeerMarshaller.java +++ b/src/java/com/threerings/crowd/peer/data/CrowdPeerMarshaller.java @@ -28,7 +28,6 @@ import com.threerings.crowd.chat.data.ChatMarshaller; import com.threerings.crowd.chat.data.UserMessage; import com.threerings.crowd.peer.client.CrowdPeerService; import com.threerings.presents.client.Client; -import com.threerings.presents.client.InvocationService; import com.threerings.presents.data.InvocationMarshaller; import com.threerings.util.Name; @@ -67,17 +66,4 @@ public class CrowdPeerMarshaller extends InvocationMarshaller arg2, arg3, listener4 }); } - - /** The method id used to dispatch {@link #getChatHistory} requests. */ - public static final int GET_CHAT_HISTORY = 3; - - // from interface CrowdPeerService - public void getChatHistory (Client arg1, Name arg2, InvocationService.ResultListener arg3) - { - InvocationMarshaller.ResultMarshaller listener3 = new InvocationMarshaller.ResultMarshaller(); - listener3.listener = arg3; - sendRequest(arg1, GET_CHAT_HISTORY, new Object[] { - arg2, listener3 - }); - } } diff --git a/src/java/com/threerings/crowd/peer/server/CrowdPeerDispatcher.java b/src/java/com/threerings/crowd/peer/server/CrowdPeerDispatcher.java index 961ea0ffd..bba201ddc 100644 --- a/src/java/com/threerings/crowd/peer/server/CrowdPeerDispatcher.java +++ b/src/java/com/threerings/crowd/peer/server/CrowdPeerDispatcher.java @@ -26,7 +26,6 @@ import javax.annotation.Generated; import com.threerings.crowd.chat.client.ChatService; import com.threerings.crowd.chat.data.UserMessage; import com.threerings.crowd.peer.data.CrowdPeerMarshaller; -import com.threerings.presents.client.InvocationService; import com.threerings.presents.data.ClientObject; import com.threerings.presents.server.InvocationDispatcher; import com.threerings.presents.server.InvocationException; @@ -72,12 +71,6 @@ public class CrowdPeerDispatcher extends InvocationDispatcher failedNodes; - - /** The things in the user's chat history, aggregated from all nodes and sorted by - * timestamp. */ - public List history; - } - /** * Creates an uninitialized peer manager. */ @@ -102,15 +89,6 @@ public abstract class CrowdPeerManager extends PeerManager _chatprov.broadcast(from, levelOrMode, bundle, msg, false); } - // from interface CrowdPeerProvider - public void getChatHistory ( - ClientObject caller, Name user, InvocationService.ResultListener lner) - throws InvocationException - { - lner.requestProcessed(Lists.newArrayList(Iterables.filter( - SpeakUtil.getChatHistory(user), IS_USER_MESSAGE))); - } - // from interface ChatProvider.ChatForwarder public boolean forwardTell (UserMessage message, Name target, ChatService.TellListener listener) @@ -148,16 +126,6 @@ public abstract class CrowdPeerManager extends PeerManager } } - /** - * Collects all chat messages heard by the given user on all peers. Must be called on the - * dobj event thread. - */ - public void collectChatHistory (Name user, ResultListener lner) - { - _omgr.requireEventThread(); - new ChatHistoryCollector(user, lner).collect(); - } - @Override // from PeerManager public void shutdown () { @@ -211,117 +179,5 @@ public abstract class CrowdPeerManager extends PeerManager */ protected abstract Name authFromViz (Name vizname); - /** - * Asynchronously collects the chat history from all nodes for a given user. - * TODO: refactor node parts into base class similar to PeerManager.NodeAction - */ - protected class ChatHistoryCollector - { - public ChatHistoryCollector (Name user, ResultListener listener) - { - _user = user; - _listener = listener; - - _result = new ChatHistoryResult(); - _result.failedNodes = Sets.newHashSet(); - _waiting = Sets.newHashSet(); - } - - public void collect () - { - _result.history = Lists.newArrayList( - Iterables.filter(SpeakUtil.getChatHistory(_user), IS_USER_MESSAGE)); - - for (PeerNode peer : _peers.values()) { - final PeerNode fpeer = peer; - ((CrowdNodeObject)peer.nodeobj).crowdPeerService.getChatHistory( - peer.getClient(), _user, new InvocationService.ResultListener() { - public void requestProcessed (Object result) { - processed(fpeer, result); - } - public void requestFailed (String cause) { - failed(fpeer, cause); - } - }); - _waiting.add(peer.getNodeName()); - } - - // timeout in 5 seconds if we haven't heard back from all nodes - final long TIMEOUT = 5 * 1000; - if (!maybeComplete()) { - new Interval(_omgr) { - @Override public void expired () { - checkTimeout(); - } - }.schedule(TIMEOUT); - } - } - - protected void processed (PeerNode node, Object result) - { - String name = node.getNodeName(); - if (!_waiting.remove(name)) { - log.warning("Double chat history response from node", "name", name); - return; - } - - @SuppressWarnings("unchecked") - List nodeMessages = (List)result; - _result.history.addAll(nodeMessages); - maybeComplete(); - } - - protected void failed (PeerNode node, String cause) - { - String name = node.getNodeName(); - _result.failedNodes.add(name); - - if (_waiting.remove(name)) { - maybeComplete(); - } else { - log.warning("Double chat history response from node", "name", name); - } - } - - protected boolean maybeComplete () - { - if (!_waiting.isEmpty()) { - return false; - } - - Collections.sort(_result.history, SORT_BY_TIMESTAMP); - _listener.requestCompleted(_result); - return true; - } - - protected void checkTimeout () - { - if (!_waiting.isEmpty()) { - _result.failedNodes.addAll(_waiting); - _waiting.clear(); - maybeComplete(); - } - } - - protected Name _user; - protected ResultListener _listener; - protected ChatHistoryResult _result; - protected Set _waiting; - } - @Inject protected ChatProvider _chatprov; - - protected static final Predicate IS_USER_MESSAGE = - new Predicate() { - public boolean apply (ChatHistoryEntry entry) { - return entry.message instanceof UserMessage; - } - }; - - protected static final Comparator SORT_BY_TIMESTAMP = - new Comparator() { - public int compare (ChatHistoryEntry e1, ChatHistoryEntry e2) { - return Longs.compare(e1.message.timestamp, e2.message.timestamp); - } - }; } diff --git a/src/java/com/threerings/crowd/peer/server/CrowdPeerProvider.java b/src/java/com/threerings/crowd/peer/server/CrowdPeerProvider.java index 824f9144e..6ced07136 100644 --- a/src/java/com/threerings/crowd/peer/server/CrowdPeerProvider.java +++ b/src/java/com/threerings/crowd/peer/server/CrowdPeerProvider.java @@ -26,7 +26,6 @@ import javax.annotation.Generated; import com.threerings.crowd.chat.client.ChatService; import com.threerings.crowd.chat.data.UserMessage; import com.threerings.crowd.peer.client.CrowdPeerService; -import com.threerings.presents.client.InvocationService; import com.threerings.presents.data.ClientObject; import com.threerings.presents.server.InvocationException; import com.threerings.presents.server.InvocationProvider; @@ -49,10 +48,4 @@ public interface CrowdPeerProvider extends InvocationProvider */ void deliverTell (ClientObject caller, UserMessage arg1, Name arg2, ChatService.TellListener arg3) throws InvocationException; - - /** - * Handles a {@link CrowdPeerService#getChatHistory} request. - */ - void getChatHistory (ClientObject caller, Name arg1, InvocationService.ResultListener arg2) - throws InvocationException; }