From fd7b3ae383deaefde97671058e04d31654254280 Mon Sep 17 00:00:00 2001 From: Jamie Doornbos Date: Tue, 10 Nov 2009 01:13:58 +0000 Subject: [PATCH] Method to poll all peers for chat history and asynchronously return the result. git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@5993 542714f4-19e9-0310-aa3c-eee0fc999fb1 --- .../crowd/peer/server/CrowdPeerManager.java | 144 +++++++++++++++++- 1 file changed, 142 insertions(+), 2 deletions(-) diff --git a/src/java/com/threerings/crowd/peer/server/CrowdPeerManager.java b/src/java/com/threerings/crowd/peer/server/CrowdPeerManager.java index d9ef20d34..ea86e4734 100644 --- a/src/java/com/threerings/crowd/peer/server/CrowdPeerManager.java +++ b/src/java/com/threerings/crowd/peer/server/CrowdPeerManager.java @@ -21,15 +21,24 @@ package com.threerings.crowd.peer.server; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.Set; + import com.google.common.base.Predicate; import com.google.common.collect.Iterables; +import com.google.common.collect.Sets; import com.google.inject.Inject; import com.google.inject.internal.Lists; +import com.samskivert.util.Comparators; +import com.samskivert.util.Interval; import com.samskivert.util.Lifecycle; +import com.samskivert.util.ResultListener; import com.threerings.util.Name; -import com.threerings.presents.client.InvocationService.ResultListener; +import com.threerings.presents.client.InvocationService; import com.threerings.presents.data.ClientObject; import com.threerings.presents.peer.data.ClientInfo; import com.threerings.presents.peer.data.NodeObject; @@ -48,12 +57,27 @@ import com.threerings.crowd.data.BodyObject; import com.threerings.crowd.peer.data.CrowdClientInfo; import com.threerings.crowd.peer.data.CrowdNodeObject; +import static com.threerings.crowd.Log.log; + /** * Extends the standard peer manager and bridges certain Crowd services. */ public abstract class CrowdPeerManager extends PeerManager implements CrowdPeerProvider, ChatProvider.ChatForwarder { + /** + * 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; + } + /** * Creates an uninitialized peer manager. */ @@ -80,7 +104,8 @@ public abstract class CrowdPeerManager extends PeerManager } // from interface CrowdPeerProvider - public void getChatHistory (ClientObject caller, Name user, ResultListener lner) + public void getChatHistory ( + ClientObject caller, Name user, InvocationService.ResultListener lner) throws InvocationException { lner.requestProcessed(Lists.newArrayList(Iterables.filter( @@ -124,6 +149,16 @@ 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 () { @@ -177,6 +212,104 @@ 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() { + @Override public void requestProcessed (Object result) { + processed(fpeer, result); + } + @Override 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 InvocationManager _invmgr; @Inject protected ChatProvider _chatprov; @@ -186,4 +319,11 @@ public abstract class CrowdPeerManager extends PeerManager return entry.message instanceof UserMessage; } }; + + protected static final Comparator SORT_BY_TIMESTAMP = + new Comparator() { + @Override public int compare (ChatHistoryEntry e1, ChatHistoryEntry e2) { + return Comparators.compare(e1.message.timestamp, e2.message.timestamp); + } + }; }