From 6adabaf44228dc226e8765b760ebab688b37a983 Mon Sep 17 00:00:00 2001 From: Michael Bayne Date: Thu, 18 Sep 2008 18:07:19 +0000 Subject: [PATCH] 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 --- .../crowd/chat/server/ChatChannelManager.java | 94 +++++++++++++++---- 1 file changed, 77 insertions(+), 17 deletions(-) diff --git a/src/java/com/threerings/crowd/chat/server/ChatChannelManager.java b/src/java/com/threerings/crowd/chat/server/ChatChannelManager.java index a0ab066a2..29c2a8fa3 100644 --- a/src/java/com/threerings/crowd/chat/server/ChatChannelManager.java +++ b/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); - if (parts == null) { + final ChannelInfo info = _channels.get(channel); + 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 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)) { + for (ClientInfo clinfo : nodeobj.clients) { + int bodyId = getBodyId(((CrowdClientInfo)clinfo).visibleName); + 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> iter = _channels.entrySet().iterator(); + while (iter.hasNext()) { + Map.Entry 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); - if (partSet != null) { + ChannelInfo info = _channelMan._channels.get(_channel); + 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> _resolving = Maps.newHashMap(); - /** A map of resolved channels to the body ids of their participants. */ - protected Map _channels = Maps.newHashMap(); + /** A map of resolved channels to metadata records. */ + protected Map _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; }