Merge pull request #291 from rsocket/send_channel_request_to_subscriber

Updated example use cases. Fixed a reactivex request_response error

Author: jell-o-fishi

examples/server_with_routing.py

@@ -73,9 +73,10 @@ async def metadata_push(payload: Payload, composite_metadata: CompositeMetadata)
                 storage.last_metadata_push = item.content
 
     @router.channel('channel')
-    async def channel_response(payload, composite_metadata):
+    async def channel_response(payload:Payload, composite_metadata):
         logging.info('Got channel request')
         subscriber = LoggingSubscriber()
+        subscriber.on_next(payload)
         channel = sample_async_response_stream(local_subscriber=subscriber)
         return channel, subscriber
 

examples/tutorial/reactivex/chat_server.py

@@ -180,7 +180,7 @@ async def receive_statistics(statistics: ClientStatistics):
             self._session.statistics = statistics
 
         @router.channel('statistics')
-        async def send_statistics() -> ReactivexChannel:
+        async def send_statistics(initial_payload: Payload) -> ReactivexChannel:
 
             async def statistics_generator():
                 while True:
@@ -201,6 +201,8 @@ def on_next(payload: Payload):
                 if request.period_seconds is not None:
                     self._session.requested_statistics.period_seconds = request.period_seconds
 
+            on_next(initial_payload)
+
             return ReactivexChannel(
                 from_observable_with_backpressure(
                     lambda backpressure: observable_from_async_generator(

rsocket/reactivex/reactivex_handler_adapter.py

@@ -48,7 +48,13 @@ async def request_fire_and_forget(self, payload: Payload):
         await self.delegate.request_fire_and_forget(payload)
 
     async def request_response(self, payload: Payload) -> asyncio.Future:
-        observable = await self.delegate.request_response(payload)
+        response = await self.delegate.request_response(payload)
+
+        if isinstance(response, asyncio.Future):
+            observable = await response
+        else:
+            observable = response
+
         return observable.pipe(
             operators.default_if_empty(Payload()),
             operators.to_future()

tests/test_reactivex/test_reactivex_server_side.py

@@ -19,6 +19,7 @@ class Handler(BaseReactivexHandler):
 
     def __init__(self, server_done: Optional[asyncio.Event] = None):
         self._server_done = server_done
+        self.received_payloads = []
 
     async def request_stream(self, payload: Payload) -> Observable:
         return reactivex.from_iterable((Payload(ensure_bytes('Feed Item: {}'.format(index))) for index in range(3)))
@@ -29,6 +30,9 @@ async def request_channel(self, payload: Payload) -> ReactivexChannel:
 
         def observer(value: Payload):
             logging.info(f'Received by test server: {value.data}')
+            self.received_payloads.append(value.data)
+
+        observer(payload)
 
         return ReactivexChannel(observable,
                                 Observer(observer,
@@ -62,9 +66,12 @@ async def test_serve_reactivex_channel(pipe: Tuple[RSocketServer, RSocketClient]
     server, client = pipe
 
     server_done_event = asyncio.Event()
+    handler = None
 
     def handler_factory():
-        return Handler(server_done_event)
+        nonlocal handler
+        handler = Handler(server_done_event)
+        return handler
 
     server.set_handler_using_factory(reactivex_handler_factory(handler_factory))
 
@@ -84,6 +91,12 @@ def handler_factory():
 
     await server_done_event.wait()
 
+    assert len(handler.received_payloads) == 4
+    assert handler.received_payloads[0] == b'request text'
+    assert handler.received_payloads[1] == b'Client item: 0'
+    assert handler.received_payloads[2] == b'Client item: 1'
+    assert handler.received_payloads[3] == b'Client item: 2'
+
 
 async def test_serve_reactivex_response(pipe: Tuple[RSocketServer, RSocketClient]):
     server, client = pipe