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

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.*;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
 
+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<String> failedNodes;
+
+        /** The things in the user's chat history, aggregated from all nodes and sorted by
+         * timestamp. */
+        public List<ChatHistoryEntry> 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<ChatHistoryResult> lner)
+    {
+        _peerMan.invokeNodeRequest (
+            new ChatCollectionRequest(user), new NodeRequestsListener<List<ChatHistoryEntry>>()
+        {
+            @Override public void requestsProcessed (
+                NodeRequestsResult<List<ChatHistoryEntry>> requestResult)
+            {
+                ChatHistoryResult result = new ChatHistoryResult();
+                result.failedNodes = requestResult.getNodeErrors().keySet();
+                result.history = Lists.newArrayList();
+                for (Map.Entry<String, List<ChatHistoryEntry>> 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<ChatHistoryEntry> IS_USER_MESSAGE =
+        new Predicate<ChatHistoryEntry>() {
+        public boolean apply (ChatHistoryEntry entry) {
+            return entry.message instanceof UserMessage;
+        }
+    };
+
+    protected static final Comparator<ChatHistoryEntry> SORT_BY_TIMESTAMP =
+        new Comparator<ChatHistoryEntry>() {
+        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

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);
 }

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
-        });
-    }
 }

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<CrowdPeerMarshalle
             );
             return;
 
-        case CrowdPeerMarshaller.GET_CHAT_HISTORY:
-            ((CrowdPeerProvider)provider).getChatHistory(
-                source, (Name)args[0], (InvocationService.ResultListener)args[1]
-            );
-            return;
-
         default:
             super.dispatchRequest(source, methodId, args);
             return;

src/java/com/threerings/crowd/peer/server/CrowdPeerManager.java

@@ -64,19 +64,6 @@ import static com.threerings.crowd.Log.log;
 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<String> failedNodes;
-
-        /** The things in the user's chat history, aggregated from all nodes and sorted by
-         * timestamp. */
-        public List<ChatHistoryEntry> 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<ChatHistoryResult> 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<ChatHistoryResult> 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<ChatHistoryEntry> nodeMessages = (List<ChatHistoryEntry>)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<ChatHistoryResult> _listener;
-        protected ChatHistoryResult _result;
-        protected Set<String> _waiting;
-    }
-
     @Inject protected ChatProvider _chatprov;
-
-    protected static final Predicate<ChatHistoryEntry> IS_USER_MESSAGE =
-        new Predicate<ChatHistoryEntry>() {
-        public boolean apply (ChatHistoryEntry entry) {
-            return entry.message instanceof UserMessage;
-        }
-    };
-
-    protected static final Comparator<ChatHistoryEntry> SORT_BY_TIMESTAMP =
-        new Comparator<ChatHistoryEntry>() {
-        public int compare (ChatHistoryEntry e1, ChatHistoryEntry e2) {
-            return Longs.compare(e1.message.timestamp, e2.message.timestamp);
-        }
-    };
 }

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;
 }