Rembus

struct RembusError <: Rembus.RembusException

Generic Rembus error.

Fields

• code::UInt8: error code

• topic::Union{Nothing, String}: topic name if available

• reason::Union{Nothing, String}: detailed error message

struct RembusTimeout{T} <: Rembus.RembusException

Exception thrown when a response is not received before the request timeout expires.

Fields

• msg::Any: request message

• status::UInt8

struct RpcMethodException <: Rembus.RembusException

Thrown when a RPC method throws an exception.

Fields

• topic::String: service name

• reason::String: remote exception description

Example

A component exposes a method that expect a string argument.

@expose foo(name::AbstractString) = "hello " * name

A RPC client invoke the method with an integer argument.

try
    @rpc foo(1)
catch e
    @error e.reason
end

The result is an exception:

┌ Error: MethodError: no method matching foo(::UInt64)
│
│ Closest candidates are:
│   foo(!Matched::AbstractString)
│    @ Main REPL[2]:1
└ @ Main REPL

struct RpcMethodLoopback <: Rembus.RembusException

Thrown when a RPC request would invoke a locally exposed method.

Fields

• topic::String: service name

struct RpcMethodNotFound <: Rembus.RembusException

Exception thrown from a rpc request when the called method is unknown.

Fields

• topic::String: service name

Example

An RPC Client request a method that does not exist.

@rpc coolservice()

The result is an exception:

ERROR: Rembus.RpcMethodNotFound("rembus", "coolservice")
Stacktrace:
...

struct RpcMethodUnavailable <: Rembus.RembusException

Thrown when a RPC method is unavailable.

A method is considered unavailable when some component that exposed the method is currently disconnected from the broker.

Fields

• topic::String: service name

close(rb::Rembus.Twin)

Close the connection and terminate the component.

isopen(rb::Rembus.Twin) -> Bool

Check if the component is connected to the broker.

wait(rb::Rembus.Twin; reactive)

Wait for RPC requests and Pub/Sub messages.

_topics(results, target::Twin, topic_map)

Shared helper for topic_impls and topic_interests: extend the results collection with every key of topic_map (either router.topic_impls or router.topic_interests) that has at least one twin other than target associated with it, then return results as a Vector (CBOR encodes vectors more compactly than sets).

Excluding target ensures a peer is never told about its own topic — relevant when the attestation response is computed for the very twin whose link is being (re)established.

admin_broadcast(router::Router, twin::Twin, msg::RembusMsg)

Flood msg to every named, open, directly-connected neighbor of router (router.id_twin), except:

• twin itself, the originator of the command;
• any neighbor whose rid is already present in msg.data["touch"] (already reached through another path — see color_admin).

Each twin that is actually sent the message is marked via color_admin before transmission. This is the link-level half of the mesh flood; the router-level half (stopping re-processing once a router has already seen the command) is mark_and_broadcast, which calls this function.

Finally, if twin registered a future for msg.id (a direct, synchronous admin request), it is resolved here with STS_SUCCESS, so the originator gets its response immediately without waiting for the flood to finish propagating through the rest of the mesh.

admin_command(router::Router, twin, msg::AdminReqMsg)

Dispatch and execute an administration command received from twin, based on msg.data[COMMAND]. This is the single entry point for every mesh-routing state change as well as for broker administration (topic privacy, authorization, configuration, shutdown, logging level).

Mesh-routing-relevant commands:

• SETUP_CMD: bulk-register twin's already-known "subscribers" and "exposers" into router.topic_interests/router.topic_impls (see update_tables for the symmetric client-side counterpart), then reply with EnableReactiveMsg. Used to (re)synchronize a link's full state in one shot, typically on reconnection (see twin_setup / reconnect).
• SUBSCRIBE_CMD / EXPOSE_CMD / UNSUBSCRIBE_CMD / UNEXPOSE_CMD: after an isauthorized check, propagate the change mesh-wide via mark_and_broadcast and, only if that router had not already processed this command, apply the corresponding local update to router.topic_interests/router.topic_impls (and twin.msg_from for subscriptions).
• PRIVATE_TOPICS_CONFIG_CMD, PRIVATE_TOPIC_CMD / PUBLIC_TOPIC_CMD, AUTHORIZE_CMD / UNAUTHORIZE_CMD: manage router.topic_auth (see private_topic, public_topic, authorize, unauthorize).
• REACTIVE_CMD, BROKER_CONFIG_CMD, LOAD_CONFIG_CMD, SAVE_CONFIG_CMD, SHUTDOWN_CMD, ENABLE_DEBUG_CMD, DISABLE_DEBUG_CMD: broker-local administration, gated by isadmin where privileged.

