updated guide examples

Author: jell-o-fishi

examples/tutorial/reactivex/chat_client.py

@@ -147,6 +147,12 @@ async def list_channels(self) -> List[str]:
         ).pipe(operators.map(lambda _: utf8_decode(_.data)),
                operators.to_list())
 
+    async def list_channel_users(self, channel_name: str) -> List[str]:
+        request = Payload(ensure_bytes(channel_name), composite(route('channel.users')))
+        return await ReactiveXClient(self._rsocket).request_stream(
+            request
+        ).pipe(operators.map(lambda _: utf8_decode(_.data)),
+               operators.to_list())
 
 async def main():
     connection1 = await asyncio.open_connection('localhost', 6565)
@@ -174,16 +180,21 @@ async def messaging_example(user1, user2):
     user1.listen_for_messages()
     user2.listen_for_messages()
 
-    await user1.join('channel1')
-    await user2.join('channel1')
+    channel_name = 'channel1'
+    await user1.join(channel_name)
+    await user2.join(channel_name)
 
     print(f'Channels: {await user1.list_channels()}')
+    print(f'Channel {channel_name} users: {await user1.list_channel_users(channel_name)}')
 
     await user1.private_message('user2', 'private message from user1')
-    await user1.channel_message('channel1', 'channel message from user1')
+    await user1.channel_message(channel_name, 'channel message from user1')
 
     await asyncio.sleep(1)
 
+    await user1.leave(channel_name)
+    print(f'Channel {channel_name} users: {await user1.list_channel_users(channel_name)}')
+
     user1.stop_listening_for_messages()
     user2.stop_listening_for_messages()
 

examples/tutorial/reactivex/chat_server.py

@@ -13,7 +13,7 @@
 
 from examples.tutorial.reactivex.shared import (Message, chat_filename_mimetype, ClientStatistics,
                                                 ServerStatisticsRequest, ServerStatistics, dataclass_to_payload,
-                                                decode_dataclass)
+                                                decode_dataclass, decode_payload)
 from rsocket.extensions.composite_metadata import CompositeMetadata
 from rsocket.extensions.helpers import composite, metadata_item
 from rsocket.frame_helpers import ensure_bytes
@@ -36,15 +36,15 @@ class SessionId(str):  # allow weak reference
 @dataclass()
 class UserSessionData:
     username: str
-    session_id: str
+    session_id: SessionId
     messages: Queue = field(default_factory=Queue)
     statistics: Optional[ClientStatistics] = None
     requested_statistics: ServerStatisticsRequest = field(default_factory=ServerStatisticsRequest)
 
 
 @dataclass(frozen=True)
 class ChatData:
-    channel_users: Dict[str, Set[str]] = field(default_factory=lambda: defaultdict(WeakSet))
+    channel_users: Dict[str, Set[SessionId]] = field(default_factory=lambda: defaultdict(WeakSet))
     files: Dict[str, bytes] = field(default_factory=dict)
     channel_messages: Dict[str, Queue] = field(default_factory=lambda: defaultdict(Queue))
     user_session_by_id: Dict[str, UserSessionData] = field(default_factory=WeakValueDictionary)
@@ -95,6 +95,13 @@ def new_statistics_data(requested_statistics: ServerStatisticsRequest):
     return ServerStatistics(**statistics_data)
 
 
+def find_username_by_session(session_id: SessionId) -> Optional[str]:
+    session = chat_data.user_session_by_id.get(session_id)
+    if session is None:
+        return None
+    return session.username
+
+
 class ChatUserSession:
 
     def __init__(self):
@@ -105,11 +112,10 @@ def remove(self):
         del chat_data.user_session_by_id[self._session.session_id]
 
     def router_factory(self):
-        router = RequestRouter()
+        router = RequestRouter(payload_mapper=decode_payload)
 
         @router.response('login')
-        async def login(payload: Payload) -> Observable:
-            username = utf8_decode(payload.data)
+        async def login(username: str) -> Observable:
             logging.info(f'New user: {username}')
             session_id = SessionId(uuid.uuid4())
             self._session = UserSessionData(username, session_id)
@@ -118,15 +124,13 @@ async def login(payload: Payload) -> Observable:
             return reactivex.just(Payload(ensure_bytes(session_id)))
 
         @router.response('channel.join')
-        async def join_channel(payload: Payload) -> Observable:
-            channel_name = utf8_decode(payload.data)
+        async def join_channel(channel_name: str) -> Observable:
             ensure_channel_exists(channel_name)
             chat_data.channel_users[channel_name].add(self._session.session_id)
             return reactivex.empty()
 
         @router.response('channel.leave')
