Rembus for Python

Rembus is a Pub/Sub and RPC middleware.

Features

• Binary message encoding using CBOR.

• Native support for exchanging DataFrames.

• Persistent storage via DuckDB DuckLake.

• Pub/Sub QOS0, QOS1 and QOS2.

• Hierarchical topic routing with wildcards (*/*/temperature).

• MQTT integration.

• WebSocket transport.

See Rembus.jl broker for a full fledged broker that supports WebSocket, ZMQ and plain tcp protocols and more features like private topics, multi-tenancy and more.

Concepts

• Component: an addressable node in a distributed system. A component connects to a broker and communicates using Pub/Sub and/or RPC semantics.

• Broker: a specialized component responsible for routing Pub/Sub messages and dispatching RPC calls between components. A broker may also persist messages and expose services.

• Topic: a "logical channel" string identifier used for Pub/Sub message routing (e.g. alarm_topic).

• Topic space: a set of topics defined by wildcard patterns (e.g. */telemetry) used for bulk subscription..

• Subscription: a callback bound to a topic or topic space; invoked automatically with the message payload when a Pub/Sub message is published. Supports wildcard topics and optional delivery of messages sent while the subscriber was offline (msgfrom=rb.LastReceived).

• RPC Service: a named function exposed by a component and registered at the broker for remote invocation.

• RPC Call: a synchronous or asynchronous request issued by a component to invoke a remote RPC service.

• Schema: an optional declarative mapping between topic patterns and persistent storage tables, used to structure and persist messages at rest.

• Data at Rest: broker capability to persist published messages into DuckDB tables defined by a schema, enabling historical queries and analytics.

• Mesh Network: since a component can simultaneously act as a broker (port= connecting "upstream" to another broker while also accepting its own "downstream" connections), multiple brokers can be chained together into a mesh network. When a node joins the mesh it advertises, through the identity/authentication handshake, the protocols and ports it is listening on; the upstream broker collects this into router.network, a list of rembus.core.Node, to keep track of the mesh topology:

server = rembus.node(port=8338)
mynode = rembus.node("ws://myhost:8338/mynode", port=9000)

print(server.router.network)
# [mynode -> ws://127.0.0.1:9000 [up]]

Beyond topology discovery, subscribe/expose (and their unsubscribe/ unexpose counterparts) are flooded across the mesh, so every broker on the path learns which neighbor leads towards a given topic. This allows pub/sub messages and RPC calls to be routed/forwarded across multiple broker hops, not just between a broker and its directly connected components. On (re)connection a node also resyncs its upstream broker with its full set of exposed/subscribed topics, so no routing state is lost across reconnections.

Getting Started

Install the package:

pip install rembus

Broker

Start a broker (sync or async):

import rembus as rb

# sync version
bro = rb.node() # equivalent to rb.node(port = 8338)
bro.wait() # event loop, not needed in interactive interpreter


# async version
bro = await rb.component()
await bro.wait() 

Broker with persistent storage

A Rembus broker can be configured with a schema to persist published messages into a DuckDB database via the DataLake extension.

A schema declaratively maps Pub/Sub topic patterns to relational tables. Topics variables are extracted from the topic path and mapped to table columns; message payload fields are mapped to the remaining columns.

Example schema.json:

{
    "tables": [
        {
            "table": "sensor",
            "topic": ":site/:type/:dn/sensor",
            "columns": [
                {"col": "site", "type": "TEXT", "nullable": false},
                {"col": "type", "type": "TEXT", "nullable": false},
                {"col": "dn", "type": "TEXT"}
            ],
            "keys": ["dn"]
        },
        {
            "table": "telemetry",
            "topic": ":dn/telemetry",
            "columns": [
                {"col": "dn", "type": "TEXT"},
                {"col": "temperature", "type": "DOUBLE"},
                {"col": "pressure", "type": "DOUBLE"}
            ],
            "extras": {
                "recv_ts": "ts",
                "slot": "time_bucket"
            }
        }
    ]
}

Schema semantics

• Each entry in tables defines a topic-to-table binding.

• Topic segments prefixed with : are variables extracted from the topic path and written to the corresponding columns.

• Message payload fields are mapped positionally or by name to remaining columns.

• keys define logical primary keys.

• extras specify broker-generated metadata (e.g. receive timestamp, time bucketing).

Example mappings

• Messages published to :site/:type/:dn/sensor are persisted in the sensor table, with site, type, and dn derived from the topic path.

• Messages published to :dn/telemetry are persisted in the telemetry table, with metric values extracted from the message payload.

If a schema is provided, DuckLake tables are created automatically if they do not already exist, and all matching publications are persisted.

For each table three special topics for RPC services are automatically created:

• upsert_<table> insert/update records;
• query_<table> retrieve records;
• delete_<table> remove records;

RPC calls to query_* topics return a Polars DataFrame.

For example:

cli.rpc("query_sensor", {"where": "type='HVAC'"})

Starting a broker with storage enabled

import rembus as rb

bro = await rb.component(schema="schema.json")
await bro.wait() 

Components

Connect to a Broker:

# named component
cli = await rb.component("ws://host:8338/myname")

# default host/port
cli = await rb.component("myname")

# anonymous
cli = await rb.component(rb.anonym(host="host", port=8338))

Why named components

• Enables authentication (RSA/ECDSA/shared secret).

• Allows persistent twin mapping; offline messages are buffered. A Component connects to a Broker using a URL-like formatted string that identifies the broker address and declare the name of the Component.

Server

rembus.server(port=8000) starts a server: it accepts connections like a broker but does not route messages between the connected components:

• an RPC request is resolved only by the methods exposed by the server itself, otherwise it fails with STS_METHOD_NOT_FOUND;
• a direct RPC to another component fails with STS_GENERIC_ERROR (direct not permitted in server mode);
• a published message is delivered only to the server's own subscribers and is not broadcast to the other connected components.

Duplicate connections

By default a broker rejects a connection whose name is already in use by a connected Component (the new connection receives an "already connected" error). To let the new connection replace the prior one, which is closed, set overwrite_connection to True in the broker configuration or export REMBUS_OVERWRITE_CONNECTION=true:

REMBUS_OVERWRITE_CONNECTION=true python my_broker.py

Pub/Sub

Publish

# Single message
await cli.publish("site/type/dn/sensor")
await cli.publish("dn/telemetry", {'temperature': 21.6, 'pressure': 980})

# DataFrame
import polars as pl
df = pl.DataFrame({"dn":["s1","s2"], "temperature":[15,18.8]})
await cli.publish("telemetry", df)

Subscribe

# Single topic
def alarm(slogan, severity):
    print(f"{slogan}: {severity}")

sub = rb.node("monitor")
sub.subscribe(alarm, topic="alarm") # equivalent to sub.subscribe(alarm)

# Topic space
def telemetry(topic, data):
    print(f"{topic}: {data}")

sub.subscribe(telemetry, topic="**/telemetry", msgfrom=rb.LastReceived)

The subscribed alarm function gets called with the using the message payload: slogan as first argument and severity as second argument:

cli.publish("alarm", "mydevice: battery very low", "CRITICAL")

The option msgfrom is for receiving published messages when the component was not subscribed to the topic, for example because it was not connected when the messages was published.

RPC

Expose

import rembus as rb

def add(x,y):
    return x+y

handle = rb.node('calculator')
handle.expose(add)
handle.wait()

Call

import rembus as rb

handle = rb.node("myclient")
result = handle.rpc('add', 1, 2)