eric-sse 0.3.2

A lightweight message dispatcher based on SSE protocol data transfer objects format

The lightweight library for async messaging nobody expects.

Installation

pip install eric-sse

Features

• Send to one listener and broadcast
• SSE format was adopted by design, is order to make library suitable for such kind of model
• Sockets
• Callbacks
• Threading support for large data processing

Possible applications

• Message delivery mechanisms based on SSE
• Message queue processing (logging, etc)
• See https://github.com/laxertu/eric-api

Trivia

Library name pretends to be a tribute to the following movie https://en.wikipedia.org/wiki/Looking_for_Eric

Entities

class eric_sse.entities.AbstractChannel(stream_delay_seconds: int = 0, retry_timeout_milliseconds: int = 5)

Base class for channels.

Provides functionalities for listeners and message delivery management. SSEChannel is the default implementation

add_listener() → MessageQueueListener

Add the default listener

broadcast(msg: Message)

Enqueue a message to all listeners

deliver_next(listener_id: str) → Message

Returns next message for given listener id. Raises a NoMessagesException if queue is empty

dispatch(listener_id: str, msg: Message)

Adds a message to listener’s queue

async message_stream(listener: MessageQueueListener) → AsyncIterable[dict]

Entry point for message streaming

In case of failure at channel resolution time, a special message with type=MESSAGE_TYPE_CLOSED is sent, and correspondant listener is stopped

register_listener(l: MessageQueueListener)

Adds a listener to channel

class eric_sse.entities.Message(type: str, payload: dict | list | str | int | float | None = None)

Models a message

It’s just a container of information identified by a type. For validation purposes you can override MessageQueueListener.on_message

class eric_sse.entities.MessageQueueListener

Base class for listeners.

Optionally you can override on_message method if you need to inject code at message delivery time.

on_message(msg: Message) → None

Event handler. It executes when a message is delivered to client

Prefab channels and listeners

class eric_sse.prefabs.DataProcessingChannel(stream_delay_seconds: int = 0, retry_timeout_milliseconds: int = 5)

Channel that invokes a callable in a Pool of threads

add_threaded_listener(callback: Callable[[Message], None], max_workers: int) → ThreadPoolListener

Adds a threaded listener

class eric_sse.prefabs.SSEChannel(stream_delay_seconds: int = 0, retry_timeout_milliseconds: int = 5)

SSE streaming channel.

See https://developer.mozilla.org/en-US/docs/Web/API/Server-sent_events/Using_server-sent_events#event_stream_format Currently, ‘id’ field is not supported.

class eric_sse.prefabs.ThreadPoolListener(callback: Callable[[Message], None], max_workers: int)

CURRENTLY NOT SUITABLE FOR PRODUCTION ENVIRONMENTS.

Listener intended for concurrent processing of data.

Relies on concurrent.futures.ThreadPoolExecutor.

MESSAGE_TYPE_CLOSED type is intended as end of stream. It should be considered as a reserved Message type.

Note that:

• same callback is invoked, no matter of message type
• callback execution order is not guaranteed (to be the same as the one while dispatching to channel)

on_message(msg: Message) → None

Event handler. It executes when a message is delivered to client

Prefab servers

class eric_sse.servers.ChannelContainer

Helper class for management of multiple SSE channels cases of use.

class eric_sse.servers.SocketServer(file_descriptor_path: str)

An implementation of a socket server that receives and broadcasts automatically all messages that receives

A static shortcut for starting a basic server is provided. See examples.

static start(file_descriptor_path: str)

Shortcut to start a server

Exceptions

exception eric_sse.exception.InvalidChannelException

exception eric_sse.exception.InvalidListenerException

exception eric_sse.exception.InvalidMessageFormat

exception eric_sse.exception.NoMessagesException

Raised when trying to fetch from an empty queue

Developers section

Update README.md scipt: update_docs.sh