-        async def leave_channel(payload: Payload) -> Observable:
-            channel_name = utf8_decode(payload.data)
+        async def leave_channel(channel_name: str) -> Observable:
             chat_data.channel_users[channel_name].discard(self._session.session_id)
             return reactivex.empty()
 
@@ -157,10 +161,17 @@ async def get_channels() -> Observable:
             return reactivex.from_iterable(
                 (Payload(ensure_bytes(channel)) for channel in chat_data.channel_messages.keys()))
 
-        @router.fire_and_forget('statistics')
-        async def receive_statistics(payload: Payload):
-            statistics = decode_dataclass(payload.data, ClientStatistics)
+        @router.stream('channel.users')
+        async def get_channel_users(channel_name: str) -> Observable:
+            if channel_name not in chat_data.channel_users:
+                return reactivex.empty()
 
+            return reactivex.from_iterable(Payload(ensure_bytes(find_username_by_session(session_id))) for
+                         session_id in
+                         chat_data.channel_users[channel_name])
+
+        @router.fire_and_forget('statistics')
+        async def receive_statistics(statistics: ClientStatistics):
             logging.info('Received client statistics. memory usage: %s', statistics.memory_usage)
 
             self._session.statistics = statistics
@@ -198,9 +209,7 @@ def on_next(payload: Payload):
                 limit_rate=2)
 
         @router.response('message')
-        async def send_message(payload: Payload) -> Observable:
-            message = decode_dataclass(payload.data, Message)
-
+        async def send_message(message: Message) -> Observable:
             logging.info('Received message for user: %s, channel: %s', message.user, message.channel)
 
             target_message = Message(self._session.username, message.content, message.channel)

examples/tutorial/reactivex/shared.py

@@ -47,3 +47,14 @@ def dataclass_to_payload(obj) -> Payload:
 
 def decode_dataclass(data: bytes, cls: Type[T]) -> T:
     return cls(**json.loads(utf8_decode(data)))
+
+
+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_client.py

@@ -77,6 +77,11 @@ async def list_channels(self) -> List[str]:
         response = await AwaitableRSocket(self._rsocket).request_stream(request)
         return list(map(lambda _: utf8_decode(_.data), response))
 
+    async def list_channel_users(self, channel_name: str):
+        request = Payload(ensure_bytes(channel_name), composite(route('channel.users')))
+        response = await AwaitableRSocket(self._rsocket).request_stream(request)
+        return list(map(lambda _: utf8_decode(_.data), response))
+
 
 async def main():
     connection1 = await asyncio.open_connection('localhost', 6565)
@@ -100,16 +105,22 @@ async def messaging_example(user1: ChatClient, user2: ChatClient):
     user1.listen_for_messages()
     user2.listen_for_messages()
 
-    await user1.join('channel1')
-    await user2.join('channel1')
+    channel_name = 'channel1'
+
+    await user1.join(channel_name)
+    await user2.join(channel_name)
 
     print(f'Channels: {await user1.list_channels()}')
+    print(f'Channel {channel_name} users: {await user1.list_channel_users(channel_name)}')
 
     await user1.private_message('user2', 'private message from user1')
-    await user1.channel_message('channel1', 'channel message from user1')
+    await user1.channel_message(channel_name, 'channel message from user1')
 
     await asyncio.sleep(1)
 
+    await user1.leave(channel_name)
+    print(f'Channel {channel_name} users: {await user1.list_channel_users(channel_name)}')
+
     user1.stop_listening_for_messages()
     user2.stop_listening_for_messages()
 

examples/tutorial/step5/chat_client.py

@@ -38,11 +38,6 @@ async def leave(self, channel_name: str):
         await self._rsocket.request_response(request)
         return self
 
-    async def get_users(self, channel_name: str) -> List[str]:
-        request = Payload(ensure_bytes(channel_name), composite(route('channel.users')))
-        users = await AwaitableRSocket(self._rsocket).request_stream(request)
-        return [utf8_decode(user.data) for user in users]
-
     def listen_for_messages(self):
         def print_message(data: bytes):
             message = decode_dataclass(data, Message)
@@ -97,6 +92,11 @@ async def list_channels(self) -> List[str]:
         response = await AwaitableRSocket(self._rsocket).request_stream(request)
         return list(map(lambda _: utf8_decode(_.data), response))
 
