Worker Messages (Configuration RPC)

A FlowBox configuration UI often needs to talk to the running worker: validate a connection string, fetch a list of topics, or echo a value back to the user. Worker Messages is the request/response channel between the configuration microfrontend and a live worker instance. On the TypeScript side the entry point is FlowBoxWorkerBridge, exported by @industream/flowmaker-sdk (source sdk/typescript/src/flow-box-worker-bridge.ts); the C# FlowBoxWorkerBridge is its direct port.

[NOTE!lightbulb/SCOPE] This page covers the worker side (TypeScript). The Svelte side—how a configuration bundle sends a request and how the platform correlates it—is documented in Worker Messages from the Configuration UI. Read that first if you have not.

Request and response flow

Worker Messages request and response flow

The platform UI owns authentication, worker selection, request IDs, timeouts and cleanup.
The worker only sees a WorkerRequest on a Socket.IO namespace and answers with a matching WorkerResponse.
The bridge dispatches each request to the handler registered for its boxVersionId — format "<boxId>/<version>", e.g. industream/data-logger/1.0.0, built by the SDK as `${registration.id}/${registration.currentVersion}`.
The requestId is echoed back unchanged, so concurrent requests stay isolated.
The wire events are the exported constants WORKER_REQUEST_EVENT ("worker:request") and WORKER_RESPONSE_EVENT ("worker:response"); you never emit or listen to them yourself.

Registering the bridge

Create a bridge, register one handler per box version, and pass the bridge as the fourth argument to FlowBoxEventLoopHandler:

import { FlowBoxEventLoopHandler, FlowBoxFactory, FlowBoxWorkerBridge } from "@industream/flowmaker-sdk";

const workerBridge = new FlowBoxWorkerBridge();
workerBridge.registerClass("myorg/my-pipe/1.0.0", MyPipeConfigurationHandler);

const eventLoop = new FlowBoxEventLoopHandler(
    process.env["FM_ROUTER_TRANSPORT_ADDRESS"] ?? "tcp://*:5560",
    process.env["FM_RUNTIME_HTTP_ADDRESS"] ?? "http://localhost:3100",
    new FlowBoxFactory([MyPipe]),
    workerBridge,
);
await eventLoop.run();

