runtimepy.net.websocket package

runtimepy.net.websocket package#

A module implementing a WebSocket connection interface.

class runtimepy.net.websocket.connection.EchoWebsocketConnection(protocol: WebSocketClientProtocol | WebSocketServerProtocol)[source]#

An echo connection for WebSocket.

class runtimepy.net.websocket.connection.NullWebsocketConnection(protocol: WebSocketClientProtocol | WebSocketServerProtocol)[source]#

A null WebSocket connection.

class runtimepy.net.websocket.connection.WebsocketConnection(protocol: WebSocketClientProtocol | WebSocketServerProtocol)[source]#

A simple websocket connection interface.

async classmethod app(stop_sig: Event, init: Callable[[T], Awaitable[bool]] = None, manager: ConnectionManager = None, serving_callback: Callable[[WebSocketServer], None] = None, **kwargs) → None[source]#: Run a WebSocket-server application.

classmethod client(uri: str, **kwargs) → AsyncIterator[T][source]#: A wrapper for connecting a client.

async classmethod create_connection(uri: str, **kwargs) → T[source]#: Connect a client to an endpoint.

classmethod create_pair() → AsyncIterator[tuple[T, T]][source]#: Obtain a connected pair of WebsocketConnection objects.

classmethod serve(init: Callable[[T], Awaitable[bool]] = None, stop_sig: Event = None, manager: ConnectionManager = None, **kwargs) → AsyncIterator[WebSocketServer][source]#: Serve a WebSocket server.

classmethod server_handler(init: Callable[[T], Awaitable[bool]] = None, stop_sig: Event = None, manager: ConnectionManager = None) → Callable[[WebSocketServerProtocol], Awaitable[None]][source]#: A wrapper for passing in a websocket handler and initializing a connection.

A module aggregating all WebSocket-related interfaces.

class runtimepy.net.websocket.EchoWebsocketConnection(protocol: WebSocketClientProtocol | WebSocketServerProtocol)[source]#

An echo connection for WebSocket.

class runtimepy.net.websocket.NullWebsocketConnection(protocol: WebSocketClientProtocol | WebSocketServerProtocol)[source]#

A null WebSocket connection.

class runtimepy.net.websocket.WebsocketConnection(protocol: WebSocketClientProtocol | WebSocketServerProtocol)[source]#

A simple websocket connection interface.

async classmethod app(stop_sig: Event, init: Callable[[T], Awaitable[bool]] = None, manager: ConnectionManager = None, serving_callback: Callable[[WebSocketServer], None] = None, **kwargs) → None[source]#: Run a WebSocket-server application.

classmethod client(uri: str, **kwargs) → AsyncIterator[T][source]#: A wrapper for connecting a client.

async classmethod create_connection(uri: str, **kwargs) → T[source]#: Connect a client to an endpoint.

classmethod create_pair() → AsyncIterator[tuple[T, T]][source]#: Obtain a connected pair of WebsocketConnection objects.

classmethod serve(init: Callable[[T], Awaitable[bool]] = None, stop_sig: Event = None, manager: ConnectionManager = None, **kwargs) → AsyncIterator[WebSocketServer][source]#: Serve a WebSocket server.

classmethod server_handler(init: Callable[[T], Awaitable[bool]] = None, stop_sig: Event = None, manager: ConnectionManager = None) → Callable[[WebSocketServerProtocol], Awaitable[None]][source]#: A wrapper for passing in a websocket handler and initializing a connection.