Added a way to package up an action and data and send it to all appropriate

servers for execution. The class in question must exist on the remote server
but all of our servers are all always running the same code, so this follows.


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

src/java/com/threerings/presents/peer/client/PeerService.java

@@ -36,4 +36,9 @@ public interface PeerService extends InvocationService
      * lock.
      */
     public void ratifyLockAction (Client client, Lock lock, boolean acquire);
+
+    /**
+     * Requests that the specified action be invoked on this server.
+     */
+    public void invokeAction (Client client, byte[] serializedAction);
 }

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

@@ -37,8 +37,19 @@ import com.threerings.presents.peer.data.NodeObject;
 public class PeerMarshaller extends InvocationMarshaller
     implements PeerService
 {
+    /** The method id used to dispatch {@link #invokeAction} requests. */
+    public static final int INVOKE_ACTION = 1;
+
+    // from interface PeerService
+    public void invokeAction (Client arg1, byte[] arg2)
+    {
+        sendRequest(arg1, INVOKE_ACTION, new Object[] {
+            arg2
+        });
+    }
+
     /** The method id used to dispatch {@link #ratifyLockAction} requests. */
-    public static final int RATIFY_LOCK_ACTION = 1;
+    public static final int RATIFY_LOCK_ACTION = 2;
 
     // from interface PeerService
     public void ratifyLockAction (Client arg1, NodeObject.Lock arg2, boolean arg3)

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

@@ -44,18 +44,26 @@ public class PeerDispatcher extends InvocationDispatcher
         this.provider = provider;
     }
 
-    // from InvocationDispatcher
+    @Override // documentation inherited
     public InvocationMarshaller createMarshaller ()
     {
         return new PeerMarshaller();
     }
 
