Skip to main content

@effectstream/event-server

Package: @effectstream/event-server · Source

A localhost-only MQTT broker built on opifex - the server side of EffectStream's event system. The runtime publishes block, transaction, primitive, and app events to this broker; frontends and workers subscribe via @effectstream/event-client.

  • Localhost-only MQTT broker (opifex) for Effectstream events.
  • Publishes from non-loopback connections are rejected at the broker.
  • Paired with @effectstream/event-client on the consuming side.
  • The engine starts a single broker ("effectstream-engine"). A batcher that enables its own event system runs a separate EventBroker instance in its own process.

Install​

bun add @effectstream/event-server
# or
npm install @effectstream/event-server

Standalone usage​

EventBroker is a thin wrapper around opifex that enforces "only localhost can publish". Drop it into any local-only pub/sub workflow without depending on the rest of EffectStream.

import { EventBroker } from "@effectstream/event-server";

// Ports are read from MQTT_ENGINE_BROKER_PORT / MQTT_ENGINE_BROKER_WS_PORT
// (or the _BATCHER_ equivalents when constructed with "Batcher").
const broker = new EventBroker("effectstream-engine");
await broker.start(); // listens on the configured TCP + WS ports

Once running, you can connect to it from @effectstream/event-client (or any MQTT client at mqtt://127.0.0.1:<configured-port>) to publish and subscribe to events.

Security: Publishes from non-loopback connections are rejected at the broker level. Subscriptions are unrestricted; gate them at the network layer if you don't want public consumers.

Inside EffectStream​

The runtime instantiates two EventBrokers (engine and batcher) so state-machine and batcher events flow on separate topic namespaces. Apps publish through EventManager.Instance.sendMessage(...) in @effectstream/event-client, which writes against the local broker exposed here.

Key exports​

  • EventBroker - broker class. Constructor takes "effectstream-engine" | "Batcher". Methods: start() (async; binds the configured TCP + WS ports), createServer() (fire-and-forget wrapper around start()), stop(). Clients publish and subscribe via the MQTT protocol, not direct class methods.

Examples​

Runnable: test/examples.test.ts.