added specific demo examples

Author: jell-o-fishi

examples/cli_demo_server/server.py

@@ -0,0 +1,65 @@
+import asyncio
+import logging
+
+import asyncclick as click
+import reactivex
+from aiohttp import web
+from reactivex import Observable
+
+from rsocket.payload import Payload
+from rsocket.routing.request_router import RequestRouter
+from rsocket.routing.routing_request_handler import RoutingRequestHandler
+from rsocket.rsocket_server import RSocketServer
+from rsocket.transports.aiohttp_websocket import TransportAioHttpWebsocket
+from rsocket.transports.tcp import TransportTCP
+
+router = RequestRouter()
+
+
+@router.response('echo')
+async def echo(payload: Payload) -> Observable:
+    return reactivex.just(Payload(payload.data))
+
+
+def websocket_handler_factory(**kwargs):
+    async def websocket_handler(request):
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        transport = TransportAioHttpWebsocket(ws)
+        RSocketServer(transport, **kwargs)
+        await transport.handle_incoming_ws_messages()
+        return ws
+
+    return websocket_handler
+
+
+@click.command()
+@click.option('--port', help='Port to listen on', default=6565, type=int)
+@click.option('--transport', is_flag=False, default='tcp')
+async def start_server(port: int, transport: str):
+    logging.basicConfig(level=logging.DEBUG)
+
+    logging.info(f'Starting {transport} server at localhost:{port}')
+
+    if transport in ['ws']:
+        app = web.Application()
+        app.add_routes([web.get('/', websocket_handler_factory(
+            handler_factory=lambda: RoutingRequestHandler(router)
+        ))])
+
+        await web._run_app(app, port=port)
+    elif transport == 'tcp':
+        def handle_client(reader, writer):
+            RSocketServer(TransportTCP(reader, writer),
+                          handler_factory=lambda: RoutingRequestHandler(router))
+
+        server = await asyncio.start_server(handle_client, 'localhost', port)
+
+        async with server:
+            await server.serve_forever()
+    else:
+        raise Exception(f'Unsupported transport {transport}')
+
+
+if __name__ == '__main__':
+    start_server()

examples/issue_290/client.py

@@ -0,0 +1,98 @@
+import asyncio
+import logging
+import sys
+from asyncio import Event
+from typing import AsyncGenerator, Tuple
+
+import aiohttp
+from rsocket.transports.asyncwebsockets_transport import websocket_client
+
+from rsocket.transports.aiohttp_websocket import TransportAioHttpClient
+
+from reactivestreams.publisher import Publisher
+from reactivestreams.subscriber import Subscriber
+from reactivestreams.subscription import Subscription
+from rsocket.extensions.helpers import route, composite, authenticate_simple
+from rsocket.extensions.mimetypes import WellKnownMimeTypes
+from rsocket.helpers import single_transport_provider
+from rsocket.payload import Payload
+from rsocket.rsocket_client import RSocketClient
+from rsocket.streams.stream_from_async_generator import StreamFromAsyncGenerator
+from rsocket.transports.tcp import TransportTCP
+
+
+def sample_publisher(wait_for_requester_complete: Event,
+                     response_count: int = 1000) -> Publisher:
+    async def generator() -> AsyncGenerator[Tuple[Payload, bool], None]:
+        current_response = 0
+        for i in range(response_count):
+            is_complete = (current_response + 1) == response_count
+
+            message = 'Item to server from client on channel: %s' % current_response
+            yield Payload(message.encode('utf-8')), is_complete
+
+            if is_complete:
+                wait_for_requester_complete.set()
+                break
+
+            current_response += 1
+
+    return StreamFromAsyncGenerator(generator)
+
+
+class ChannelSubscriber(Subscriber):
+
+    def __init__(self, wait_for_responder_complete: Event) -> None:
+        super().__init__()
+        self._wait_for_responder_complete = wait_for_responder_complete
+        self.values = []
+
+    def on_subscribe(self, subscription: Subscription):
+        self.subscription = subscription
+
+    def on_next(self, value: Payload, is_complete=False):
+        logging.info('From server on channel: ' + value.data.decode('utf-8'))
+        self.values.append(value.data)
+        if is_complete:
+            self._wait_for_responder_complete.set()
+
+    def on_error(self, exception: Exception):
+        logging.error('Error from server on channel' + str(exception))
+        self._wait_for_responder_complete.set()
+
+    def on_complete(self):
+        logging.info('Completed from server on channel')
+        self._wait_for_responder_complete.set()
+
+async def request_channel(client: RSocketClient):
+
+    channel_completion_event = Event()
+    requester_completion_event = Event()
+    payload = Payload(b'The first item in the stream', composite(
+        route('channel'),
+        authenticate_simple('user', '12345')
+    ))
+    publisher = sample_publisher(requester_completion_event)
+
+    requested = client.request_channel(payload, publisher)
+
+    subscriber = ChannelSubscriber(channel_completion_event)
+    requested.initial_request_n(5).subscribe(subscriber)
+
+    await channel_completion_event.wait()
+    await requester_completion_event.wait()
+
+
+async def application(serve_port: int):
+        async with websocket_client('http://localhost:%s/rsocket' % serve_port,
+                                    metadata_encoding=WellKnownMimeTypes.MESSAGE_RSOCKET_COMPOSITE_METADATA,) as client:
+            await request_channel(client)
+
+
+async def command():
+    logging.basicConfig(level=logging.DEBUG)
+    await application(7878)
+
+
+if __name__ == '__main__':
+    asyncio.run(command())