+    async def list_channel_users(self, channel_name: str):
+        request = Payload(ensure_bytes(channel_name), composite(route('channel.users')))
+        response = await AwaitableRSocket(self._rsocket).request_stream(request)
+        return list(map(lambda _: utf8_decode(_.data), response))
+
 
 async def main():
     connection1 = await asyncio.open_connection('localhost', 6565)
@@ -129,15 +129,15 @@ async def messaging_example(user1: ChatClient, user2: ChatClient):
     await user2.join(channel_name)
 
     print(f'Channels: {await user1.list_channels()}')
-    print(f'Channel {channel_name} users: {await user1.get_users(channel_name)}')
+    print(f'Channel {channel_name} users: {await user1.list_channel_users(channel_name)}')
 
     await user1.private_message('user2', 'private message from user1')
     await user1.channel_message(channel_name, 'channel message from user1')
 
     await asyncio.sleep(1)
 
     await user1.leave(channel_name)
-    print(f'Channel {channel_name} users: {await user1.get_users(channel_name)}')
+    print(f'Channel {channel_name} users: {await user1.list_channel_users(channel_name)}')
 
     user1.stop_listening_for_messages()
     user2.stop_listening_for_messages()

examples/tutorial/step5/chat_server.py

@@ -9,7 +9,7 @@
 
 from more_itertools import first
 
-from examples.tutorial.step5.shared import (Message, chat_filename_mimetype, dataclass_to_payload, decode_dataclass)
+from examples.tutorial.step5.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
@@ -95,11 +95,10 @@ def remove(self):
         del chat_data.user_session_by_id[self._session.session_id]
 
     def router_factory(self):
-        router = RequestRouter()
+        router = RequestRouter(payload_mapper=decode_payload)
 
         @router.response('login')
-        async def login(payload: Payload) -> Awaitable[Payload]:
-            username = utf8_decode(payload.data)
+        async def login(username: str) -> Awaitable[Payload]:
             logging.info(f'New user: {username}')
             session_id = SessionId(uuid.uuid4())
             self._session = UserSessionData(username, session_id)
@@ -108,22 +107,18 @@ 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()
 
         @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()
 
@@ -160,9 +155,7 @@ async def get_channels() -> Publisher:
             return StreamFromGenerator(lambda: generator)
 
         @router.response('message')
-        async def send_message(payload: Payload) -> Awaitable[Payload]:
-            message = decode_dataclass(payload.data, Message)
-
+        async def send_message(message: Message) -> Awaitable[Payload]:
             logging.info('Received message for user: %s, channel: %s', message.user, message.channel)
 
             target_message = Message(self._session.username, message.content, message.channel)

examples/tutorial/step5/shared.py

@@ -30,3 +30,14 @@ def dataclass_to_payload(obj) -> Payload:
 
 def decode_dataclass(data: bytes, cls: Type[T]) -> T:
     return cls(**json.loads(utf8_decode(data)))
+
+
+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/step6/chat_client.py

@@ -139,6 +139,11 @@ async def list_channels(self) -> List[str]:
         response = await AwaitableRSocket(self._rsocket).request_stream(request)
         return list(map(lambda _: utf8_decode(_.data), response))
 
+    async def list_channel_users(self, channel_name: str):
+        request = Payload(ensure_bytes(channel_name), composite(route('channel.users')))
+        response = await AwaitableRSocket(self._rsocket).request_stream(request)
+        return list(map(lambda _: utf8_decode(_.data), response))
+
 
 async def main():
     connection1 = await asyncio.open_connection('localhost', 6565)
@@ -162,20 +167,26 @@ async def main():
             await files_example(user1, user2)
 
 
-async def messaging_example(user1, user2):
+async def messaging_example(user1: ChatClient, user2: ChatClient):
     user1.listen_for_messages()
     user2.listen_for_messages()
 
-    await user1.join('channel1')
-    await user2.join('channel1')
+    channel_name = 'channel1'
+
+    await user1.join(channel_name)
+    await user2.join(channel_name)
 
     print(f'Channels: {await user1.list_channels()}')
+    print(f'Channel {channel_name} users: {await user1.list_channel_users(channel_name)}')
 
     await user1.private_message('user2', 'private message from user1')
-    await user1.channel_message('channel1', 'channel message from user1')
+    await user1.channel_message(channel_name, 'channel message from user1')
 
     await asyncio.sleep(1)
 
+    await user1.leave(channel_name)
+    print(f'Channel {channel_name} users: {await user1.list_channel_users(channel_name)}')
+
     user1.stop_listening_for_messages()
     user2.stop_listening_for_messages()