Close resolved channels that have been idle for 5 minutes.

git-svn-id: svn+ssh://src.earth.threerings.net/narya/trunk@5383 542714f4-19e9-0310-aa3c-eee0fc999fb1

src/java/com/threerings/crowd/chat/server/ChatChannelManager.java

@@ -21,6 +21,7 @@
 
 package com.threerings.crowd.chat.server;
 
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 
@@ -32,6 +33,7 @@ import com.google.inject.Singleton;
 
 import com.samskivert.util.ArrayIntSet;
 import com.samskivert.util.IntSet;
+import com.samskivert.util.Interval;
 import com.threerings.util.Name;
 
 import com.threerings.presents.annotation.AnyThread;
@@ -40,7 +42,8 @@ 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.presents.server.PresentsDObjectMgr;
+import com.threerings.presents.server.ShutdownManager;
 
 import com.threerings.crowd.data.BodyObject;
 import com.threerings.crowd.data.CrowdCodes;
@@ -59,7 +62,7 @@ import static com.threerings.crowd.Log.log;
  */
 @Singleton
 public abstract class ChatChannelManager
-    implements ChannelSpeakProvider
+    implements ChannelSpeakProvider, ShutdownManager.Shutdowner
 {
     /**
      * When a body becomes a member of a channel, this method should be called so that any server
@@ -111,12 +114,29 @@ public abstract class ChatChannelManager
         });
     }
 
+    // from interface ShutdownManager.Shutdowner
+    public void shutdown ()
+    {
+        // stop our channel closer; always be closing... except now
+        _closer.cancel();
+        _closer = null;
+    }
+
     /**
      * Creates our singleton manager and registers our invocation service.
      */
-    @Inject protected ChatChannelManager (InvocationManager invmgr)
+    @Inject protected ChatChannelManager (PresentsDObjectMgr omgr, InvocationManager invmgr,
+                                          ShutdownManager shutmgr)
     {
         invmgr.registerDispatcher(new ChannelSpeakDispatcher(this), CrowdCodes.CROWD_GROUP);
+
+        // create and start our idle channel closer (always be closing)
+        _closer = new Interval(omgr) {
+            public void expired () {
+                closeIdleChannels();
+            }
+        };
+        _closer.schedule(IDLE_CHANNEL_CHECK_PERIOD, true);
     }
 
     /**
@@ -163,7 +183,9 @@ public abstract class ChatChannelManager
     protected void resolutionComplete (ChatChannel channel, IntSet parts)
     {
         // map the participants of our now resolved channel
-        _channels.put(channel, parts);
+        ChannelInfo info = new ChannelInfo();
+        info.participants = parts;
+        _channels.put(channel, info);
 
         // dispatch any pending messages now that we know where they go
         for (UserMessage msg : _resolving.remove(channel)) {
@@ -190,8 +212,8 @@ public abstract class ChatChannelManager
      */
     protected void dispatchSpeak (ChatChannel channel, UserMessage message)
     {
-        final IntSet parts = _channels.get(channel);
+        final ChannelInfo info = _channels.get(channel);
-        if (parts == null) {
+        if (info == null) {
             // TODO: maybe we should just reresolve the channel...
             log.warning("Requested to dispatch speak on unhosted channel", "channel", channel,
                         "msg", message);
@@ -199,21 +221,24 @@ public abstract class ChatChannelManager
         }
 
         // validate the speaker
-        if (!parts.contains(getBodyId(message.speaker))) {
+        if (!info.participants.contains(getBodyId(message.speaker))) {
             log.warning("Dropping channel chat message from non-speaker", "channel", channel,
                         "message", message);
             return;
         }
 
+        // note that we're dispatching a message on this channel
+        info.lastMessage = System.currentTimeMillis();
+
         // 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<String,int[]> partMap = Maps.newHashMap();
         _peerMan.applyToNodes(new Function<NodeObject,Void>() {
             public Void apply (NodeObject nodeobj) {
                 ArrayIntSet nodeBodyIds = new ArrayIntSet();
-                for (ClientInfo info : nodeobj.clients) {
+                for (ClientInfo clinfo : nodeobj.clients) {
-                    int bodyId = getBodyId(((CrowdClientInfo)info).visibleName);
+                    int bodyId = getBodyId(((CrowdClientInfo)clinfo).visibleName);
-                    if (parts.contains(bodyId)) {
+                    if (info.participants.contains(bodyId)) {
                         nodeBodyIds.add(bodyId);
                     }
                 }
@@ -241,6 +266,22 @@ public abstract class ChatChannelManager
         }
     }
 
+    /**
+     * Called periodically to check for and close any channels that have been idle too long.
+     */
+    protected void closeIdleChannels ()
+    {
+        long now = System.currentTimeMillis();
+        Iterator<Map.Entry<ChatChannel, ChannelInfo>> iter = _channels.entrySet().iterator();
+        while (iter.hasNext()) {
+            Map.Entry<ChatChannel, ChannelInfo> entry = iter.next();
+            if (now - entry.getValue().lastMessage > IDLE_CHANNEL_CLOSE_TIME) {
+                ((CrowdNodeObject)_peerMan.getNodeObject()).removeFromHostedChannels(entry.getKey());
+                iter.remove();
+            }
+        }
+    }
+
     /**
      * 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
@@ -293,12 +334,12 @@ public abstract class ChatChannelManager
         public ParticipantChanged () {
         }
         @Override protected void execute () {
-            IntSet partSet = _channelMan._channels.get(_channel);
+            ChannelInfo info = _channelMan._channels.get(_channel);
-            if (partSet != null) {
+            if (info != null) {
                 if (_added) {
-                    partSet.add(_bodyId);
+                    info.participants.add(_bodyId);
                 } else {
-                    partSet.remove(_bodyId);
+                    info.participants.remove(_bodyId);
                 }
             } else if (_channelMan._resolving.containsKey(_channel)) {
                 log.warning("Oh for fuck's sake, distributed systems are complicated",
@@ -344,12 +385,31 @@ public abstract class ChatChannelManager
         protected int[] _bodyIds;
     }
 
+    /** Contains metadata for a particular channel. */
+    protected static class ChannelInfo
+    {
+        /** The body ids of the participants of this channel. */
+        public IntSet participants;
+
+        /** The time at which a message was last dispatched on this channel. */
+        public long lastMessage;
+    }
+
+    /** Used to close channels that have not had any activity in a few minutes. */
+    protected Interval _closer;
+
     /** Contains pending messages for all channels currently being resolved. */
     protected Map<ChatChannel,List<UserMessage>> _resolving = Maps.newHashMap();
 
-    /** A map of resolved channels to the body ids of their participants. */
+    /** A map of resolved channels to metadata records. */
-    protected Map<ChatChannel,IntSet> _channels = Maps.newHashMap();
+    protected Map<ChatChannel,ChannelInfo> _channels = Maps.newHashMap();
 
+    /** Provides peer services. */
     @Inject protected CrowdPeerManager _peerMan;
-    @Inject protected PresentsInvoker _invoker;
+
+    /** The period on which we check for idle channels. */
+    protected static final long IDLE_CHANNEL_CHECK_PERIOD = 5 * 1000L;
+
+    /** The amount of idle time (in milliseconds) after which we close a channel. */
+    protected static final long IDLE_CHANNEL_CLOSE_TIME = 5 * 60 * 1000L;
 }