examples/issue_290/server.py

@@ -0,0 +1,109 @@
+import asyncio
+import logging
+
+from aiohttp import web
+
+from reactivestreams.subscriber import DefaultSubscriber
+from reactivestreams.subscription import Subscription
+from rsocket.helpers import DefaultPublisherSubscription
+from rsocket.payload import Payload
+from rsocket.routing.request_router import RequestRouter
+from rsocket.routing.routing_request_handler import RoutingRequestHandler
+from rsocket.rsocket_server import RSocketServer
+from rsocket.transports.aiohttp_websocket import TransportAioHttpWebsocket
+
+
+class ChannelPublisher(DefaultPublisherSubscription):
+    def __init__(self):
+        super().__init__()
+        self.subscriber = None
+
+    def publish(self, data):
+        print(f"ChannelPublisher - publish: {data}")
+        print(f"ChannelPublisher - publish subscriber: {self.subscriber}")
+
+        if (self.subscriber is not None):
+            self.subscriber.on_next(Payload(data.encode('utf-8')))
+
+    def subscribe(self, subscriber):
+        print("ChannelPublisher - subscribe")
+
+        self.subscriber = subscriber
+
+        subscriber.on_subscribe(self)
+
+
+class ChannelSubscriber(DefaultSubscriber):
+
+    def __init__(self, publisher: ChannelPublisher):
+        self.subscription = None
+        self.publisher = publisher
+
+    def on_subscribe(self, subscription: Subscription):
+        print("ChannelSubscriber - on_subscribe")
+        subscription.request(3)
+
+    def on_next(self, value: Payload, is_complete=False):
+        user_message = value.data.decode('utf-8')
+        print(f"ChannelSubscriber - on_next: {user_message}")
+
+        self.publisher.publish(f"Some text and message: {user_message}")
+
+
+def handler_factory_factory():
+    router = RequestRouter()
+
+    @router.channel('channel')
+    async def channel_response(payload: Payload, composite_metadata):
+        print('Got channel request')
+
+        publisher = ChannelPublisher()
+        subscriber = ChannelSubscriber(publisher)
+
+        # WHY?: if passing received payload to ChannelSubscriber here can be processed before at ChannelPublisher "subscribe" method got subscriber
+        # in other words - we do "one_next()", it will be processed with some logic and result can be published with "publish()"
+        # subscriber.on_next(payload)
+
+        return publisher, subscriber
+
+    def handler_factory():
+        return RoutingRequestHandler(router)
+
+    return handler_factory
+
+
+def websocket_handler_factory(**kwargs):
+    async def websocket_handler(request):
+        ws = web.WebSocketResponse()
+        await ws.prepare(request)
+        transport = TransportAioHttpWebsocket(ws)
+        RSocketServer(transport, **kwargs)
+        await transport.handle_incoming_ws_messages()
+        return ws
+
+    return websocket_handler
+
+
+async def start_server():
+    logging.basicConfig(level=logging.DEBUG)
+
+    print(f'Starting server at localhost: 7878')
+
+    app = web.Application()
+    app.add_routes(
+        [
+            web.get(
+                '/rsocket',
+                websocket_handler_factory(
+                    handler_factory=handler_factory_factory()
+                )
+            ),
+            # web.static('/static', 'static'),
+        ]
+    )
+
+    await web._run_app(app, port=7878, ssl_context=None)
+
+
+if __name__ == '__main__':
+    asyncio.run(start_server())