Returns a ResMsg carrying the resulting status and, for query-style commands, the requested data.

admin_msg(router::Router, msg)

Entry point for inbound AdminReqMsgs received on a twin's connection: after the command_permitted check, dispatch to admin_command and deliver the result back onto twin's own inbox:

• if the command returned EnableReactiveMsg (the REACTIVE_CMD case), start a dedicated start_reactive process to replay data-at-rest and reply STS_SUCCESS directly;
• otherwise the ResMsg produced by admin_command is delivered as-is (it may need to travel back through respond/sendto_origin if the requestor is several hops away).

Returns false without doing anything else if the permission check fails.

Return an url representing an anonymous component.

attestation(router, twin, msg, authenticate=true)

Server side of the connection handshake, run when an IdentityMsg is received from a connecting component: authenticate the client (unless authenticate is false), bind its identity with setidentity, and reply with get_topics(router, twin) as the response payload — this router's own exports (local methods/subscribers plus whatever is already reachable through other connected twins). The peer applies this payload with update_tables on its side (see authenticate), which is how a newly joined mesh node immediately learns the routing state of the broker it just connected to.

If authentication fails, the websocket is closed instead.

authenticate(router::Router, twin::Twin)

Client side of the connection handshake: declare this node's identity to the peer by sending an IdentityMsg (carrying the bare component id twin.uid.id and the local listener ports) and, if challenged, reply with the signed attestation (see attestate).

On success, the peer's response carries its view of the mesh (its own local methods/subscriptions plus whatever it already knows through other connected twins, computed by get_topics on the peer side) and is applied locally with update_tables, seeding router.topic_impls/ router.topic_interests for the twin representing this link. This is the mirror operation of attestation, which runs on the peer that accepts the connection.

No-op if twin is anonymous or the transport does not require authentication.

authorize(router, twin, msg)

Administration command handler for AUTHORIZE_CMD: grants the component identified by msg.data[CID] access to msg.topic, whether to:

• publish or subscribe to a private topic;
• make RPC requests to, or expose, a remote method.

Requires twin to be an admin and msg.data[CID] to be a non-empty component id. The topic is implicitly declared private (an empty router.topic_auth[msg.topic] entry is created if missing) before the grant is recorded, same as private_topic this is local to the router handling the command.

function authorize(rb, client::AbstractString, topic::AbstractString)

Authorize the client component to use the private topic.

The component must have the admin role for granting topic accessibility.

await_challenge(router::Router, twin::Twin)

Client side counterpart of do_connect for brokers configured with connection_mode === authenticated: instead of proactively sending an IdentityMsg (see authenticate), wait for the peer to send a challenge first (registered as a pending future keyed by CONNECTION_ID) and block until it is answered, raising on timeout or authentication failure. Table synchronization still happens the same way once the attestation completes, via update_tables.

broker(; <keyword arguments>)

Start a broker node and return a handle for interacting with it.

The broker acts as a central node to manage routing of RPC requests and Pub/Sub messages between nodes.

It supports multiple communication protocols (WebSocket, TCP, and ZMQ) and allows for customizable security, authentication, and routing policies.

Keyword arguments

• name::AbstractString="broker": The unique identifier for the broker supervisor process.

• ws=nothing: The WebSocket (ws/wss) listening port. Set to nothing to disable.

• tcp=nothing: The TCP (tcp/tls) listening port. Set to nothing to disable.

• zmq=nothing: The ZMQ Router listening port. Set to nothing to disable.

• prometheus=nothing: The Prometheus port for scraping monitoring metrics. Set to nothing to disable.

• secure=false: If true, enables WSS (WebSocket Secure) and TLS (Transport Layer Security) protocols for encrypted connections.

• authenticated=false: If true, only allows connections from named and authenticated nodes.

• policy::String="first_up": The routing policy used when topics are served by multiple nodes. Possible options include:

  • "first_up": Selects the first connected node from the list of nodes exposing the RPC method.
  • "round_robin": Distributes requests evenly across nodes in a round-robin fashion.
  • "less_busy": Chooses the node with fewer outstanding requests.

  Default Behavior

  If ws, tcp, and zmq are all set to nothing, the broker will default to listening for WebSocket connections on port 8338.

• enc=Rembus.CBOR: wire message format for sending messages Set to Rembus.JSON for JSON-RPC-2.0 text encoding.

color_admin(tw::Twin, msg)

Send msg to the neighbor tw, after "coloring" it: append rid(tw) to msg.data["touch"] (creating the list if absent) just before transmitting.