The bridge constructor is new FlowBoxWorkerBridge(socketUrlPrefix?, options?). The Socket.IO base URL defaults to FM_WORKER_LOG_SOCKET_IO_ENDPOINT (or http://localhost:4123); options.timeoutMs defaults to 30000.

[INFO!info/AUTO CONNECT] You never call bridge.connect(...) yourself. After the worker registers with the runtime, FlowBoxEventLoopHandler.run() calls bridge.connect(workerId, boxVersionIds) automatically, using the workerId from the box registrations (conventionally process.env.FM_WORKER_ID), and calls bridge.disconnect() on shutdown. connect is only public for advanced or test scenarios.

Handling events

A handler receives the full WorkerRequest ({ requestId, boxVersionId, event, payload }) and returns the response payload, either synchronously or as a Promise. There are two registration shapes.

(a) Class handler with `registerClass` (recommended)

Implement FlowBoxWorkerHandler (a single handle(request) method) and register the class. The bridge instantiates it exactly once per box version and reuses that instance for all requests; registerClass returns the instance and is idempotent—calling it again for the same boxVersionId returns the existing instance instead of creating a second one.

import type { FlowBoxWorkerHandler, WorkerRequest } from "@industream/flowmaker-sdk";

class MyPipeConfigurationHandler implements FlowBoxWorkerHandler {
    async handle(request: WorkerRequest): Promise<unknown> {
        if (request.event === "configuration:sample") {
            return { options: request.payload, fields: [] };
        }

        throw new Error(`Unsupported worker event: ${request.event}`);
    }
}

const instance = workerBridge.registerClass("myorg/my-pipe/1.0.0", MyPipeConfigurationHandler);

(b) Raw function handler with `registerHandler`

For full control—or when the handler already exists as a closure—register a WorkerMessageHandler delegate directly:

workerBridge.registerHandler("myorg/my-pipe/1.0.0", (request) => {
    if (request.event === "my-pipe:double") {
        const { value } = request.payload as { value: number };
        return `your doubled number is : ${value * 2}`;
    }

    throw new Error(`Unsupported worker event: ${request.event}`);
});

Unlike C#, the TypeScript SDK ships no built-in per-event router (WorkerConfigHandlers): dispatch on request.event inside your single handler, as in both examples above. unregisterHandler(boxVersionId) removes a handler again and returns whether one was removed.

Error handling

Throw from any handler to return a typed error. The bridge catches it and puts error.message and error.code in the response's error; the platform rejects the Svelte-side request() promise with that message. An error without a code is reported as HANDLER_ERROR. To send a custom code, assign one to the thrown error:

throw Object.assign(new Error("Connection refused by device"), { code: "DEVICE_REFUSED" });

The bridge-reserved error codes are:

Code	When
`UNKNOWN_BOX`	`boxVersionId` is not served by this worker
`NO_HANDLER`	Box is served but its handler was removed after connect
`TIMEOUT`	Handler exceeded the configured timeout (default 30s)
`HANDLER_ERROR`	Handler threw an error without a `code` (default)

[WARNING!warning/SILENT DROPS] A request whose requestId, boxVersionId, or event is missing or empty is dropped without any response—the UI-side promise can only time out. Requests arriving while the socket is disconnected are queued and flushed on (re)connect, so brief reconnects do not lose calls.

The `:health` event

The bridge answers :health (exported as WORKER_HEALTH_EVENT) with { status: "ok" } for every registered box version, before dispatching to any handler.

[NOTE!info/AUTO HEALTH] Do not implement :health yourself. It is answered automatically by FlowBoxWorkerBridge. A configuration bundle can always ping it to check that a matching worker is live.

Full worked example

The pattern from the SDK's own reference example: the box, its configuration handler class, and the wiring.

// my-pipe.ts
import { FlowBoxCore, FlowBoxRegistration } from "@industream/flowmaker-sdk";
import process from "process";

@FlowBoxRegistration({
    id: "myorg/my-pipe",
    displayName: "My Pipe",
    currentVersion: "1.0.0",
    type: "pipe",
    icon: "transform",
    stateless: true,
    agent: "flowmaker-node/1.0.0",
    workerId: process.env.FM_WORKER_ID!,
    workerDataConnectionString: process.env.FM_WORKER_TRANSPORT_ADV_ADDRESS!,
    ioInterfaces: { inputs: [], outputs: [] },
    uiConfig: { defaultOptions: {}, implementation: "svelte-bundle" },
})
export class MyPipe extends FlowBoxCore {
    protected async onInput(inputId, header, data) {
        // Normal runtime flow processing happens here.
    }
}

// my-pipe-configuration.ts
import type { FlowBoxWorkerHandler, WorkerRequest } from "@industream/flowmaker-sdk";

export class MyPipeConfigurationHandler implements FlowBoxWorkerHandler {
    async handle(request: WorkerRequest): Promise<unknown> {
        if (request.event === "configuration:sample") {
            return { options: request.payload, fields: [] };
        }

        throw new Error(`Unsupported worker event: ${request.event}`);
    }
}

// index.ts
import { FlowBoxEventLoopHandler, FlowBoxFactory, FlowBoxWorkerBridge } from "@industream/flowmaker-sdk";
import { MyPipe } from "./my-pipe.js";
import { MyPipeConfigurationHandler } from "./my-pipe-configuration.js";

const workerBridge = new FlowBoxWorkerBridge();
workerBridge.registerClass("myorg/my-pipe/1.0.0", MyPipeConfigurationHandler);

const eventLoop = new FlowBoxEventLoopHandler(
    process.env["FM_ROUTER_TRANSPORT_ADDRESS"] ?? "tcp://*:5560",
    process.env["FM_RUNTIME_HTTP_ADDRESS"] ?? "http://localhost:3100",
    new FlowBoxFactory([MyPipe]),
    workerBridge,
);
await eventLoop.run();

MyPipe is created per flow execution; MyPipeConfigurationHandler is created once for myorg/my-pipe/1.0.0 and reused for all configuration requests.

Common Pitfalls

Registering the same box version twice

// !!! WRONG !!! : Two handlers for the same box version
workerBridge.registerHandler("myorg/my-box/1.0.0", firstHandler);
workerBridge.registerHandler("myorg/my-box/1.0.0", secondHandler); // Error: Handler already registered

// +++ CORRECT +++ : One handler per box version, route events inside it
workerBridge.registerHandler("myorg/my-box/1.0.0", (request) => {
    const handlers = {
        "my-box:a": handleA,
        "my-box:b": handleB,
    } as const;
    const handler = handlers[request.event as keyof typeof handlers];
    if (!handler) throw new Error(`Unsupported worker event: ${request.event}`);
    return handler(request.payload);
});

[ERROR!error/DUPLICATE REGISTRATION] registerHandler throws if a handler is already registered for the box version. registerClass is idempotent and silently returns the existing instance—so if you expect two classes to replace each other, they will not. One handler per box version; fan out events inside it.

Never-resolving handlers

// !!! WRONG !!! : Handler awaits something that may never settle
workerBridge.registerHandler("myorg/my-box/1.0.0", async (request) => {
    const data = await someEventEmitterPromise; // no timeout, no abort
    return data;
});

// +++ CORRECT +++ : Bound the wait yourself and throw a typed error
workerBridge.registerHandler("myorg/my-box/1.0.0", async (request) => {
    const data = await withTimeout(someEventEmitterPromise, 5_000);
    return data;
});

[WARNING!warning/HANDLER TIMEOUT] Handlers race against the bridge timeout (options.timeoutMs, default 30 seconds). A handler that never settles returns TIMEOUT to the UI while your promise keeps running in the background. Keep handlers short and add your own timeouts for outbound I/O.

Forgetting to pass the bridge to the event loop

// !!! WRONG !!! : Bridge built but never attached — it never connects
const workerBridge = new FlowBoxWorkerBridge();
workerBridge.registerClass("myorg/my-box/1.0.0", MyHandler);
await new FlowBoxEventLoopHandler(routerAddress, runtimeUri, factory).run();

// +++ CORRECT +++ : Pass the bridge as the fourth constructor argument
await new FlowBoxEventLoopHandler(routerAddress, runtimeUri, factory, workerBridge).run();

[ERROR!error/BRIDGE NEVER CONNECTS] Without the bridge being handed to FlowBoxEventLoopHandler, run() never calls connect, so the worker never joins the /worker/<workerId> namespace and every request from the UI fails.

Manually implementing `:health`

// !!! WRONG !!! : Re-implementing the reserved health event
workerBridge.registerHandler("myorg/my-box/1.0.0", (request) => {
    if (request.event === ":health") return { status: "ok" };
    // ...
});

// +++ CORRECT +++ : Leave it alone — the bridge answers :health automatically
// (dispatch only your application events)

[NOTE!sync/RESERVED EVENT] :health is answered by the bridge before your handler runs. Handling it inside your handler is dead code.

Reference

Worker Messages from the Configuration UI (Svelte side)
C# worker messages guide (the ported API)
sdk/typescript/src/flow-box-worker-bridge.ts
sdk/typescript/src/flow-box-event-loop.class.ts