revert changes and update examples

Author: jell-o-fishi

examples/server_with_routing.py

@@ -72,10 +72,11 @@ async def metadata_push(payload: Payload, composite_metadata: CompositeMetadata)
             if item.encoding == b'text/plain':
                 storage.last_metadata_push = item.content
 
-    @router.channel('channel', send_request_payload_to_subscriber=True)
-    async def channel_response(payload, composite_metadata):
+    @router.channel('channel')
+    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()

rsocket/routing/request_router.py

@@ -17,21 +17,20 @@
 
 
 class RouteInfo:
-    def __init__(self, method, send_channel_request_payload_to_subscriber=False):
+    def __init__(self, method):
         self.method: Callable = method
         self.signature = signature(method)
-        self.send_channel_request_payload_to_subscriber = send_channel_request_payload_to_subscriber
 
 
-def decorator_factory(container: dict, route: str, send_channel_request_payload_to_subscriber: bool = False):
+def decorator_factory(container: dict, route: str):
     def decorator(function: decorated_method):
         if safe_len(route) == 0:
             raise RSocketEmptyRoute(function.__name__)
 
         if route in container:
             raise KeyError('Duplicate route "%s" already registered', route)
 
-        container[route] = RouteInfo(function, send_channel_request_payload_to_subscriber)
+        container[route] = RouteInfo(function)
         return function
 
     return decorator
@@ -107,9 +106,8 @@ def wrapper(function):
 
         return wrapper
 
-    def channel(self, route: str, send_request_payload_to_subscriber: bool = False):
-        return decorator_factory(self._channel_routes, route,
-                                 send_channel_request_payload_to_subscriber=send_request_payload_to_subscriber)
+    def channel(self, route: str):
+        return decorator_factory(self._channel_routes, route)
 
     def channel_unknown(self):
         def wrapper(function):
@@ -164,11 +162,6 @@ async def route(self,
 
             return create_future(result)
 
-        elif frame_type == FrameType.REQUEST_CHANNEL:
-            publisher, subscriber = result
-            if route_info.send_channel_request_payload_to_subscriber:
-                subscriber.on_next(payload)
-
         return result
 
     def _collect_route_arguments(self,