This is the mechanism that marks a link as already crossed by an admin command, so that tw (or whichever router/component re-broadcasts the message further) never bounces it back to a twin whose id is already in touch. See admin_broadcast and the Mesh Routing and Forwarding guide for the full flood/anti-loop scheme.

command_permitted(router::Router, twin::Twin)

Return true unless router requires authentication (router_isauthenticated) and twin has not authenticated (isauthenticated) and is not connected from the loopback interface (peer_isloopback), in which case the socket is closed and false is returned. Guards every inbound Pub/Sub (pubsub_msg), RPC (rpc_request) and admin (admin_msg) message before any mesh-routing logic (authorization, broadcast, forwarding) is applied.

component(url::AbstractString; <keyword arguments>)

Start a component node and return a handle for interacting with it.

In case of connection lost the underlying supervision logic attempts to reconnect to the broker until it succeed.

The url argument specifies the connection details for the component. For example, the URL ws://127.0.0.1:8338/foo specifies:

• Protocol: ws (WebSocket). Other supported protocols: wss, tcp, tls, zmq.
• Address: 127.0.0.1 (localhost).
• Port: 8338.
• Component Name: foo.

Anonymous connections omit the path part of the URL.

If not specified, Rembus considers the above values as the default values.

This means the URL ws://127.0.0.1:8338/foo is equivalent to simply foo.

Additionally, a component may listen for incoming connections on configured ports, enabling it to act as a broker. These ports are specified using keyword arguments.

Keyword Arguments

• name=missing: Unique string identifier for the component's supervisor process. Defaults to the path part of the url argument if missing.
• ws=nothing: WebSocket (ws/wss) listening port. Set to nothing to disable.
• tcp=nothing: TCP (tcp/tls) listening port. Set to nothing to disable.
• zmq=nothing: ZMQ Router listening port. Set to nothing to disable.
• secure=false: If true, enables WSS (WebSocket Secure) and TLS (Transport Layer Security) protocols for encrypted connections.
• authenticated=false: If true, only allows connections from named and authenticated nodes.
• policy::String="first_up": The routing policy used when topics are served by multiple nodes. Options:
  • "first_up": Selects the first connected node from the list of nodes exposing the RPC method.
  • "round_robin": Distributes requests evenly across nodes in a round-robin fashion.
  • "less_busy": Chooses the node with fewer outstanding requests.
• enc=Rembus.CBOR: wire message format for sending messages Set to Rembus.JSON for JSON-RPC-2.0 text encoding.
• failovers=[]: A list of failover connection URLs to use when the primary connection is down. Each URL should be a string formatted similarly to the url argument.

component(urls::Vector)

Start a component that connects to a pool of nodes defined by the urls array.

connect(
    url::AbstractString;
    enc,
    datadir,
    dbpath
) -> Rembus.Twin

Connect to the remote endpoint defined by url.

A disconnection from the remote endpoint will not trigger automatic reconnection.

Example

rb = connect("ws://127.0.0.1:8338/mycomponent")

destroy_twin(twin, router)

Remove the twin from the system.

Detach twin, shut down its Visor process, and remove it from the router with remove_twin, fully evicting it as a mesh neighbor. Called for anonymous (unnamed) twins when their receiver loop ends; named twins are kept around (just detached) to allow reconnection, see end_receiver.

do_connect(twin::Twin)

Connect to the endpoint declared with REMBUS_BASE_URL env variable.

REMBUS_BASE_URL default to ws://127.0.0.1:8338

A component is considered anonymous when a different and random UUID is used as component identifier each time the application connect to the broker.

Performs the transport-level connection and, for non pre-shared-key authentication, the identity/attestation handshake (see authenticate and await_challenge) that seeds this node's mesh routing tables with the peer's exports via update_tables.

end_receiver(twin::Twin)

Called when a twin's receiver loop terminates (connection closed): keep named twins around (just detached, so reconnection can reuse the same mesh-routing entries once the link comes back up) but fully evict anonymous twins with destroy_twin, since an anonymous component has no stable identity to reconnect against.

expand!(df::DataFrame)

Given a DataFrame df containing a topic string column (e.g. :topic) and a pattern column (e.g. :regexp) with placeholders like :regione/:loc/temperature, this function extracts the named parts from topic according to the regexp and adds them as new columns to df.

Example: df = DataFrame( topic=["veneto/agordo/temperature", "veneto/feltre/temperature"], table=["temperature", "temperature"], regexp=[":regione/:loc/temperature", ":regione/:loc/temperature"] )

expand!(df, :topic, :regexp)

expose(rb, topic::AbstractString, fn::Function)
expose(rb, fn::Function)

Expose the methods of function fn to be executed by rpc clients using topic as RPC method name.

