Added cache management to the peer manager so it can inform servers that persistent data has

changed and they'll need to reload it


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

src/java/com/threerings/presents/peer/data/NodeObject.java

@@ -21,6 +21,8 @@
 
 package com.threerings.presents.peer.data;
 
+import com.threerings.io.Streamable;
+
 import com.threerings.presents.dobj.DObject;
 import com.threerings.presents.dobj.DSet;
 
@@ -29,14 +31,41 @@ import com.threerings.presents.dobj.DSet;
  */
 public class NodeObject extends DObject
 {
+    /** Used for informing peers of changes to persistent data. */
+    public static class CacheData implements Streamable
+    {
+        /** The cache that should be purged. */
+        public String cache;
+
+        /** The stale data in the cache. */
+        public Streamable data;
+
+        public CacheData () 
+        { 
+        }
+
+        public CacheData (String cache, Streamable data)
+        {
+            this.cache = cache;
+            this.data = data;
+        }
+
+    }
+
     // AUTO-GENERATED: FIELDS START
     /** The field name of the <code>clients</code> field. */
     public static final String CLIENTS = "clients";
+
+    /** The field name of the <code>cacheData</code> field. */
+    public static final String CACHE_DATA = "cacheData";
     // AUTO-GENERATED: FIELDS END
 
     /** Contains information on all clients connected to this node. */
     public DSet<ClientInfo> clients = new DSet<ClientInfo>();
 
+    /** A field we use to broadcast changes to possible cached data. */
+    public CacheData cacheData;
+
     // AUTO-GENERATED: METHODS START
     /**
      * Requests that the specified entry be added to the
@@ -85,5 +114,21 @@ public class NodeObject extends DObject
             (value == null) ? null : value.typedClone();
         this.clients = clone;
     }
+
+    /**
+     * Requests that the <code>cacheData</code> field be set to the
+     * specified value. The local value will be updated immediately and an
+     * event will be propagated through the system to notify all listeners
+     * that the attribute did change. Proxied copies of this object (on
+     * clients) will apply the value change when they received the
+     * attribute changed notification.
+     */
+    public void setCacheData (NodeObject.CacheData value)
+    {
+        NodeObject.CacheData ovalue = this.cacheData;
+        requestAttributeChange(
+            CACHE_DATA, value, ovalue);
+        this.cacheData = value;
+    }
     // AUTO-GENERATED: METHODS END
 }

src/java/com/threerings/presents/peer/server/PeerManager.java

@@ -30,11 +30,15 @@ import com.samskivert.io.PersistenceException;
 import com.samskivert.jdbc.ConnectionProvider;
 import com.samskivert.util.Interval;
 import com.samskivert.util.Invoker;
+import com.samskivert.util.ObserverList;
 
+import com.threerings.io.Streamable;
 import com.threerings.util.Name;
 
 import com.threerings.presents.client.Client;
 import com.threerings.presents.client.ClientObserver;
+import com.threerings.presents.dobj.AttributeChangedEvent;
+import com.threerings.presents.dobj.AttributeChangeListener;
 import com.threerings.presents.dobj.ObjectAccessException;
 import com.threerings.presents.dobj.Subscriber;
 import com.threerings.presents.server.ClientManager;
@@ -58,6 +62,18 @@ import static com.threerings.presents.Log.log;
 public class PeerManager
     implements ClientManager.ClientObserver
 {
+    /**
+     * Used by entities that wish to know when cached data has become stale due to a change on
+     * one of our peer servers.
+     */
+    public static interface StaleCacheObserver
+    {
+        /**
+         * Called when some possibly cached data has changed on one of our peer servers.
+         */
+        public void changedCacheData (Streamable data);
+    }
+
     /**
      * Creates a peer manager which will create a {@link NodeRepository} which will be used to
      * publish our existence and discover the other nodes.
@@ -209,6 +225,42 @@ public class PeerManager
         log.warning("Session ended for unregistered client [who=" + username + "].");
     }
 
+    /**
+     * Registers a stale cache observer.
+     */
+    public void addStaleCacheObserver (String cache, StaleCacheObserver observer)
+    {
+        ObserverList<StaleCacheObserver> list = _cacheobs.get(cache);
+        if (list == null) {
+            list = new ObserverList<StaleCacheObserver>(ObserverList.FAST_UNSAFE_NOTIFY);
+            _cacheobs.put(cache, list);
+        }
+        list.add(observer);
+    }
+
+    /**
+     * Removes a stale cache observer registration.
+     */
+    public void removeStaleCacheObserver (String cache, StaleCacheObserver observer)
+    {
+        ObserverList<StaleCacheObserver> list = _cacheobs.get(cache);
+        if (list == null) {
+            return;
+        }
+        list.remove(observer);
+        if (list.isEmpty()) {
+            _cacheobs.remove(cache);
+        }
+    }
+
+    /**
+     * Called when cached data has changed on the local server and needs to inform our peers.
+     */
+    public void broadcastStaleCacheData (String cache, Streamable data)
+    {
+        _nodeobj.setCacheData(new NodeObject.CacheData(cache, data));
+    }
+
     /**
      * Called after we have finished our initialization.
      */
@@ -281,6 +333,25 @@ public class PeerManager
         return new ClientInfo();
     }
 
+    /**
+     * Called when possibly cached data has changed on one of our peer servers.
+     */
+    protected void changedCacheData (String cache, final Streamable data)
+    {
+        // see if we have any observers
+        ObserverList<StaleCacheObserver> list = _cacheobs.get(cache);
+        if (list == null) {
+            return;
+        }
+        // if so, notify them
+        list.apply(new ObserverList.ObserverOp<StaleCacheObserver>() {
+            public boolean apply (StaleCacheObserver observer) {
+                observer.changedCacheData(data);
+                return true;
+            }
+        });
+    }
+
     /**
      * Initializes the supplied client info for the supplied client.
      */
@@ -301,7 +372,7 @@ public class PeerManager
      * Contains all runtime information for one of our peer nodes.
      */
     protected class PeerNode
-        implements ClientObserver, Subscriber<NodeObject>
+        implements ClientObserver, Subscriber<NodeObject>, AttributeChangeListener
     {
         /** This peer's node object. */
         public NodeObject nodeobj;
@@ -406,6 +477,7 @@ public class PeerManager
         public void objectAvailable (NodeObject object)
         {
             nodeobj = object;
+            nodeobj.addListener(this);
             // TODO: stuff!
         }
 
@@ -416,6 +488,15 @@ public class PeerManager
                         "[peer=" + _record + ", cause=" + cause + "].");
         }
 
+        // documentation inherited from interface AttributeChangeListener
+        public void attributeChanged (AttributeChangedEvent event)
+        {
+            if (NodeObject.CACHE_DATA.equals(event.getName())) {
+                NodeObject.CacheData cd = (NodeObject.CacheData)event.getValue();
+                changedCacheData(cd.cache, cd.data);
+            }
+        }
+
         protected NodeRecord _record;
         protected Client _client;
         protected long _lastConnectStamp;
@@ -427,4 +508,8 @@ public class PeerManager
     protected NodeRepository _noderepo;
     protected NodeObject _nodeobj;
     protected HashMap<String,PeerNode> _peers = new HashMap<String,PeerNode>();
+
+    /** Our stale cache observers. */
+    protected HashMap<String, ObserverList<StaleCacheObserver>> _cacheobs =
+        new HashMap<String, ObserverList<StaleCacheObserver>>();
 }