A new type of general communication between peers - the NodeRequest - it's a lot like a NodeAction but produces a result which is sent back via the ResultListener.

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

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

@@ -41,6 +41,12 @@ public interface PeerService extends InvocationService
      */
     void invokeAction (Client client, byte[] serializedAction);
 
+    /**
+     * Requests that the specified request be invoked on this server and wants a confirmation
+     *  when it's complete.
+     */
+    void invokeRequest (Client client, byte[] serializedAction, ResultListener listener);
+
     /**
      * Generates a server status report for this peer and returns it to the supplied listener. The
      * result must be a string.

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

@@ -64,8 +64,21 @@ public class PeerMarshaller extends InvocationMarshaller
         });
     }
 
+    /** The method id used to dispatch {@link #invokeRequest} requests. */
+    public static final int INVOKE_REQUEST = 3;
+
+    // from interface PeerService
+    public void invokeRequest (Client arg1, byte[] arg2, InvocationService.ResultListener arg3)
+    {
+        InvocationMarshaller.ResultMarshaller listener3 = new InvocationMarshaller.ResultMarshaller();
+        listener3.listener = arg3;
+        sendRequest(arg1, INVOKE_REQUEST, new Object[] {
+            arg2, listener3
+        });
+    }
+
     /** The method id used to dispatch {@link #ratifyLockAction} requests. */
-    public static final int RATIFY_LOCK_ACTION = 3;
+    public static final int RATIFY_LOCK_ACTION = 4;
 
     // from interface PeerService
     public void ratifyLockAction (Client arg1, NodeObject.Lock arg2, boolean arg3)

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

@@ -70,6 +70,12 @@ public class PeerDispatcher extends InvocationDispatcher<PeerMarshaller>
             );
             return;
 
+        case PeerMarshaller.INVOKE_REQUEST:
+            ((PeerProvider)provider).invokeRequest(
+                source, (byte[])args[0], (InvocationService.ResultListener)args[1]
+            );
+            return;
+
         case PeerMarshaller.RATIFY_LOCK_ACTION:
             ((PeerProvider)provider).ratifyLockAction(
                 source, (NodeObject.Lock)args[0], ((Boolean)args[1]).booleanValue()

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

@@ -65,6 +65,7 @@ import com.threerings.presents.dobj.Subscriber;
 
 import com.threerings.presents.annotation.MainInvoker;
 import com.threerings.presents.client.Client;
+import com.threerings.presents.client.InvocationService;
 import com.threerings.presents.data.ClientObject;
 import com.threerings.presents.server.ClientManager;
 import com.threerings.presents.server.InvocationException;
@@ -164,6 +165,20 @@ public abstract class PeerManager
         protected abstract void execute ();
     }
 
+    public static abstract class NodeRequest implements Streamable
+    {
+        /** Invokes the action on the target server. */
+        public void invoke (final InvocationService.ResultListener listener) {
+            try {
+                execute(listener);
+            } catch (Throwable t) {
+                log.warning(getClass().getName() + " failed.", t);
+            }
+        }
+
+        protected abstract void execute (InvocationService.ResultListener listener);
+    }
+
     /** Returned by {@link #getStats}. */
     public static class Stats implements Cloneable
     {
@@ -420,6 +435,21 @@ public abstract class PeerManager
         }
     }
 
+    /**
+     * Invokes a node request on a specific node and returns the result through the listener.
+     */
+    public void invokeNodeRequest (String nodeName, NodeRequest request,
+        InvocationService.ResultListener listener)
+    {
+        PeerNode peer = _peers.get(nodeName);
+        if (peer != null) {
+            peer.nodeobj.peerService.invokeRequest(
+                peer.getClient(), flattenRequest(request), listener);
+        } else if (nodeName.equals(_nodeName)) {
+            invokeRequest(null, flattenRequest(request), listener);
+        }
+    }
+
     /**
      * Initiates a proxy on an object that is managed by the specified peer. The object will be
      * proxied into this server's distributed object space and its local oid reported back to the
@@ -822,6 +852,26 @@ public abstract class PeerManager
         }
     }
 
+    // from interface PeerProvider
+    public void invokeRequest (ClientObject caller, byte[] serializedAction,
+        InvocationService.ResultListener listener)
+    {
+        NodeRequest request = null;
+        try {
+            ObjectInputStream oin =
+                new ObjectInputStream(new ByteArrayInputStream(serializedAction));
+            request = (NodeRequest)oin.readObject();
+            _injector.injectMembers(request);
+            request.invoke(listener);
+
+        } catch (Exception e) {
+            log.warning("Failed to execute node request",
+                        "from", (caller == null) ? "self" : caller.who(),
+                        "request", request, "serializedSize", serializedAction.length, e);
+            listener.requestFailed("Failed to execute node request");
+        }
+    }
+
     // from interface PeerProvider
     public void generateReport (ClientObject caller, String type,
                                 PeerService.ResultListener listener)
@@ -1142,6 +1192,22 @@ public abstract class PeerManager
         }
     }
 
+    /**
+     * Flattens the supplied node request into bytes.
+     */
+    protected byte[] flattenRequest (NodeRequest request)
+    {
+        try {
+            ByteArrayOutputStream bout = new ByteArrayOutputStream();
+            ObjectOutputStream oout = new ObjectOutputStream(bout);
+            oout.writeObject(request);
+            return bout.toByteArray();
+        } catch (Exception e) {
+            throw new IllegalArgumentException(
+                "Failed to serialize node request [request=" + request + "].", e);
+        }
+    }
+
     protected void lockAcquired (NodeObject.Lock lock, long wait, ResultListener<String> listener)
     {
         _nodeobj.addToLocks(lock);

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

@@ -48,6 +48,12 @@ public interface PeerProvider extends InvocationProvider
      */
     void invokeAction (ClientObject caller, byte[] arg1);
 
+    /**
+     * Handles a {@link PeerService#invokeRequest} request.
+     */
+    void invokeRequest (ClientObject caller, byte[] arg1, InvocationService.ResultListener arg2)
+        throws InvocationException;
+
     /**
      * Handles a {@link PeerService#ratifyLockAction} request.
      */