If the topic argument is omitted the function name equals to the RPC method name.

fn returns the RPC response.

Expose the methods of function fn to be invoked by RPC clients, using topic as the RPC method name.

If the topic argument is omitted, the function name is used as the RPC method name.

fn is expected to return the RPC response. Any exceptions thrown by fn are caught and returned as an RPC exception.

get_private_topics(rb)

Return a dictionary mapping private topics to their lists of authorized components.

get_topics(r::Router, target::Twin)

Return [topic_impls(r, target),topic_interests(r, target)], the pair of exports lists sent to target as the reason payload of a successful attestation response (see attestation). The connecting peer applies them with update_tables, completing the handshake half of mesh table synchronization (the other half being the SETUP_CMD admin command, see admin_command).

handle_connection(twin::Twin, failovers)

Setup the twin connection handling and initiate the connection.

• Add failover URLs to the twin and setup the reconnect handler.
• Setup the reconnect handler.
• Connect the twin.

ifdown_block(rb::Rembus.Twin)

Block rpc and publish calls until the twin is connected to the broker.

inject(rb, ctx)

Bind a ctx context to the rb component.

When a ctx context is bound, the method signatures of subscribed and exposed functions change as follows:

• the first argument is the ctx context.
• the second argument is the rb component.
• The remaining arguments correspond to the RPC request's arguments.

The ctx is useful for maintaining local state contextualized to the rb component.

Example

using Rembus

# keep the number of processed messages
mutable struct Context
    msgcount::UInt
end

function topic(context::Context, rb, arg1, arg2)
    context.msgcount += 1
    some_logic(arg1, arg2)
end

ctx = Context(0)
rb = component("myname")
subscribe(rb, topic)
inject(rb, ctx)

In this example, when a message is published:

publish(rb, "topic", arg1, arg2)

The invoked method will receive the context and component as the first two arguments:

foo(container, rb, arg2, arg2)

isadmin(router, twin, cmd)

Return true if twin is listed in router.admins, i.e. is allowed to execute the administration command cmd.

Logs an @error and returns false otherwise. Used as a guard at the beginning of every privileged admin command handler (private_topic, public_topic, authorize, unauthorize, SHUTDOWN_CMD, BROKER_CONFIG_CMD, ...).

isauthenticated(rb)

Return true if the component is authenticated.

isauthorized(router::Router, twin::Twin, topic::AbstractString)

Return true if topic is public (absent from router.topic_auth) or twin is explicitly listed in router.topic_auth[topic].

Every mesh hop re-evaluates this check independently with its own local topic_auth table, so a private topic stays protected regardless of how many brokers a subscribe/expose/publish/request message has to cross: there is no "trusted upstream" shortcut, each router enforces authorization on its own.

ismultipath(router)

Return true if router should maintain mesh routing tables (topic_interests/topic_impls) at all, i.e. it could have more than one outstanding route: it is a real broker or a pool component, as opposed to a plain single-link component.

Currently always returns true; kept as an explicit gate (and extension point) around every table update in admin_command and update_tables so the optimization for non-routing components can be reintroduced without touching call sites.

jsonrpc_pubsub(twin, pkt::Dict, msg_id, params) -> PubSubMsg

Parse a PubSub message: the data is in params.data and params.type holds the QoS/slot flags (default QOS0).

jsonrpc_request(pkt::Dict, msg_id, params) -> RembusMsg

Parse a JSON-RPC request and return the appropriate RembusMsg subtype.

Get the list of distributed services or subscribers.

Arguments:

• router: the router object
• cbtype: callback type ("services" or "subscribers")
• getbody: return the callback code if true

manage_target(router, twin, target_twin, msg)

Deliver a directed RPC request/response msg (one that named an explicit msg.target component or group, see rpc_request) to target_twin:

• STS_TARGET_DOWN back to twin if target_twin's socket is closed;
• resolve it locally via local_fn if target_twin === twin (loopback call to a method exposed by the requestor itself);
• otherwise forward msg onto target_twin's inbox, provided the topic is actually known (router.topic_impls or a built-in command) — letting target_twin (possibly itself a neighbor broker) continue resolving or relaying it further along the mesh; STS_METHOD_NOT_FOUND is replied otherwise.

mark_and_broadcast(router, twin, msg)

Router-level loop guard around admin_broadcast for mesh-wide subscribe/expose/unsubscribe/unexpose propagation.

Stamps msg.data["rmark"] with router.eid:

• if router.eid is already present, this router has already processed msg (the flood looped back to it through a cycle in the mesh graph); the function returns false without re-broadcasting, so the caller must not reapply the corresponding local table update either;
• otherwise router.eid is appended, admin_broadcast is called to flood msg to this router's neighbors, and true is returned so the caller proceeds to update its local topic_interests/topic_impls.

