Lihil Web Framework

The Socket Channel protocol uses a standardized MessageEnvelope to facilitate request-response pairing and event tracking between clients and the server.

Inbound Envelope (Client to Server)

When a client sends a message, it should include:

• topic: The target topic string.
• event: The event name (e.g., join, leave, send_message).
• payload: The event data.
• ref: A client-generated request ID used to pair the request with a subsequent server response.
• join_ref: An ID used to distinguish the lifecycle of a specific topic subscription on the same socket.

Outbound Envelope (Server to Client)

When the server sends a message, it includes:

• topic: The target topic.
• event: The server-side event name.
• payload: The event data.
• ref: If this is a response to a request, this must contain the original client ref.
• join_ref: The topic lifecycle reference.
• event_id: A unique server-side event ID (useful for replay/recovery).
• seq: A monotonic sequence number for the connection (useful for ordering and debugging).

// Example Inbound Envelope
{
  "topic": "session:abc",
  "event": "join",
  "payload": {},
  "ref": "1",
  "join_ref": null
}

// Example Outbound Envelope
{
  "topic": "session:abc",
  "event": "reply",
  "payload": {
    "status": "ok",
    "response": {}
  },
  "ref": "1",
  "join_ref": "1",
  "event_id": "evt_123",
  "seq": 12
}

The SocketHub is a high-level WebSocket feature using class-based channels.

Key Concepts:

• Channels: Subclass ChannelBase and declare a topic using Topic("pattern:{param}").
• Typed Params: Declare parameters directly in lifecycle hooks like async def on_join(self, room_id: str). Lihil injects these and validates them.
• Bus Fanout: Use await self.publish(payload, event="event_name") to broadcast messages to all subscribers of a topic.
• Task Lifecycle: Use self.start_task(name, coro) for long-running work. These tasks are automatically cancelled on disconnect.
• Dependency Injection: Access dependencies within channels using self.resolver.aresolve(...).
• Registration: Register a channel with hub.channel(MyChannel) and mount the hub to the application using app = Lihil(hub).

Protocol Envelope: Communication uses a stable envelope containing ref (request reference), join_ref (for connection-local pairing), and seq (sequence) fields to allow clients to pair requests with replies.

Lihil uses a plugin system that allows you to integrate external libraries by wrapping endpoint functionality. Plugins implement the IPlugin protocol and can access full endpoint context, including signatures and dependency graphs.

Plugin Execution Flow

Plugins are applied during setup (left to right) and execute at runtime using a nested/onion pattern. When multiple plugins are provided in the plugins list, the first plugin in the list becomes the outermost layer of the onion, and the last plugin becomes the innermost layer.

Execution Order: Request $\rightarrow$ Plugin[0] $\rightarrow$ Plugin[1] $\dots$ $\rightarrow$ Original Function $\rightarrow$ Plugin[n] $\dots$ $\rightarrow$ Plugin[0] $\rightarrow$ Response

Example Flow: If you define an endpoint with plugins=[plugin.timeout(5), plugin.retry(max_attempts=3), plugin.cache(expire_s=60)]:

1. timeout (Outermost)
2. retry (Middle)
3. cache (Innermost)
4. endpoint function
5. cache $\rightarrow$ retry $\rightarrow$ timeout $\rightarrow$ Response

@app.sub("/api").get(plugins=[
    plugin.timeout(5),           # Applied 1st → Executes Outermost
    plugin.retry(max_attempts=3), # Applied 2nd → Executes Middle
    plugin.cache(expire_s=60),   # Applied 3rd → Executes Innermost
])

SocketHub is a high-level routing class designed for managed, channel-based WebSocket communication. Unlike the low-level WebSocketRoute, SocketHub does not require a user-supplied handler function; instead, it uses an internal managed loop to dispatch messages based on defined channels and hooks.

Key features include:

• Hooks: Use @on_connect and @on_disconnect to manage connection lifecycles.
• Channels: Define topic patterns using .channel(pattern) and handle events via .on_join, .on_exit, and .on_receive.
• Composition: Reuse channel trees by including other hubs using hub.include(subhub, parent_prefix=...).

Note: SocketHub is intended to replace the older managed features in WebSocketRoute to provide a cleaner separation between low-level socket handling and high-level channel management.

class SocketHub(RouteBase):
    def __init__(self, path: str = "", *, graph: Graph | None = None, middlewares=None):
        # ...

    def channel(self, pattern: str) -> Channel:
        # ...

    def on_connect(self, func: IAsyncFunc[..., None]) -> IAsyncFunc[..., None]:
        # ...

    def on_disconnect(self, func: IAsyncFunc[..., None]) -> IAsyncFunc[..., None]:
        # ...

To implement managed WebSocket logic, use SocketHub to register connection hooks and channel patterns. Channels allow you to match incoming message topics to specific logic.

• on_connect(func): Decorator to register a function called when a new WebSocket connects.
• on_disconnect(func): Decorator to register a function called when a WebSocket disconnects.
• channel(pattern): Registers a new Channel object. The pattern can include placeholders (e.g., "user/{id}"). Channels are matched by complexity (more placeholders take precedence).

When a client sends a message with an event of "join", the channel's on_join callback is triggered. When the client sends "leave", on_exit is triggered.

hub = SocketHub(path="/ws")

@hub.on_connect
async def handle_connect(sock):
    print(f"Connected: {sock.id}")

chat_channel = hub.channel("chat/{room_id}")

@chat_channel.on_join
async def handle_join(params, sock):
    await sock.send_json({"event": "welcome", "payload": params})

To create a custom plugin, implement a class that provides a decorate method. This method receives an IEndpointInfo object, which provides access to the original function, its parsed signature, and the dependency injection graph. The decorate method must return a callable that wraps the original function.

An IPlugin can be either a callable or a class with a decorate method.

from lihil.plugins.interface import IPlugin, IEndpointInfo
from lihil.interface import IAsyncFunc, P, R
from typing import Callable, Awaitable

class MyCustomPlugin:
    def __init__(self, external_service):
        self.service = external_service

    def decorate(self, ep_info: IEndpointInfo[P, R]) -> Callable[P, Awaitable[R]]:
        original_func = ep_info.func

        async def wrapper(*args: P.args, **kwargs: P.kwargs) -> R:
            # Pre-processing
            await self.service.before_request(ep_info.sig)
            try:
                result = await original_func(*args, **kwargs)
                # Post-processing
                return await self.service.process_result(result)
            except Exception as e:
                await self.service.handle_error(e)
                raise

        return wrapper

# Usage
plugin = MyCustomPlugin(ExternalService())

@app.sub("/api/data").get(plugins=[plugin.decorate])
async def get_data() -> dict:
    return {"data": "value"}