User session

Let's add a server side session to store the logged-in user's state. Later on it will be used to temporarily store the messages which will be delivered to the client.

See resulting code on GitHub

Server side

Data-classes

First we will add some dataclasses to represent a single user session, and a lookup dictionary for all user sessions by their id.

from dataclasses import dataclass, field
from typing import Dict
from weakref import WeakValueDictionary

class SessionId(str):
    pass

@dataclass()
class UserSessionData:
    username: str
    session_id: SessionId

@dataclass(frozen=True)
class ChatData:
    user_session_by_id: Dict[SessionId, UserSessionData] = field(default_factory=WeakValueDictionary)

chat_data = ChatData()

The SessionId defined in Lines 5-6 is required in order to store the string session-id as a weak reference later on.

Lines 8-11 define the UserSessionData dataclass which represents the user's session. It contains two fields:

• username - Human readable name specified in the login payload.
• session_id - unique id (e.g. UUID4) generated to identify the session.

Lines 13-15 define the ChatData dataclass which represents the application's data. For now, it contains only a dict for looking up user sessions by their id. It is initialized as a WeakValueDictionary in order for the session to be removed when the user disconnects.

Line 17 instantiates a global instance of this class.

Login endpoint

Next we will modify the login endpoint to create a user session:

import logging
import uuid
from typing import Optional, Awaitable

from rsocket.frame_helpers import ensure_bytes
from rsocket.helpers import utf8_decode, create_response
from rsocket.payload import Payload
from rsocket.routing.request_router import RequestRouter

class ChatUserSession:

    def __init__(self):
        self._session: Optional[UserSessionData] = None

    def router_factory(self):
        router = RequestRouter()

        @router.response('login')
        async def login(payload: Payload) -> Awaitable[Payload]:
            username = utf8_decode(payload.data)

            logging.info(f'New user: {username}')

            session_id = SessionId(uuid.uuid4())
            self._session = UserSessionData(username, session_id)
            chat_data.user_session_by_id[session_id] = self._session

            return create_response(ensure_bytes(session_id))

        return router

In order to keep a reference to the UserSessionData we will modify the request handler factory. The ChatUserSession class will keep the reference to the session data, and define the request routes.

Below is the modified handler_factory:

class CustomRoutingRequestHandler(RoutingRequestHandler):
    def __init__(self, session: ChatUserSession):
        super().__init__(session.router_factory())
        self._session = session

def handler_factory():
    return CustomRoutingRequestHandler(ChatUserSession())

The CustomRoutingRequestHandler class (Lines 1-4) is the actual request handler which will wrap the ChatUserSession instance.

Finally, we modify the handler_factory (Lines 6-7) to instantiate the handler and the session.

Client side

Below is the modified ChatClient:

from rsocket.extensions.helpers import composite, route
from rsocket.frame_helpers import ensure_bytes
from rsocket.payload import Payload
from rsocket.rsocket_client import RSocketClient

class ChatClient:
    def __init__(self, rsocket: RSocketClient):
        self._rsocket = rsocket
        self._session_id = None
        self._username: Optional[str] = None

    async def login(self, username: str):
        payload = Payload(ensure_bytes(username), composite(route('login')))
        response = await self._rsocket.request_response(payload)
        self._session_id = response.data
        self._username = username
        return self

Instead of a greeting from the server, we now receive a session id in the response payload. Line 14 stores this session on the client class.

We also store the username on the client for later printing as part of incoming messages.