Together with the per-link touch marker used by admin_broadcast, this guarantees a command reaches every router in the mesh exactly once, regardless of topology (tree, ring, or arbitrary graph).

peer_isloopback(socket)

Return true if socket is connected to a peer on the loopback interface.

Fail closed: false is returned when the peer address is not available (for example TLS and ZeroMQ sockets) or cannot be determined.

private_topic(router, twin, msg)

Administration command handler for PRIVATE_TOPIC_CMD: declares msg.topic private by creating an (initially empty) entry in router.topic_auth. Requires twin to be an admin (see isadmin).

This is a purely local-router operation: it is not flooded across the mesh by mark_and_broadcast, so a topic must be declared private on each broker where it should be protected.

private_topic(rb, topic::AbstractString)

Set the specified topic to private.

The component must have the admin role to change the privacy level.

public_topic(router, twin, msg)

Administration command handler for PUBLIC_TOPIC_CMD: resets msg.topic back to public by deleting its entry from router.topic_auth. Requires twin to be an admin (see isadmin).

Like private_topic, this is a local-router-only operation.

public_topic(rb, topic::AbstractString)

Set the specified topic to public.

The component must have the admin role to change the privacy level.

publish(rb, topic::AbstractString, data...; qos=Rembus.QOS0)

Publish one or more data values to a topic.

Each element in data is passed as an argument to the remote method subscribed to topic.

The data arguments may be of any type. For interoperability across components implemented in different languages, the values should be either DataFrames or primitive types that are CBOR-encodable (see RFC 8949).

The keyword argument qos specifies the quality of service (QoS) for message delivery:

• Rembus.QOS0 (default): At most once delivery. A message may be delivered zero or one time.
• Rembus.QOS1: At least once delivery. A message is delivered one or more times.
• Rembus.QOS2: Exactly once delivery. A message is delivered once and only once.

Examples

If the subscriber is a method that expects two arguments:

mytopic(x, y) = ...  # do something with x and y

You can publish a message with two data values::

rb = component("myname")
publish(rb, "mytopic", 1, 2)

If the remote subscribed method has no arguments, publish without data:

publish(rb, "mytopic")

pubsub_msg(router::Router, msg, relay_rule=always_true)

Handle an inbound PubSubMsg: after the command_permitted/ isauthorized checks, (QoS2 de-duplication aside) deliver it in two complementary ways so it reaches every interested party regardless of how far away they are in the mesh:

• local_subscribers(router, msg.twin, msg): run any Julia callback bound directly on this router for msg.topic (exact or glob match);
• broadcast_msg(router, msg): put! the message onto the inbox of every twin in router.topic_interests[msg.topic] (plus glob subscribers), which includes twins that are themselves neighbor brokers — causing pubsub_msg to run again on the next hop and the message to keep walking the mesh towards the real subscribers. See the Mesh Routing and Forwarding guide.

broadcast_msg is skipped unless relay_rule(msg.twin.uid) holds: a server (see server_task) relays only the messages it publishes itself.

Returns false (without delivering) if the permission/authorization checks fail, true otherwise.

put(rb, topic::AbstractString, data...; qos=Rembus.QOS0)

Publish one or more data values to a channel identified by topic, which is automatically prefixed with the component name.

Each element in data is passed as an argument to the remote method subscribed to topic.

The data arguments may be of any type. For interoperability across components implemented in different languages, the values should be either DataFrames or primitive types that are CBOR-encodable (see RFC 8949).

The keyword argument qos specifies the quality of service (QoS) for message delivery:

• Rembus.QOS0 (default): At most once delivery. A message may be delivered zero or one time.
• Rembus.QOS1: At least once delivery. A message is delivered one or more times.
• Rembus.QOS2: Exactly once delivery. A message is delivered once and only once.

Examples

rb = component("myname")
put(rb, "mytopic", 1, 2)

This publishes the values 1 and 2 to the topic myname/mytopic.

reactive(rb, from::Union{Real,Period,Dates.CompoundPeriod}=Day(1))

Enable the reception of published messages for topics to which the node is subscribed via subscribe.

The from argument specifies the starting point in time from which messages published while the component was offline will be delivered upon reconnection.

This value applies to all subscribed topics but can be overridden by the from argument in the subscribe method for a specific topic — though only to define narrower time ranges.

Possible from values:

• Rembus.Now: Equivalent to 0.0, ignores old messages, and starts receiving only new messages from now.
• Rembus.LastReceived: Receives all messages published since the last disconnection.
• n::Float64: Receives messages published within the last n seconds.
• Dates.CompoundPeriod: Defines a custom period using a CompoundPeriod value.

