implement progressive call results

Author: Mahad-10

.github/workflows/main.yaml

@@ -26,7 +26,7 @@ jobs:
         run: |
           git clone https://github.com/xconnio/xconn-aat-setup.git
           cd xconn-aat-setup
-          make build-docker-xconn
+          make build-docker-nxt
           make build-docker-crossbar
           docker compose up -d
           sudo snap install wick --classic

examples/asyncio/rpc_progressive_call_results/callee.py

@@ -0,0 +1,32 @@
+import asyncio
+
+from xconn import run
+from xconn.types import Result, Invocation
+from xconn.async_client import connect_anonymous
+
+
+async def invocation_handler(invocation: Invocation) -> Result:
+    file_size = 100
+    for i in range(0, file_size + 1, 10):
+        progress = i * 100 // file_size
+        try:
+            await invocation.send_progress([progress], {})
+        except Exception as err:
+            return Result(["wamp.error.canceled", str(err)])
+        await asyncio.sleep(0.5)
+
+    return Result(["Download complete!"])
+
+
+async def main() -> None:
+    test_procedure_progress_download = "io.xconn.progress.download"
+
+    # create and connect a callee client to server
+    callee = await connect_anonymous("ws://localhost:8080/ws", "realm1")
+
+    await callee.register(test_procedure_progress_download, invocation_handler)
+    print(f"Registered procedure '{test_procedure_progress_download}'")
+
+
+if __name__ == "__main__":
+    run(main())

examples/asyncio/rpc_progressive_call_results/caller.py

@@ -0,0 +1,21 @@
+from xconn import run
+from xconn.types import Result
+from xconn.async_client import connect_anonymous
+
+
+async def progress_handler(res: Result) -> None:
+    progress = res.args[0]
+    print(f"Download progress: {progress}%")
+
+
+async def main() -> None:
+    test_procedure_progress_download = "io.xconn.progress.download"
+
+    # create and connect a callee client to server
+    caller = await connect_anonymous("ws://localhost:8080/ws", "realm1")
+
+    await caller.call_progress(test_procedure_progress_download, progress_handler)
+
+
+if __name__ == "__main__":
+    run(main())

examples/sync/rpc_progressive_call_results/callee.py

@@ -0,0 +1,44 @@
+import sys
+import time
+import signal
+
+from xconn.client import connect_anonymous
+from xconn.types import Result, Invocation
+
+
+def invocation_handler(invocation: Invocation) -> Result:
+    file_size = 100
+    for i in range(0, file_size + 1, 10):
+        progress = i * 100 // file_size
+        try:
+            invocation.send_progress([progress], {})
+        except Exception as err:
+            return Result(["wamp.error.canceled", str(err)])
+        time.sleep(0.5)
+
+    return Result(["Download complete!"])
+
+
+if __name__ == "__main__":
+    test_procedure_progress_download = "io.xconn.progress.download"
+
+    # create and connect a callee client to server
+    callee = connect_anonymous("ws://localhost:8080/ws", "realm1")
+
+    download_progress_registration = callee.register(test_procedure_progress_download, invocation_handler)
+    print(f"Registered procedure '{test_procedure_progress_download}'")
+
+    def handle_sigint(signum, frame):
+        print("SIGINT received. Cleaning up...")
+
+        # unregister procedure "io.xconn.progress.download"
+        download_progress_registration.unregister()
+
+        # close connection to the server
+        callee.leave()
+
+        sys.exit(0)
+
+
+# register signal handler
+signal.signal(signal.SIGINT, handle_sigint)

examples/sync/rpc_progressive_call_results/caller.py

@@ -0,0 +1,18 @@
+from xconn.types import Result
+from xconn.client import connect_anonymous
+
+
+def progress_handler(res: Result) -> None:
+    progress = res.args[0]
+    print(f"Download progress: {progress}%")
+
+
+if __name__ == "__main__":
+    test_procedure_progress_download = "io.xconn.progress.download"
+
+    # create and connect a callee client to server
+    caller = connect_anonymous("ws://localhost:8080/ws", "realm1")
+
+    caller.call_progress(test_procedure_progress_download, progress_handler)
+
+    caller.leave()

xconn/async_session.py

@@ -4,7 +4,7 @@
 import inspect
 from dataclasses import dataclass
 from asyncio import Future, get_event_loop
-from typing import Callable, Union, Awaitable, Any
+from typing import Callable, Awaitable, Any
 
 from wampproto import messages, idgen, session
 
@@ -68,17 +68,15 @@ def __init__(self, base_session: types.IAsyncBaseSession):
         # RPC data structures
         self.call_requests: dict[int, Future[types.Result]] = {}
         self.register_requests: dict[int, RegisterRequest] = {}
-        self.registrations: dict[
-            int,
-            Union[Callable[[types.Invocation], types.Result], Callable[[types.Invocation], Awaitable[types.Result]]],
-        ] = {}
+        self.registrations: dict[int, Callable[[types.Invocation], Awaitable[types.Result]]] = {}
         self.unregister_requests: dict[int, types.UnregisterRequest] = {}
 
         # PubSub data structures
         self.publish_requests: dict[int, Future[None]] = {}
         self.subscribe_requests: dict[int, SubscribeRequest] = {}
         self.subscriptions: dict[int, Callable[[types.Event], Awaitable[None]]] = {}
         self.unsubscribe_requests: dict[int, types.UnsubscribeRequest] = {}
+        self.progress_handlers: dict[int, Callable[[types.Result], Awaitable[None]]] = {}
 
         self.goodbye_request = Future()
 