-    @SuppressWarnings("unchecked") // from InvocationDispatcher
+    @SuppressWarnings("unchecked")
+    @Override // documentation inherited
     public void dispatchRequest (
         ClientObject source, int methodId, Object[] args)
         throws InvocationException
     {
         switch (methodId) {
+        case PeerMarshaller.INVOKE_ACTION:
+            ((PeerProvider)provider).invokeAction(
+                source,
+                (byte[])args[0]
+            );
+            return;
+
         case PeerMarshaller.RATIFY_LOCK_ACTION:
             ((PeerProvider)provider).ratifyLockAction(
                 source,

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

@@ -21,6 +21,12 @@
 
 package com.threerings.presents.peer.server;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.Iterator;
@@ -37,6 +43,7 @@ import com.samskivert.util.ObjectUtil;
 import com.samskivert.util.ObserverList;
 import com.samskivert.util.ResultListener;
 import com.samskivert.util.ResultListenerList;
+import com.samskivert.util.StringUtil;
 import com.samskivert.util.Tuple;
 
 import com.threerings.io.Streamable;
@@ -134,6 +141,43 @@ public class PeerManager
         public void fail (String peerName);
     }
 
+    /**
+     * Encapsulates code that is meant to be executed one or more servers.
+     *
+     * <p><b>Note well</b>: the action you provide is serialized and sent to the server to which
+     * the member is currently connection. This means you MUST NOT instantiate a NodeAction
+     * anonymously and make reference to other classes because those implicit references will cause
+     * the referenced classes to be included in the anonymous inner class's serialization closure.
+     * Instead use the provided varargs constructor to pass along any information you need which
+     * will be serialized and sent to the destination server. This means that said arguments must
+     * of course be {@link Serializable}.
+     */
+    public static abstract class NodeAction implements Serializable
+    {
+        public NodeAction (Serializable ... arguments) {
+            _arguments = arguments;
+        }
+
+        /** Returns true if this action should be executed on the specified node. This will be
+         * called on the originating server to decide whether or not to deliver the action to the
+         * server in question. */
+        public abstract boolean isApplicable (NodeObject nodeobj);
+
+        /** Invokes the action on the target server. */
+        public void invoke () {
+            try {
+                execute(_arguments);
+            } catch (Throwable t) {
+                log.log(Level.WARNING, getClass().getName() + " failed " +
+                        StringUtil.safeToString(_arguments) + ".");
+            }
+        }
+
+        protected abstract void execute (Object[] arguments);
+
+        protected Object[] _arguments;
+    }
+
     /**
      * Returns the distributed object that represents this node to its peers.
      */
@@ -212,6 +256,15 @@ public class PeerManager
         didInit();
     }
 
+    /**
+     * Returns true if the supplied peer credentials match our shared secret.
+     */
+    public boolean isAuthenticPeer (PeerCreds creds)
+    {
+        return PeerCreds.createPassword(creds.getNodeName(), _sharedSecret).equals(
+            creds.getPassword());
+    }
+
     /**
      * Locates the client with the specified name. Returns null if the client is not logged onto
      * any peer.
@@ -278,12 +331,45 @@ public class PeerManager
     }
 
     /**
-     * Returns true if the supplied peer credentials match our shared secret.
+     * Invokes the supplied action on this and any other server that it indicates is appropriate.
+     * The action will be executed on the distributed object thread, but this method does not need
+     * to be called from the distributed object thread.
      */
-    public boolean isAuthenticPeer (PeerCreds creds)
+    public void invokeNodeAction (final NodeAction action)
     {
-        return PeerCreds.createPassword(creds.getNodeName(), _sharedSecret).equals(
-            creds.getPassword());
+        // if we're not on the dobjmgr thread, get there
+        if (!PresentsServer.omgr.isDispatchThread()) {
+            PresentsServer.omgr.postRunnable(new Runnable() {
+                public void run () {
+                    invokeNodeAction(action);
+                }
+            });
+            return;
+        }
+
+        // first serialize the action to make sure we can
+        byte[] actionBytes;
+        try {
+            ByteArrayOutputStream bout = new ByteArrayOutputStream();
+            ObjectOutputStream oout = new ObjectOutputStream(bout);
+            oout.writeObject(action);
+            actionBytes = bout.toByteArray();
+        } catch (Exception e) {
+            log.log(Level.WARNING, "Failed to serialize node action [action=" + action + "].", e);
+            return;
+        }
+
+        // invoke the action on our local server if appropriate
+        if (action.isApplicable(_nodeobj)) {
+            action.invoke();
+        }
+
+        // now send it to any remote node that is also appropriate
+        for (PeerNode peer : _peers.values()) {
+            if (peer.nodeobj != null && action.isApplicable(peer.nodeobj)) {
+                peer.nodeobj.peerService.invokeAction(peer.getClient(), actionBytes);
+            }
+        }
     }
 
     /**
@@ -658,6 +744,21 @@ public class PeerManager
         }
     }
 
+    // from interface PeerProvider
+    public void invokeAction (ClientObject caller, byte[] serializedAction)
+    {
+        NodeAction action = null;
+        try {
+            ObjectInputStream oin =
+                new ObjectInputStream(new ByteArrayInputStream(serializedAction));
+            action = (NodeAction)oin.readObject();
+            action.invoke();
+        } catch (Exception e) {
+            log.log(Level.WARNING, "Failed to execute node action [from=" + caller.who() +
+                    ", action=" + action + ", serializedSize=" + serializedAction.length + "].");
+        }
+    }
+
     // from interface ClientManager.ClientObserver
     public void clientSessionDidStart (PresentsClient client)
     {

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

@@ -33,6 +33,11 @@ import com.threerings.presents.server.InvocationProvider;
  */
 public interface PeerProvider extends InvocationProvider
 {
+    /**
+     * Handles a {@link PeerService#invokeAction} request.
+     */
+    public void invokeAction (ClientObject caller, byte[] arg1);
+
     /**
      * Handles a {@link PeerService#ratifyLockAction} request.
      */