From 7101012413c645f678b281e8468f3d857e65c21f Mon Sep 17 00:00:00 2001 From: Mike Thomas Date: Tue, 1 Jun 2010 23:48:50 +0000 Subject: [PATCH] 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 --- .../presents/peer/client/PeerService.java | 6 ++ .../presents/peer/data/PeerMarshaller.java | 15 ++++- .../presents/peer/server/PeerDispatcher.java | 6 ++ .../presents/peer/server/PeerManager.java | 66 +++++++++++++++++++ .../presents/peer/server/PeerProvider.java | 6 ++ 5 files changed, 98 insertions(+), 1 deletion(-) diff --git a/src/java/com/threerings/presents/peer/client/PeerService.java b/src/java/com/threerings/presents/peer/client/PeerService.java index 7a53cb6ba..ae8c40af5 100644 --- a/src/java/com/threerings/presents/peer/client/PeerService.java +++ b/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. diff --git a/src/java/com/threerings/presents/peer/data/PeerMarshaller.java b/src/java/com/threerings/presents/peer/data/PeerMarshaller.java index 204afa103..ebadb5bca 100644 --- a/src/java/com/threerings/presents/peer/data/PeerMarshaller.java +++ b/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) diff --git a/src/java/com/threerings/presents/peer/server/PeerDispatcher.java b/src/java/com/threerings/presents/peer/server/PeerDispatcher.java index e59bbdb44..a794e10b0 100644 --- a/src/java/com/threerings/presents/peer/server/PeerDispatcher.java +++ b/src/java/com/threerings/presents/peer/server/PeerDispatcher.java @@ -70,6 +70,12 @@ public class PeerDispatcher extends InvocationDispatcher ); 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() diff --git a/src/java/com/threerings/presents/peer/server/PeerManager.java b/src/java/com/threerings/presents/peer/server/PeerManager.java index 4dc08bde7..454a60364 100644 --- a/src/java/com/threerings/presents/peer/server/PeerManager.java +++ b/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 listener) { _nodeobj.addToLocks(lock); diff --git a/src/java/com/threerings/presents/peer/server/PeerProvider.java b/src/java/com/threerings/presents/peer/server/PeerProvider.java index c8aac16d0..9344c770f 100644 --- a/src/java/com/threerings/presents/peer/server/PeerProvider.java +++ b/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. */