added streaming file example from discord

Author: jell-o-fishi

examples/bugs/__init__.py

@@ -0,0 +1,46 @@
+import asyncio
+import logging
+from _codecs import utf_8_decode
+from typing import List
+
+from reactivex import operators
+
+from rsocket.extensions.helpers import composite, route
+from rsocket.extensions.mimetypes import WellKnownMimeTypes
+from rsocket.helpers import single_transport_provider
+from rsocket.payload import Payload
+from rsocket.reactivex.reactivex_client import ReactiveXClient
+from rsocket.rsocket_client import RSocketClient
+from rsocket.transports.tcp import TransportTCP
+
+
+def got_data(payload):
+    logging.info('From server: ' + payload.data.decode('utf-8'))
+    return utf_8_decode(payload.data)
+
+
+class Client:
+
+    def __init__(self, rs: RSocketClient):
+        self._rs = rs
+
+    async def stream_file(self) -> List[str]:
+        request = Payload(metadata=composite(route('audio')))
+        return await ReactiveXClient(self._rs).request_stream(request).pipe(
+            operators.map(got_data),
+            operators.to_list()
+        )
+
+
+async def main():
+    connection = await asyncio.open_connection('localhost', 8000)
+    async with RSocketClient(single_transport_provider(TransportTCP(*connection)),
+                             metadata_encoding=WellKnownMimeTypes.MESSAGE_RSOCKET_COMPOSITE_METADATA,
+                             fragment_size_bytes=1_000_000) as client:
+        c = Client(client)
+        r = await c.stream_file()
+        print(r)
+
+if __name__ == '__main__':
+    logging.basicConfig(level=logging.DEBUG)
+    asyncio.run(main())

examples/bugs/streaming_file/__init__.py

@@ -0,0 +1,26 @@
+gsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfggsdfgsdfg
+sdfgsdfgsdfgvvv

examples/bugs/streaming_file/client.py

@@ -0,0 +1,64 @@
+import asyncio
+import base64
+import logging
+from typing import Callable
+
+import aiofiles
+from reactivex import Subject, Observable, operators
+
+from rsocket.payload import Payload
+from rsocket.reactivex.back_pressure_publisher import from_observable_with_backpressure, observable_from_queue
+from rsocket.reactivex.reactivex_handler_adapter import reactivex_handler_factory
+from rsocket.routing.request_router import RequestRouter
+from rsocket.routing.routing_request_handler import RoutingRequestHandler
+from rsocket.rsocket_server import RSocketServer
+from rsocket.transports.tcp import TransportTCP
+
+
+async def read_file(o: asyncio.Queue):
+    async with aiofiles.open("my_file", 'rb') as f:
+        while True:
+            buffer = await f.read(1024)
+            logging.info("Reading buffer")
+            if buffer == b'':
+                o.put_nowait(None)
+                break
+            await o.put(base64.b64encode(buffer))
+
+
+def handler_factory() -> RoutingRequestHandler:
+    router = RequestRouter()
+
+    @router.stream('audio')
+    async def audio() -> Callable[[Subject], Observable]:
+        q = asyncio.Queue()
+        # await q.put(b'sf') # testing whether a simpler queuing scheme fixes the issue
+        asyncio.create_task(read_file(q))
+        return from_observable_with_backpressure(
+            lambda backpressure: observable_from_queue(
+                q, backpressure=backpressure).pipe(
+                operators.map(lambda buffer: Payload(buffer))
+            )
+        )
+
+    return RoutingRequestHandler(router)
+
+
+async def run_server():
+    def session(*connection):
+        RSocketServer(TransportTCP(*connection),
+                      handler_factory=reactivex_handler_factory(handler_factory),
+                      fragment_size_bytes=1_000_000)
+
+    async with await asyncio.start_server(session, 'localhost', 8000) as server:
+        await server.serve_forever()
+
+
+def main():
+    logging.basicConfig(level=logging.DEBUG)
+    logging.info("Starting server...")
+    asyncio.run(run_server())
+
+
+if __name__ == '__main__':
+    main()