Example

rb = component("myname")
subscribe(rb, "mytopic1", Rembus.Now)
subscribe(rb, "mytopic2", Rembus.LastReceived)
subscribe(rb, "mytopic3", Hour(1))

reactive(rb, Day(1))

In this example:

• mytopic1 will receive messages starting from now.
• mytopic2 will receive messages published within the last day, even if subscribe() uses Rembus.LastReceived.
• mytopic3 will receive messages published within the last hour.

reconnect(twin::Twin, url::RbURL)

Attempt to reconnect to the given url.

The url is the primary url of the twin or one of its failover urls.`

On a successful do_connect, re-synchronizes mesh state with twin_setup (sends the explicit SETUP_CMD admin command with this node's current exports, see admin_command): since a reconnection may land on a broker that lost all prior incremental subscribe/expose state for this link, the full local configuration is replayed in one shot rather than relying on the one-time attestation exchange used for a fresh connection (see authenticate). If this re-sync fails, the connection is torn down again rather than left in an inconsistent state.

If this twin exposes listener ports (it is itself a broker) and the new connection succeeded, any other currently-connected node is disconnected too, forcing them to reconnect to the (possibly new) primary broker rather than staying attached to a stale failover.

reconnect(twin::Twin)

Reconnection loop driver: repeatedly try twin.failovers in order (sleeping reconnect_period between attempts) via reconnect(twin, url) until one succeeds or the twin is closing. Installed as the HR_CONN_DOWN handler in handle_connection, it is what makes a mesh link self-heal (and re-run mesh table synchronization) after a transient network failure.

register(
    cid::AbstractString,
    pin::AbstractString;
    scheme::UInt8
)

Register the component with name cid.

To register a component a single pin or a set of tenants must be configured in the tenants.json file.

The pin shared secret is a 8 hex digits string (for example "deedbeef"). It is not sent on the wire: the Register message carries SHA256(CBOR([pin, cid])).

remove_twin(router::Router, twin::Twin)

Remove twin from router.id_twin, dropping it from the set of direct mesh neighbors reachable from router (it will no longer be a candidate for admin_broadcast, direct-RPC target resolution, or any other neighbor lookup). Does not touch topic_impls/topic_interests; that cleanup is cleanup's responsibility.

request_timeout!(rb, value::Real)

Set the request timeout value for the component rb.

request_timeout(rb)

Get the request timeout value for the component rb.

rid(rb::Rembus.Twin) -> String

Return the name of the component (Rrembus IDentifier).

rb = component("ws://myhost.org:8338/myname")
rid(rb) === "myname"

rpc(rb, service::AbstractString, data...)

Make a request for a remote service method using Vararg data values as method arguments.

Return a value or throw an error if a request timeout occurs or the remotely invoked method throws an exception.

Example

rb = component("myclient")
# invoke mysum(1, 2) on the remote site
result = rpc(rb, "mysum", 1, 2)

rpc_request(router::Router, msg, implementor_rule)

Handle an inbound RpcReqMsg, after the command_permitted/ isauthorized checks:

• Direct RPC (msg.target !== nothing, i.e. the caller named a specific component or group): resolve msg.target against router.id_twin — either an exact id match, or a "<group>@..." pooled-member match (in which case select_twin applies the router's load-balancing policy among the matching members) — and delegate delivery to manage_target. STS_TARGET_NOT_FOUND is replied if no matching twin is connected here.
• Routable RPC (msg.target === nothing): try the router's own local_fn first; otherwise, if implementor_rule(twin.uid) allows it (used to prevent re-forwarding loops/policy restrictions upstream), call find_implementor(router, msg) to pick a connected exposer — possibly itself a neighbor broker, in which case this same function runs again there, hopping the request towards the real implementor across the mesh.

Returns msg unchanged (the various outcomes are delivered as side effects by put!ing responses/forwarded requests onto the relevant twins' inboxes).

send_data_at_rest(twin::Twin, max_period::Float64, con::DuckDB.DB

Send persisted and cached messages.

max_period is a time barrier set by the reactive command.

Valid time period for sending old messages: [min(twin.topic.msg_from, max_period), now_ts]

send_data_at_rest(twin::Rembus.Twin, max_period::Float64, ::FileStore

Send persisted and cached messages.

max_period is a time barrier set by the reactive command.

Valid time period for sending old messages: [min(twin.topic.msg_from, max_period), now_ts]

sendto_origin(twin, msg)

Short-circuit a response that targets a locally-pending direct request, without going through the full broadcast_msg/mesh-relay path.

If twin.socket.direct has a future registered under msg.id (i.e. this twin itself issued the matching direct request — such as an admin command or a direct RPC call awaiting synchronous completion), fulfill that future with msg, cancel its timeout timer and return true. Returns false when there is no such pending future, meaning the caller (respond, admin_broadcast) must fall back to delivering msg through the normal inbox/broadcast mechanism so it keeps propagating towards its real destination across the mesh.

Method-specific overloads exist for FutureResponse (always false, nothing to resolve) and WsPing (sends a WebSocket ping frame instead of resolving a future).

server(; <keyword arguments>)

Start a server node and return a handle for interacting with it.

The server accepts connection from client nodes.

It supports multiple communication protocols (WebSocket, TCP, and ZMQ) and allows for customizable security and authentication.

Keyword arguments

• name::AbstractString="broker": The unique identifier for the server supervisor process.
• ws=nothing: The WebSocket (ws/wss) listening port. Set to nothing to disable.
• tcp=nothing: The TCP (tcp/tls) listening port. Set to nothing to disable.
• zmq=nothing: The ZMQ Router listening port. Set to nothing to disable.
• prometheus=nothing: The Prometheus port for scraping monitoring metrics. Set to nothing to disable.
• secure=false: If true, enables WSS (WebSocket Secure) and TLS (Transport Layer Security) protocols for encrypted connections.
• authenticated=false: If true, only allows connections from named and authenticated nodes.

Default Behavior

If ws, tcp, and zmq are all set to nothing, the broker will default to listening for WebSocket connections on port 8338.

setidentity(router, twin, msg; isauth=false)

Update twin identity parameters.

Re-keys router.id_twin from the (likely ephemeral) previous id to rid(twin) after twin.uid is set from msg.cid, so the mesh's neighbor-lookup table (id_twin, used by admin_broadcast and direct-RPC target resolution) reflects the component's definitive name as soon as it authenticates.

settable!(router, df)

Given a router with defined tables and a DataFrame df containing a :topic column, this function matches each topic against the router's table patterns and assigns the corresponding table name and pattern to new :table and :regexp columns in df. If no pattern matches, the topic name itself is used as the table name and nothing for the pattern.

shutdown_broker(router)

Administration action for SHUTDOWN_CMD: shut down the broker's Visor supervisor tree, terminating the broker process. Invoked asynchronously by admin_command so the STS_SUCCESS response can still be sent back to the caller before the process goes down.

start_broker(;
    wait=true,
    secure=nothing,
    ws=nothing,
    tcp=nothing,
    zmq=nothing,
    http=nothing,
    prometheus=nothing,
    name="broker",
    authenticated=false,
    reset=nothing,
    plugin=nothing,
    context=nothing
)

Start the node.

Return immediately when wait is false, otherwise blocks until shutdown is requested.

Overwrite command line arguments if args is not empty.

subscribe(rb, fn::Function, from=Rembus.Now)

Subscribe to messages published to a topic and register a callback function fn to handle incoming messages.

If the topic argument is omitted, the function name must be equal to the topic name.

The from (default=Rembus.Now) argument defines the starting point in time from which messages published while the component was offline will be sent upon reconnection. Possible from values:

• Rembus.Now: Equivalent to 0.0, ignores old messages, and starts receiving only new messages from now.
• Rembus.LastReceived: Receives all messages published since the last disconnection.
• n::Float64: Receives messages published within the last n seconds.
• Dates.CompoundPeriod: Defines a custom period using a CompoundPeriod value.

To enable the reception of published messages, the reactive function must be called.

Example

rb = component("myname")

# Define a callback function for the "mytopic" topic
function mytopic(data)
    println("Received: ", data)
end

# Subscribe to "mytopic", receiving only new messages from now
subscribe(rb, mytopic, from=Rembus.Now)
reactive(rb)

subscribe(rb, topic::AbstractString, fn::Function, from=Rembus.Now)

topic_impls(router::Router, target::Twin)

Compute the list of topics target should be told this router implements (or knows an implementor for): the router's own local Julia methods (router.local_function, excluding subscribers and built-ins) unioned, via _topics, with every topic in router.topic_impls that has an implementor other than target itself.

Used to build the exports sent back in the attestation response — see get_topics and attestation.

topic_interests(router::Router, target::Twin)

Compute the list of topics target should be told this router (or some twin reachable through it) is subscribed to: the router's own local subscriber callbacks (router.local_function filtered by router.local_subscriber) unioned, via _topics, with every topic in router.topic_interests that has a subscriber other than target itself.

Used to build the exports sent back in the attestation response — see get_topics and attestation.

unauthorize(router, twin, msg)

Administration command handler for UNAUTHORIZE_CMD: revokes the grant previously given with authorize for the component identified by msg.data[CID] on msg.topic. Requires twin to be an admin and msg.data[CID] to be a non-empty component id.

function unauthorize(rb, client::AbstractString, topic::AbstractString)

Revoke authorization to the client component for use of the private topic.

The component must have the admin role for revoking topic accessibility.

unexpose(rb, service::AbstractString)

Stop servicing RPC requests targeting service.

unexpose(rb, fn::Function)

Stop servicing RPC requests targeting fn function.

unreactive(rb)

Stops the delivery of published messages to the rb component.

unregister(twin)

Unregister the connected component.

Only a connected and authenticated component may execute the unregister command.

using Rembus

twin = connect("authenticated_component")
Rembus.unregister(twin)
close(twin)

unsubscribe(rb, topic::AbstractString)

Stops delivering messages published on the specified topic to the rb component.

unsubscribe(rb, fn::Function)

Stops delivering messages to the specified fn function.

update_tables(router::Router, twin::Twin, exports)

Register the peer's already-known topics into the local mesh routing tables when a connection is (re)established.

exports is the two-element list [exposers, subscribers] returned by the peer's get_topics (via the attestation response, see authenticate): exports[1] are topics the peer implements, exports[2] are topics the peer (or some twin reachable through it) is subscribed to. For each, twin is added to router.topic_impls/ router.topic_interests, so this router can immediately forward RPC requests and Pub/Sub messages towards the peer without waiting for individual expose/subscribe admin commands to propagate.

A no-op if exports is nothing or ismultipath is false.

shutdown(rb::Rembus.Twin)

Close the connection and terminate the component.

@component "url"

Set the name of the component and the protocol for connecting to the broker.

url may be:

• "myname": use $REMBUS_BASE_URL for connection parameters
• "tcp://host:port/myname": tcp connection
• "ws://host:port/myname": web socket connection
• "zmq://host:port/myname": ZeroMQ connection

@expose function fn(arg1,...)
    ...
end

Expose the function expression.

@expose fn

Expose all the methods of the function fn.

Example

Expose the function mycalc that implements a service that may accept two numbers or a string and number:

mycalc(x::Number, y::Number) = x+y
mycalc(x::String, y::Number) = length(x)*y

@expose mycalc

Call mycal service using the correct types of arguments:

# ok
julia> response = @rpc mycalc(1,2)
0x0000000000000003

# ok
julia> response = @rpc mycalc("hello",2.0)
10.0

If the RPC client call mycalc with the argument's type that do not respect the signatures of the exposed service then it throws RpcMethodException

julia> response = @rpc mycalc("hello","world")
ERROR: RpcMethodException("rembus", "mycalc", "MethodError: no method matching mycalc(::String, ::String) ...

 @inject container

Binds a container object, which is passed as the first argument to subscribed component functions.

See inject for more details.

@publish topic(arg1,arg2,...)

Publish a message to topic logic channel.

The function topic(arg1,arg2,...) will be called on each connected component subscribed to topic.

Publisher

@publish foo("gfr", 54.2)

Subscriber

function foo(name, value)
    println("do something with $name=$value")
end

@subscribe foo
@reactive

supervise()

@reactive

The subscribed methods start to handle published messages.

@rpc service(arg1,...)

Call the remote service method and return its outcome.

The outcome may be the a return value or a RpcMethodException if the remote throws an exception.

The service method must match the signature of an exposed remote service method.

Components may subscribe to service for receiving the service request.

Exposer

function mymethod(x, y)
    return evaluate(x,y)
end

@expose mymethod
supervise()

RPC client

response = @rpc mymethod(x,y)

Subscriber

function service(x, y)
    ...
end

@subscribe service
@reactive

supervise()

@subscribe topic [mode]

Setup a subscription to topic logic channel to handle messages from @publish or @rpc.

mode values`:

• from_now (default): receive messages published from now.
• before_now: receive messages published when the component was offline.

Messages starts to be delivered to topic when reactivity is enabled with @reactive macro.

Subscriber

function foo(arg1, arg2)
    ...
end

@subscribe foo
@reactive

supervise()

Publisher

@publish foo("gfr", 54.2)

@subscribe function fn(args...)
    ...
end [mode]

Subscribe the function expression.

@unexpose fn

The methods of fn function is no more available to rpc clients.

@unreactive

The subscribed methods stop to handle published messages.

@unsubscribe mytopic

mytopic's methods stop to handle messages published to topic mytopic.

@wait

Block forever waiting for Ctrl-C/InterruptException or root supervisor shutdown.