Merge branch 'master' into master

Author: jell-o-fishi

CHANGELOG.rst

@@ -1,6 +1,10 @@
 Changelog
 ---------
 
+v0.4.12
+=======
+- Fixed fragmentation for fire and forget
+
 v0.4.11
 =======
 - Breaking change: RequestRouter argument 'payload_mapper' was replaced with 'payload_deserializer' and 'payload_serializer'

bugs/165_client.py

@@ -0,0 +1,30 @@
+import asyncio
+import logging
+from datetime import timedelta
+
+from rsocket.helpers import single_transport_provider
+from rsocket.payload import Payload
+from rsocket.rsocket_client import RSocketClient
+from rsocket.transports.tcp import TransportTCP
+
+
+async def main(server_port):
+    logging.info("Connecting to server at localhost:%s", server_port)
+
+    connection = await asyncio.open_connection("localhost", server_port)
+
+    async with RSocketClient(
+            single_transport_provider(TransportTCP(*connection)),
+            fragment_size_bytes=10240,
+            keep_alive_period=timedelta(seconds=10),
+    ) as client:
+        # huge_array = bytearray(16777209) # Works
+        huge_array = bytearray(16777210)  # rsocket.exceptions.ParseError: Frame too short: 0 bytes
+        payload = Payload(huge_array)
+        await client.fire_and_forget(payload)
+        await asyncio.sleep(1)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
+    asyncio.run(main(10000))

bugs/165_server.py

@@ -0,0 +1,32 @@
+import asyncio
+import logging
+
+from rsocket.helpers import create_future
+from rsocket.local_typing import Awaitable
+from rsocket.payload import Payload
+from rsocket.request_handler import BaseRequestHandler
+from rsocket.rsocket_server import RSocketServer
+from rsocket.transports.tcp import TransportTCP
+
+
+class Handler(BaseRequestHandler):
+    async def request_fire_and_forget(self, payload: Payload) -> Awaitable[Payload]:
+        print(f"Receiving {len(payload.data)} bytes")
+        return create_future(Payload(b"OK"))
+
+
+async def run_server(server_port):
+    logging.info("Starting server at localhost:%s", server_port)
+
+    def session(*connection):
+        RSocketServer(TransportTCP(*connection), handler_factory=Handler, fragment_size_bytes=10240)
+
+    server = await asyncio.start_server(session, "localhost", server_port)
+
+    async with server:
+        await server.serve_forever()
+
+
+if __name__ == "__main__":
+    logging.basicConfig(filename="rsocket.log", format="%(asctime)s %(message)s", filemode="w")
+    asyncio.run(run_server(10000))

bugs/__init__.py

@@ -24,7 +24,7 @@
 author = '[email protected]'
 
 # The full version, including alpha/beta/rc tags
-release = '0.4.11'
+release = '0.4.12'
 
 # -- General configuration ---------------------------------------------------
 

docs/conf.py

@@ -1,22 +1,22 @@
 Rx==3.2.0
 aiohttp==3.8.4
-aioquic==0.9.20
+aioquic==0.9.21
 asyncclick==8.1.3.4
 asyncstdlib==3.10.8
 coverage==6.5.0
 coveralls==3.3.1
 decoy==2.0.2
 flake8==5.0.4
-pytest-asyncio==0.21.0
+pytest-asyncio==0.21.1
 pytest-cov==4.1.0
 pytest-profiling==1.7.0
-pytest-rerunfailures==11.1.2
+pytest-rerunfailures==12.0
 pytest-timeout==2.1.0
 pytest-xdist==3.3.1
 pytest==7.4.0
 quart==0.18.4
 reactivex==4.0.4
-starlette==0.28.0
+starlette==0.30.0
 cbitstruct==1.0.9
 cloudevents==1.9.0
-pydantic==1.10.9
+pydantic==1.10.11

requirements.txt

@@ -1 +1 @@
-__version__ = '0.4.11'
+__version__ = '0.4.12'

rsocket/__init__.py

@@ -487,7 +487,7 @@ def fire_and_forget(self, payload: Payload) -> Awaitable[None]:
         logger().debug('%s: fire-and-forget: %s', self._log_identifier(), payload)
 
         stream_id = self._allocate_stream()
-        frame = to_fire_and_forget_frame(stream_id, payload)
+        frame = to_fire_and_forget_frame(stream_id, payload, self._fragment_size_bytes)
         self.send_request(frame)
         frame.sent_future.add_done_callback(lambda _: self.finish_stream(stream_id))
         return frame.sent_future

rsocket/rsocket_base.py

@@ -36,6 +36,30 @@ def handler_factory():
     await asyncio.sleep(2)  # wait for server to close
 
 
+async def test_request_fire_and_forget_fragmented(lazy_pipe):
+    handler: Optional[FireAndForgetHandler] = None
+
+    def handler_factory():
+        nonlocal handler
+        handler = FireAndForgetHandler()
+        return handler
+
+    async with lazy_pipe(
+            server_arguments={'handler_factory': handler_factory,
+                              'fragment_size_bytes': 10240},
+            client_arguments={'fragment_size_bytes': 10240}) as (server, client):
+        data = bytearray(16777210)
+
+        await client.fire_and_forget(Payload(data))
+
+        await handler.received.wait()
+
+        assert handler.received_payload.data == data
+        assert handler.received_payload.metadata == b''
+
+    await asyncio.sleep(2)  # wait for server to close
+
+
 async def test_request_fire_and_forget_awaitable_client(lazy_pipe):
     handler: Optional[FireAndForgetHandler] = None