diff --git a/autobahn/src/main/java/io/crossbar/autobahn/wamp/Session.java b/autobahn/src/main/java/io/crossbar/autobahn/wamp/Session.java index 80f83875..21a1eff8 100644 --- a/autobahn/src/main/java/io/crossbar/autobahn/wamp/Session.java +++ b/autobahn/src/main/java/io/crossbar/autobahn/wamp/Session.java @@ -322,7 +322,6 @@ private void onPreSessionMessage(IMessage message) throws Exception { private void onMessage(IMessage message) throws Exception { if (message instanceof Result) { Result msg = (Result) message; - CallRequest request = getOrDefault(mCallRequests, msg.request, null); if (request == null) { throw new ProtocolError(String.format( diff --git a/demo-gallery/src/main/java/io/crossbar/autobahn/demogallery/ProgressiveCallResultsExample.java b/demo-gallery/src/main/java/io/crossbar/autobahn/demogallery/ProgressiveCallResultsExample.java new file mode 100644 index 00000000..746f2ece --- /dev/null +++ b/demo-gallery/src/main/java/io/crossbar/autobahn/demogallery/ProgressiveCallResultsExample.java @@ -0,0 +1,61 @@ +package io.crossbar.autobahn.demogallery; + +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; + +import io.crossbar.autobahn.wamp.Client; +import io.crossbar.autobahn.wamp.Session; +import io.crossbar.autobahn.wamp.types.CallOptions; +import io.crossbar.autobahn.wamp.types.CallResult; +import io.crossbar.autobahn.wamp.types.ExitInfo; +import io.crossbar.autobahn.wamp.types.InvocationDetails; +import io.crossbar.autobahn.wamp.types.InvocationResult; +import io.crossbar.autobahn.wamp.types.Registration; + +public class ProgressiveCallResultsExample { + + public static CompletableFuture registerProgressive(String wsAddress, String realm) { + Session wampSession = new Session(); + wampSession.addOnJoinListener((session, details) -> { + CompletableFuture regFuture = session.register( + "io.crossbar.longop", + (List args, Map kwargs, InvocationDetails invocationDetails) -> { + for (int i = 0; i < 5; i++) { + List argsList = new ArrayList<>(); + argsList.add(i); + invocationDetails.progress.sendProgress(argsList, null); + } + List resultArgs = new ArrayList<>(); + resultArgs.add(7); + return CompletableFuture.completedFuture(new InvocationResult(resultArgs)); + }); + + regFuture.whenComplete((registration, throwable) -> { + System.out.println(String.format( + "Registered procedure %s", registration.procedure)); + }); + }); + + Client wampClient = new Client(wampSession, wsAddress, realm); + return wampClient.connect(); + } + + + public static CompletableFuture callProgressive(String wsAddress, String realm) { + Session wampSession = new Session(); + wampSession.addOnJoinListener((session, details) -> { + CompletableFuture callFuture = session.call( + "io.crossbar.longop", + new CallOptions(result -> System.out.println("Receive Progress: " + result.results))); + + callFuture.whenComplete((callResult, throwable) -> { + System.out.println(String.format("Call result: %s", callResult.results)); + }); + }); + + Client wampClient = new Client(wampSession, wsAddress, realm); + return wampClient.connect(); + } +}