@@ -118,29 +116,68 @@ async def process_incoming_message(self, msg: messages.Message):
             del self.registrations[request.registration_id]
             request.future.set_result(None)
         elif isinstance(msg, messages.Result):
-            request = self.call_requests.pop(msg.request_id)
-            request.set_result(types.Result(msg.args, msg.kwargs, msg.options))
+            progress = msg.options.get("progress", False)
+            if progress:
+                progress_handler = self.progress_handlers.get(msg.request_id, None)
+                if progress_handler is not None:
+                    try:
+                        await progress_handler(types.Result(msg.args, msg.kwargs, msg.options))
+                    except Exception as e:
+                        # TODO: implement call canceling
+                        print(e)
+            else:
+                request = self.call_requests.pop(msg.request_id, None)
+                if request is not None:
+                    request.set_result(types.Result(msg.args, msg.kwargs, msg.options))
+                self.progress_handlers.pop(msg.request_id)
         elif isinstance(msg, messages.Invocation):
             try:
                 endpoint = self.registrations[msg.registration_id]
-                result = await endpoint(types.Invocation(msg.args, msg.kwargs, msg.details))
-
-                if result is None:
-                    data = self.session.send_message(messages.Yield(messages.YieldFields(msg.request_id)))
-                elif isinstance(result, types.Result):
-                    data = self.session.send_message(
-                        messages.Yield(messages.YieldFields(msg.request_id, result.args, result.kwargs, result.details))
-                    )
-                else:
-                    message = "Endpoint returned invalid result type. Expected types.Result or None, got: " + str(
-                        type(result)
-                    )
-                    msg_to_send = messages.Error(
-                        messages.ErrorFields(msg.TYPE, msg.request_id, xconn_uris.ERROR_INTERNAL_ERROR, [message])
-                    )
-                    data = self.session.send_message(msg_to_send)
-
-                await self.base_session.send(data)
+                invocation = types.Invocation(msg.args, msg.kwargs, msg.details)
+                receive_progress = msg.details.get("receive_progress", False)
+                if receive_progress:
+
+                    async def _progress_func(args: list[Any] | None, kwargs: dict[str, Any] | None):
+                        yield_msg = messages.Yield(
+                            messages.YieldFields(msg.request_id, args, kwargs, {"progress": True})
+                        )
+                        data = self.session.send_message(yield_msg)
+                        await self.base_session.send(data)
+
+                    invocation.send_progress = _progress_func
+
+                async def handle_endpoint_invocation():
+                    try:
+                        result = await endpoint(invocation)
+                        if result is None:
+                            data = self.session.send_message(messages.Yield(messages.YieldFields(msg.request_id)))
+                        elif isinstance(result, types.Result):
+                            data = self.session.send_message(
+                                messages.Yield(
+                                    messages.YieldFields(msg.request_id, result.args, result.kwargs, result.details)
+                                )
+                            )
+                        else:
+                            message = (
+                                "Endpoint returned invalid result type. Expected types.Result or None, got: "
+                                + str(type(result))
+                            )
+                            msg_to_send = messages.Error(
+                                messages.ErrorFields(
+                                    msg.TYPE, msg.request_id, xconn_uris.ERROR_INTERNAL_ERROR, [message]
+                                )
+                            )
+                            data = self.session.send_message(msg_to_send)
+                    except Exception as e:
+                        message = f"unexpected error calling endpoint {endpoint.__name__}, error is: {e}"
+                        msg_to_send = messages.Error(
+                            messages.ErrorFields(msg.TYPE, msg.request_id, xconn_uris.ERROR_INTERNAL_ERROR, [message])
+                        )
+                        data = self.session.send_message(msg_to_send)
+                    await self.base_session.send(data)
+
+                current_loop = get_event_loop()
+                current_loop.create_task(handle_endpoint_invocation())
             except ApplicationError as e:
                 msg_to_send = messages.Error(messages.ErrorFields(msg.TYPE, msg.request_id, e.message, e.args))
                 data = self.session.send_message(msg_to_send)
@@ -215,6 +252,15 @@ async def register(
 
         return await f
 
+    async def _call(self, call_msg: messages.Call) -> types.Result:
+        f = Future()
+        self.call_requests[call_msg.request_id] = f
+
+        data = self.session.send_message(call_msg)
+        await self.base_session.send(data)
+
+        return await f
+
     async def call(self, procedure: str, *args, **kwargs) -> types.Result:
         options = kwargs.pop("options", None)
         call = messages.Call(messages.CallFields(self.idgen.next(), procedure, args, kwargs, options=options))
@@ -227,6 +273,23 @@ async def call(self, procedure: str, *args, **kwargs) -> types.Result:
 
         return await f
 
+    async def call_progress(
+        self,
+        procedure: str,
+        progress_handler: Callable[[types.Result], Awaitable[None]],
+        args: list[Any] | None = None,
+        kwargs: dict[str, Any] | None = None,
+        options: dict[str, Any] | None = None,
+    ) -> types.Result:
+        if options is None:
+            options = {}
+
+        options["receive_progress"] = True
+        call_msg = messages.Call(messages.CallFields(self.idgen.next(), procedure, args, kwargs, options))
+        self.progress_handlers[call_msg.request_id] = progress_handler
+
+        return await self._call(call_msg)
+
     async def subscribe(
         self, topic: str, event_handler: Callable[[types.Event], Awaitable[None]], options: dict | None = None
     ) -> Subscription: