refactoring and updating to match guide

Author: jell-o-fishi

examples/tutorial/reactivex/chat_client.py

@@ -7,7 +7,7 @@
 
 from reactivex import operators
 
-from examples.tutorial.reactivex.models import (Message, chat_filename_mimetype, ServerStatistics, ClientStatistics,
+from examples.tutorial.reactivex.shared import (Message, chat_filename_mimetype, ServerStatistics, ClientStatistics,
                                                 ServerStatisticsRequest, encode_dataclass, dataclass_to_payload,
                                                 decode_dataclass)
 from rsocket.extensions.helpers import composite, route, metadata_item

examples/tutorial/reactivex/chat_server.py

@@ -11,7 +11,7 @@
 from more_itertools import first
 from reactivex import Observable, operators, Subject, Observer
 
-from examples.tutorial.reactivex.models import (Message, chat_filename_mimetype, ClientStatistics,
+from examples.tutorial.reactivex.shared import (Message, chat_filename_mimetype, ClientStatistics,
                                                 ServerStatisticsRequest, ServerStatistics, dataclass_to_payload,
                                                 decode_dataclass)
 from rsocket.extensions.composite_metadata import CompositeMetadata

examples/tutorial/reactivex/models.py renamed to examples/tutorial/reactivex/shared.py

@@ -2,7 +2,7 @@
 import logging
 from typing import Optional
 
-from examples.tutorial.step3.models import Message, encode_dataclass, decode_dataclass
+from examples.tutorial.step3.shared import Message, encode_dataclass, decode_dataclass
 from reactivestreams.subscriber import DefaultSubscriber
 from reactivestreams.subscription import DefaultSubscription
 from rsocket.extensions.helpers import composite, route

examples/tutorial/step3/chat_client.py

@@ -8,7 +8,7 @@
 
 from more_itertools import first
 
-from examples.tutorial.step3.models import Message, dataclass_to_payload, decode_dataclass
+from examples.tutorial.step3.shared import Message, dataclass_to_payload, decode_dataclass
 from reactivestreams.publisher import DefaultPublisher, Publisher
 from reactivestreams.subscriber import Subscriber
 from reactivestreams.subscription import DefaultSubscription

examples/tutorial/step3/chat_server.py

@@ -2,7 +2,7 @@
 import logging
 from typing import List, Optional
 
-from examples.tutorial.step4.models import Message, encode_dataclass, decode_dataclass
+from examples.tutorial.step4.shared import Message, encode_dataclass, decode_dataclass
 from reactivestreams.subscriber import DefaultSubscriber
 from reactivestreams.subscription import DefaultSubscription
 from rsocket.awaitable.awaitable_rsocket import AwaitableRSocket

examples/tutorial/step3/models.py renamed to examples/tutorial/step3/shared.py

@@ -9,8 +9,7 @@
 
 from more_itertools import first
 
-from examples.tutorial.step4.models import (Message, chat_filename_mimetype, dataclass_to_payload, decode_dataclass,
-                                            decode_payload)
+from examples.tutorial.step4.shared import (Message, chat_filename_mimetype, dataclass_to_payload, decode_payload)
 from reactivestreams.publisher import DefaultPublisher, Publisher
 from reactivestreams.subscriber import Subscriber
 from reactivestreams.subscription import DefaultSubscription
@@ -89,7 +88,7 @@ def __init__(self):
         self._session: Optional[UserSessionData] = None
 
     def router_factory(self):
-        router = RequestRouter(decode_payload)
+        router = RequestRouter(payload_mapper=decode_payload)
 
         @router.response('login')
         async def login(payload: Payload) -> Awaitable[Payload]:
@@ -102,15 +101,13 @@ async def login(payload: Payload) -> Awaitable[Payload]:
             return create_response(ensure_bytes(session_id))
 
         @router.response('channel.join')
-        async def join_channel(payload: Payload) -> Awaitable[Payload]:
-            channel_name = utf8_decode(payload.data)
+        async def join_channel(channel_name: str) -> Awaitable[Payload]:
             ensure_channel_exists(channel_name)
             chat_data.channel_users[channel_name].add(self._session.session_id)
             return create_response()
 
         @router.response('channel.leave')
-        async def leave_channel(payload: Payload) -> Awaitable[Payload]:
-            channel_name = utf8_decode(payload.data)
+        async def leave_channel(channel_name: str) -> Awaitable[Payload]:
             chat_data.channel_users[channel_name].discard(self._session.session_id)
             return create_response()
 
@@ -157,9 +154,7 @@ async def _message_sender(self):
             return MessagePublisher(self._session)
 
         @router.stream('channel.users')
-        async def get_channel_users(payload: Payload) -> Publisher:
-            channel_name = utf8_decode(payload.data)
-
+        async def get_channel_users(channel_name: str) -> Publisher:
             if channel_name not in chat_data.channel_users:
                 return EmptyStream()
 

examples/tutorial/step4/chat_client.py

@@ -32,5 +32,12 @@ def decode_dataclass(data: bytes, cls: Type[T]) -> T:
     return cls(**json.loads(utf8_decode(data)))
 
 
-def decode_payload(cls: Type[T], payload: Payload) -> T:
-    return decode_dataclass(payload.data, cls)
+def decode_payload(cls, payload: Payload):
+    data = payload.data
+
+    if cls is bytes:
+        return data
+    if cls is str:
+        return utf8_decode(data)
+
+    return decode_dataclass(data, cls)

examples/tutorial/step4/chat_server.py

@@ -2,7 +2,7 @@
 import logging
 from typing import List, Optional
 
-from examples.tutorial.step5.models import Message, chat_filename_mimetype, encode_dataclass, decode_dataclass
+from examples.tutorial.step5.shared import Message, chat_filename_mimetype, encode_dataclass, decode_dataclass
 from reactivestreams.subscriber import DefaultSubscriber
 from reactivestreams.subscription import DefaultSubscription
 from rsocket.awaitable.awaitable_rsocket import AwaitableRSocket