Merge pull request #57 from rsocket/correct_fragmentation_implementation

Correct fragmentation implementation

Author: jell-o-fishi

CHANGELOG.rst

@@ -1,7 +1,21 @@
 Changelog
 ---------
 
-v0.3.1
+v0.4.0
 ======
 
-- Added ability to await fire_and_forget and push_metadata. Waits until the client finishes sending the frame.
+- Breaking change: Added ability to await fire_and_forget and push_metadata:
+    - Both now return a future which resolves when the frame (or all fragments) finished sending.
+- Fixed fragmentation implementation (misunderstood spec):
+    - fragments after first one are now correctly of type PayloadFrame
+    - fragment size now includes frame header and length.
+    - Added checking fragment size limit (minimum 64) as in java implementation
+    - Updated examples
+
+v0.3.0
+======
+Initial mostly complete implementation after long time from previous release (0.2.0)
+
+v0.2.0
+======
+Legacy. Unknown history.

README.md

@@ -11,12 +11,14 @@ pip install rsocket
 ```
 
 or install any of the extras:
-  * rx
-  * aiohttp
-  * quart
-  * uic
+
+* rx
+* aiohttp
+* quart
+* uic
 
 Example:
+
 ```shell
 pip install --pre rsocket[rx]
 ```
@@ -47,17 +49,18 @@ The **examples/test_examples.py** shows which pairs of client/server work with e
 all the examples
 (except for the client_springboot.py which is set up to work against https://github.com/benwilcock/spring-rsocket-demo)
 
-| server (python)             | server (java) | client (python)                    | client(java)    |
-|-----------------------------|---------------|------------------------------------|-----------------|
-| server.py                   |               | client.py                          |                 |
-| server_quic.py              |               | client_quic.py                     |                 |
-| server_with_lease.py        |               |                                    | ClientWithLease |
-| server_with_routing.py      |               | client_with_routing.py             | Client          |
-| server_with_routing.py      |               | client_rx.py                       |                 |
-| server_with_routing.py      |               | client_reconnect.py                |                 |
-|                             | Server        | run_against_example_java_server.py |                 |
-| server_quart_websocket.py   |               | client_websocket.py                |                 |
-| server_aiohttp_websocket.py |               | client_websocket.py                |                 |
+| server (python)             | server (java)           | client (python)                    | client(java)    |
+|-----------------------------|-------------------------|------------------------------------|-----------------|
+| server.py                   |                         | client.py                          |                 |
+| server_quic.py              |                         | client_quic.py                     |                 |
+| server_with_lease.py        |                         |                                    | ClientWithLease |
+| server_with_routing.py      |                         | client_with_routing.py             | Client          |
+| server_with_routing.py      |                         | client_rx.py                       |                 |
+| server_with_routing.py      |                         | client_reconnect.py                |                 |
+|                             | Server                  | run_against_example_java_server.py |                 |
+|                             | ServerWithFragmentation | client_with_routing.py             |                 |
+| server_quart_websocket.py   |                         | client_websocket.py                |                 |
+| server_aiohttp_websocket.py |                         | client_websocket.py                |                 |
 
 # Build Status
 

examples/client.py

@@ -2,9 +2,9 @@
 import logging
 import sys
 
+from examples.shared_tests import simple_client_server_test
 from reactivestreams.subscriber import DefaultSubscriber
 from rsocket.helpers import single_transport_provider
-from rsocket.payload import Payload
 from rsocket.rsocket_client import RSocketClient
 from rsocket.transports.tcp import TransportTCP
 
@@ -22,22 +22,7 @@ async def main(server_port):
     connection = await asyncio.open_connection('localhost', server_port)
 
     async with RSocketClient(single_transport_provider(TransportTCP(*connection))) as client:
-        payload = Payload(b'%Y-%m-%d %H:%M:%S')
-
-        async def run_request_response():
-            try:
-                while True:
-                    result = await client.request_response(payload)
-                    logging.info('Response: {}'.format(result.data))
-                    await asyncio.sleep(1)
-            except asyncio.CancelledError:
-                pass
-
-        task = asyncio.create_task(run_request_response())
-
-        await asyncio.sleep(5)
-        task.cancel()
-        await task
+        await simple_client_server_test(client)
 
 
 if __name__ == '__main__':

examples/client_quic.py

@@ -5,8 +5,8 @@
 
 from aioquic.quic.configuration import QuicConfiguration
 
+from examples.shared_tests import simple_client_server_test
 from rsocket.helpers import single_transport_provider
-from rsocket.payload import Payload
 from rsocket.rsocket_client import RSocketClient
 from rsocket.transports.aioquic_transport import rsocket_connect
 
@@ -23,22 +23,7 @@ async def main(server_port):
     async with rsocket_connect('localhost', server_port,
                                configuration=client_configuration) as transport:
         async with RSocketClient(single_transport_provider(transport)) as client:
-            payload = Payload(b'%Y-%m-%d %H:%M:%S')
-
-            async def run_request_response():
-                try:
-                    while True:
-                        result = await client.request_response(payload)
-                        logging.info('Response: {}'.format(result.data))
-                        await asyncio.sleep(1)
-                except asyncio.CancelledError:
-                    pass
-
-            task = asyncio.create_task(run_request_response())
-
-            await asyncio.sleep(5)
-            task.cancel()
-            await task
+            await simple_client_server_test(client)
 
 
 if __name__ == '__main__':

examples/client_rx.py

@@ -5,12 +5,13 @@
 from typing import AsyncGenerator, Tuple
 
 from rx import operators
+from rx.core.operators.map import _map
 
+from examples.shared_tests import assert_result_data
 from reactivestreams.subscriber import Subscriber
 from reactivestreams.subscription import Subscription
 from rsocket.extensions.helpers import route, composite, authenticate_simple, metadata_item
 from rsocket.extensions.mimetypes import WellKnownMimeTypes
-from rsocket.fragment import Fragment
 from rsocket.helpers import single_transport_provider
 from rsocket.payload import Payload
 from rsocket.rsocket_client import RSocketClient
@@ -21,13 +22,13 @@
 
 def sample_publisher(wait_for_requester_complete: Event,
                      response_count: int = 3):
-    async def generator() -> AsyncGenerator[Tuple[Fragment, bool], None]:
+    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 Fragment(message.encode('utf-8')), is_complete
+            yield Payload(message.encode('utf-8')), is_complete
 
             if is_complete:
                 wait_for_requester_complete.set()
@@ -96,7 +97,9 @@ async def request_response(client: RxRSocket):
         authenticate_simple('user', '12345')
     ))
 
-    await client.request_response(payload).pipe()
+    result = await client.request_response(payload).pipe()
+
+    assert_result_data(result, b'single_response')
 
 
 async def request_last_metadata(client: RxRSocket):
@@ -107,7 +110,7 @@ async def request_last_metadata(client: RxRSocket):
 
     result = await client.request_response(payload).pipe()
 
-    assert result.data == b'audit info'
+    assert_result_data(result, b'audit info')
 
 
 async def request_last_fnf(client: RxRSocket):
@@ -118,11 +121,10 @@ async def request_last_fnf(client: RxRSocket):
 
     result = await client.request_response(payload).pipe()
 
-    assert result.data == b'aux data'
+    assert_result_data(result, b'aux data')
 
 
 async def metadata_push(client: RxRSocket, metadata: bytes):
-
     await client.metadata_push(composite(
         route('metadata_push'),
         authenticate_simple('user', '12345'),
@@ -173,8 +175,12 @@ async def request_stream(client: RxRSocket):
         route('stream'),
         authenticate_simple('user', '12345')
     ))
-    result = await client.request_stream(payload).pipe(operators.to_list())
-    print(result)
+    result = await client.request_stream(payload).pipe(_map(lambda p: p.data), operators.to_list())
+
+    if result != [b'Item on channel: 0',
+                  b'Item on channel: 1',
+                  b'Item on channel: 2']:
+        raise Exception(result)
 
 
 async def request_slow_stream(client: RxRSocket):
@@ -186,15 +192,6 @@ async def request_slow_stream(client: RxRSocket):
     print(result)
 
 
-async def request_fragmented_stream(client: RxRSocket):
-    payload = Payload(b'The quick brown fox', composite(
-        route('fragmented_stream'),
-        authenticate_simple('user', '12345')
-    ))
-    result = await client.request_stream(payload).pipe(operators.to_list())
-    print(result)
-
-
 async def main(server_port):
     logging.info('Connecting to server at localhost:%s', server_port)
 
@@ -208,7 +205,6 @@ async def main(server_port):
         await request_slow_stream(rx_client)
         await request_channel(rx_client)
         await request_stream_invalid_login(rx_client)
-        await request_fragmented_stream(rx_client)
 
         await metadata_push(rx_client, b'audit info')
         await request_last_metadata(rx_client)