Merge pull request #120 from rsocket/refactoring

refactoring

Author: jell-o-fishi

README.md

@@ -15,14 +15,15 @@ pip install rsocket
 
 You may also install using some **extras**:
 
-| Extra      | Functionality                       |
-|------------|-------------------------------------|
-| rx         | ReactiveX (v3) integration          |
-| reactivex  | ReactiveX (v4) integration          |
-| aiohttp    | Websocket transport (server/client) |
-| quart      | Websocket transport (server only)   |
-| quic       | QUIC transport                      |
-| cli        | Command line                        |
+| Extra     | Functionality                       |
+|-----------|-------------------------------------|
+| rx        | ReactiveX (v3) integration          |
+| reactivex | ReactiveX (v4) integration          |
+| aiohttp   | Websocket transport (server/client) |
+| quart     | Websocket transport (server only)   |
+| quic      | QUIC transport                      |
+| cli       | Command line                        |
+| optimized | Frame parse/serialize optimizations |
 
 ```shell
 pip install rsocket[reactivex]

docs/conf.py

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

performance/performance_client.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import sys
 from asyncio import Event
 from typing import AsyncGenerator, Tuple
 
@@ -14,7 +15,7 @@
 from rsocket.streams.stream_from_async_generator import StreamFromAsyncGenerator
 from rsocket.transports.tcp import TransportTCP
 from tests.rsocket.helpers import to_json_bytes, create_large_random_data
-
+from tests.tools.helpers import measure_time
 
 data_size = 1920 # * 1080 * 3
 large_data = create_large_random_data(data_size)
@@ -104,7 +105,7 @@ async def request_fragmented_stream(self):
     async def __aenter__(self):
         logging.info('Connecting to server at localhost:%s', self._server_port)
 
-        connection = await asyncio.open_connection('localhost', self._server_port)
+        connection = await asyncio.open_connection('localhost', self._server_port, limit=data_size + 3000)
 
         self._client = AwaitableRSocket(RSocketClient(
             single_transport_provider(TransportTCP(*connection, read_buffer_size=data_size + 3000)),
@@ -116,3 +117,16 @@ async def __aenter__(self):
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         await self._client.__aexit__(exc_type, exc_val, exc_tb)
+
+
+async def run_client():
+    async with PerformanceClient(6565) as client:
+        for i in range(10000):
+            result = await measure_time(client.large_request())
+            # print(result.delta)
+
+
+if __name__ == '__main__':
+    port = sys.argv[1] if len(sys.argv) > 1 else 6565
+    logging.basicConfig(level=logging.ERROR)
+    asyncio.run(run_client())

performance/performance_server.py

@@ -17,7 +17,7 @@
 from performance.sample_responses import response_stream_2, response_stream_1, LoggingSubscriber
 from tests.rsocket.helpers import create_large_random_data
 
-data_size = 1920 # * 1080 * 3
+data_size = 1920 * 1080 * 3
 large_data = create_large_random_data(data_size)
 
 router = RequestRouter()
@@ -124,7 +124,7 @@ def handle_client(reader, writer):
 async def run_server(server_port, on_ready=None):
     logging.info('Starting server at localhost:%s', server_port)
 
-    server = await asyncio.start_server(client_handler_factory(on_ready), 'localhost', server_port)
+    server = await asyncio.start_server(client_handler_factory(on_ready), 'localhost', server_port, limit=data_size)
 
     async with server:
         await server.serve_forever()

requirements.txt

@@ -16,4 +16,5 @@ pytest-xdist==3.2.0
 pytest==7.2.1
 quart==0.18.3
 reactivex==4.0.4
-starlette==0.25.0
+starlette==0.25.0
+cbitstruct==1.0.9

rsocket/extensions/authentication_content.py

@@ -15,9 +15,7 @@ def __init__(self, authentication: Optional[Authentication] = None):
         self.authentication = authentication
 
     def serialize(self) -> bytes:
-        serialized = b''
-
-        serialized += serialize_well_known_encoding(self.authentication.type, WellKnownAuthenticationTypes.get_by_name)
+        serialized = serialize_well_known_encoding(self.authentication.type, WellKnownAuthenticationTypes.get_by_name)
 
         serialized += self.authentication.serialize()
 
@@ -29,10 +27,11 @@ def parse(self, buffer: bytes):
         self.authentication.parse(buffer[offset:])
 
 
-def authentication_item_factory(metadata_encoding: bytes) -> Type[Authentication]:
-    metadata_item_factory_by_type = {
-        WellKnownAuthenticationTypes.SIMPLE.value.name: AuthenticationSimple,
-        WellKnownAuthenticationTypes.BEARER.value.name: AuthenticationBearer,
-    }
+metadata_item_factory_by_type = {
+    WellKnownAuthenticationTypes.SIMPLE.value.name: AuthenticationSimple,
+    WellKnownAuthenticationTypes.BEARER.value.name: AuthenticationBearer,
+}
 
+
+def authentication_item_factory(metadata_encoding: bytes) -> Type[Authentication]:
     return metadata_item_factory_by_type[metadata_encoding]

rsocket/extensions/authentication_types.py

@@ -2,7 +2,8 @@
 from typing import Optional
 
 from rsocket.exceptions import RSocketUnknownAuthType
-from rsocket.helpers import WellKnownType, map_types_by_id, map_types_by_name
+from rsocket.helpers import WellKnownType, map_type_names_by_id, \
+    map_type_ids_by_name
 
 
 class WellKnownAuthenticationType(WellKnownType):
@@ -26,5 +27,5 @@ def get_by_name(cls, metadata_name: str) -> Optional[WellKnownAuthenticationType
         return type_by_name.get(metadata_name)
 
 
-type_by_id = map_types_by_id(WellKnownAuthenticationTypes)
-type_by_name = map_types_by_name(WellKnownAuthenticationTypes)
+type_by_id = map_type_names_by_id(WellKnownAuthenticationTypes)
+type_by_name = map_type_ids_by_name(WellKnownAuthenticationTypes)

rsocket/extensions/composite_metadata.py

@@ -18,14 +18,15 @@ def default_or_value(value, default=None):
     return value
 
 
-def metadata_item_factory(metadata_encoding: bytes) -> Type[CompositeMetadataItem]:
-    metadata_item_factory_by_type = {
-        WellKnownMimeTypes.MESSAGE_RSOCKET_ROUTING.value.name: RoutingMetadata,
-        WellKnownMimeTypes.MESSAGE_RSOCKET_MIMETYPE.value.name: StreamDataMimetype,
-        WellKnownMimeTypes.MESSAGE_RSOCKET_ACCEPT_MIMETYPES.value.name: StreamDataMimetypes,
-        WellKnownMimeTypes.MESSAGE_RSOCKET_AUTHENTICATION.value.name: AuthenticationContent
-    }
+metadata_item_factory_by_type = {
+    WellKnownMimeTypes.MESSAGE_RSOCKET_ROUTING.value.name: RoutingMetadata,
+    WellKnownMimeTypes.MESSAGE_RSOCKET_MIMETYPE.value.name: StreamDataMimetype,
+    WellKnownMimeTypes.MESSAGE_RSOCKET_ACCEPT_MIMETYPES.value.name: StreamDataMimetypes,
+    WellKnownMimeTypes.MESSAGE_RSOCKET_AUTHENTICATION.value.name: AuthenticationContent
+}
+
 
+def metadata_item_factory(metadata_encoding: bytes) -> Type[CompositeMetadataItem]:
     return metadata_item_factory_by_type.get(metadata_encoding, CompositeMetadataItem)
 
 

rsocket/extensions/mimetypes.py

@@ -4,7 +4,7 @@
 from rsocket.exceptions import RSocketUnknownMimetype
 from rsocket.extensions.mimetype import WellKnownMimeType
 from rsocket.frame_helpers import ensure_bytes
-from rsocket.helpers import map_types_by_id, map_types_by_name
+from rsocket.helpers import map_type_names_by_id, map_type_ids_by_name
 
 
 @unique
@@ -80,8 +80,8 @@ def get_by_name(cls, metadata_name: str) -> Optional[WellKnownMimeType]:
         return mimetype_by_name.get(metadata_name)
 
 
-mimetype_by_id = map_types_by_id(WellKnownMimeTypes)
-mimetype_by_name = map_types_by_name(WellKnownMimeTypes)
+mimetype_by_id = map_type_names_by_id(WellKnownMimeTypes)
+mimetype_by_name = map_type_ids_by_name(WellKnownMimeTypes)
 
 
 def ensure_encoding_name(encoding: Union[WellKnownMimeTypes, str, bytes]) -> bytes: