Flowmaker SDK [beta]Docs
search
GitHub

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 C# side the entry point is FlowBoxWorkerBridge (namespace Industream.FlowMaker.Sdk.WorkerBridge), the direct port of the TypeScript SDK's FlowBoxWorkerBridge.

[NOTE!lightbulb/SCOPE] This page covers the worker side (C#). 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.
  • The requestId is echoed back unchanged, so concurrent requests stay isolated.

Registering the bridge

Create a bridge, register one handler per box version, and hand it to the client with UseWorkerBridge. The real cs-data-logger worker does exactly this in Program.cs:

using Industream.FlowMaker.Sdk.Clients;
using Industream.FlowMaker.Sdk.WorkerBridge;

var bridge = new FlowBoxWorkerBridge(host.Services.GetRequiredService<ILoggerFactory>());
bridge.RegisterHandler($"industream/data-logger/{version}", DataLoggerConfigurationHandler.Create().Build());

await host.Services
    .GetRequiredService<FlowMakerClientBuilder>()
    .DeclareFlowElement(CsDataLoggerBox.GetRegistrationInfo())
    .UseWorkerBridge(bridge)
    .Build()
    .Run();

The bridge constructor is FlowBoxWorkerBridge(ILoggerFactory? loggerFactory = null, string? socketUrlPrefix = null, TimeSpan? handlerTimeout = null). The Socket.IO base URL defaults to FM_WORKER_LOG_SOCKET_IO_ENDPOINT (or http://localhost:4123); the handler timeout defaults to 30 seconds.

[INFO!info/AUTO CONNECT] You never call bridge.Connect(...) yourself. After the worker registers with the runtime, the client calls bridge.Connect(workerId, boxVersionIds) automatically, using the FM_WORKER_ID environment variable as workerId. Connect is only public for advanced or test scenarios.

Handling events

A handler receives a WorkerRequest and returns the response payload (or throws). WorkerRequest exposes RequestId, BoxVersionId, Event, Payload (a System.Text.Json.JsonElement), and DeserializePayload<T>(JsonSerializerOptions? options = null) which defaults to JsonSerializerOptions.Web. There are three ways to register a handler for a box version.

(a) Typed router with WorkerConfigHandlers (recommended)

WorkerConfigHandlers is a fluent per-event router. Map each event name to a typed handler, then pass .Build() to RegisterHandler. A malformed payload is returned as INVALID_PAYLOAD, an unregistered event as UNKNOWN_EVENT.

using System.Text.Json;
using Industream.FlowMaker.Sdk.WorkerBridge;

public static class DataLoggerConfigurationHandler
{
    public static WorkerConfigHandlers Create()
        => new WorkerConfigHandlers()
            .On<JsonElement, string>("data-logger:double", (payload, _) =>
            {
                var raw = payload.TryGetProperty("value", out var valueElement) ? valueElement.ToString() : null;
                if (!double.TryParse(raw, out var value))
                {
                    throw new WorkerBridgeException($"Invalid number: {raw}");
                }

                return ValueTask.FromResult($"your doubled number is : {value * 2}");
            });
}

The available overloads are On<TIn, TOut>(@event, Func<TIn, CancellationToken, ValueTask<TOut>>), On<TIn, TOut>(@event, Func<TIn, ValueTask<TOut>>), and On<TOut>(@event, Func<CancellationToken, ValueTask<TOut>>) for events with no payload.

Prefer a typed input record when you want a clean contract and are happy with the SDK's generic error on bad JSON:

public record DoubleInput(double Value);

new WorkerConfigHandlers()
    .On<DoubleInput, string>("data-logger:double", (input, _) =>
        ValueTask.FromResult($"your doubled number is : {input.Value * 2}"));

Here a payload that is not { "value": <number> } fails deserialization and the router surfaces INVALID_PAYLOAD. Use the loose JsonElement form (as the data-logger does) only when you need a custom error message or partial parsing; use a record otherwise.

(b) Class handler with RegisterClass<T>

Implement IWorkerMessageHandler and register the type. The bridge creates exactly one instance per box version and reuses it for all requests.

using Industream.FlowMaker.Sdk.WorkerBridge;

public sealed class DataLoggerHandler : IWorkerMessageHandler
{
    public ValueTask<object?> HandleAsync(WorkerRequest request, CancellationToken cancellationToken = default)
    {
        var input = request.DeserializePayload<DoubleInput>();
        return ValueTask.FromResult<object?>($"your doubled number is : {input!.Value * 2}");
    }
}

bridge.RegisterClass<DataLoggerHandler>($"industream/data-logger/{version}");

(c) Raw handler delegate

For full control, register the erased delegate directly. This is the same shape .Build() produces, useful when you already have a dispatch function.

bridge.RegisterHandler($"industream/data-logger/{version}", (request, cancellationToken) =>
{
    var input = request.DeserializePayload<DoubleInput>();
    return ValueTask.FromResult<object?>($"your doubled number is : {input!.Value * 2}");
});

Error handling

Throw a WorkerBridgeException(message, code) from any handler to return a typed error. The bridge catches it, puts message and code in the response's Error, and the platform rejects the Svelte-side request() promise with that message. Any other exception is reported as HANDLER_ERROR.

throw new WorkerBridgeException("Connection refused by device", WorkerBridgeErrorCodes.HandlerError);

The reserved error codes are defined on WorkerBridgeErrorCodes:

Code Constant When
UNKNOWN_BOX UnknownBox boxVersionId is not served by this worker
NO_HANDLER NoHandler Box is served but has no handler registered
TIMEOUT Timeout Handler exceeded the configured timeout (default 30s)
HANDLER_ERROR HandlerError Handler threw a non-WorkerBridgeException
INVALID_PAYLOAD InvalidPayload Payload failed to deserialize into TIn
UNKNOWN_EVENT UnknownEvent Event not registered on the WorkerConfigHandlers router

The :health event

The bridge answers :health 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 (see WorkerBridgeEvents.HealthEvent). A configuration bundle can always ping it to check that a matching worker is live.

Full worked example

The two pieces from cs-data-logger together: the typed router and the wiring.

// Boxes/DataLoggerConfigurationHandler.cs
using System.Text.Json;
using Industream.FlowMaker.Sdk.WorkerBridge;

namespace CsDataLogger.Boxes;

public static class DataLoggerConfigurationHandler
{
    public static WorkerConfigHandlers Create()
        => new WorkerConfigHandlers()
            .On<JsonElement, string>("data-logger:double", (payload, _) =>
            {
                var raw = payload.TryGetProperty("value", out var valueElement) ? valueElement.ToString() : null;
                if (!double.TryParse(raw, out var value))
                {
                    throw new WorkerBridgeException($"Invalid number: {raw}");
                }

                return ValueTask.FromResult($"your doubled number is : {value * 2}");
            });
}

// Program.cs
var bridge = new FlowBoxWorkerBridge(host.Services.GetRequiredService<ILoggerFactory>());
bridge.RegisterHandler($"industream/data-logger/{version}", DataLoggerConfigurationHandler.Create().Build());

await host.Services
    .GetRequiredService<FlowMakerClientBuilder>()
    .DeclareFlowElement(CsDataLoggerBox.GetRegistrationInfo())
    .UseWorkerBridge(bridge)
    .Build()
    .Run();

Common Pitfalls

Registering the same box version twice

// !!! WRONG !!! : Two handlers for the same box version
bridge.RegisterHandler("myorg/my-box/1.0.0", firstHandler);
bridge.RegisterHandler("myorg/my-box/1.0.0", secondHandler); // InvalidOperationException

// +++ CORRECT +++ : One handler per box version, route events inside it
var router = new WorkerConfigHandlers()
    .On<A, RA>("my-box:a", HandleA)
    .On<B, RB>("my-box:b", HandleB);
bridge.RegisterHandler("myorg/my-box/1.0.0", router.Build());

[ERROR!error/DUPLICATE REGISTRATION] RegisterHandler throws InvalidOperationException if a handler is already registered for the box version. Use a single WorkerConfigHandlers router to fan out multiple events for one box.

Blocking the handler

// !!! WRONG !!! : Blocking call inside the handler
.On<JsonElement, string>("my-box:load", (payload, _) =>
{
    var data = httpClient.GetString(url).Result; // blocks a thread up to the timeout
    return ValueTask.FromResult(data);
})

// +++ CORRECT +++ : Await asynchronously and honor the token
.On<JsonElement, string>("my-box:load", async (payload, ct) =>
{
    var data = await httpClient.GetStringAsync(url, ct);
    return data;
})

[WARNING!warning/HANDLER TIMEOUT] Handlers run off the Socket.IO receive loop with a default 30-second timeout. A blocking call (.Result, .Wait(), sync I/O) wastes a thread and eventually returns TIMEOUT. Use async/await and pass the CancellationToken through.

Forgetting UseWorkerBridge

// !!! WRONG !!! : Bridge built but never attached — it never connects
var bridge = new FlowBoxWorkerBridge(loggerFactory);
bridge.RegisterHandler("myorg/my-box/1.0.0", handler);
await clientBuilder.DeclareFlowElement(registration).Build().Run();

// +++ CORRECT +++ : Attach the bridge to the client
await clientBuilder
    .DeclareFlowElement(registration)
    .UseWorkerBridge(bridge)
    .Build()
    .Run();

[ERROR!error/BRIDGE NEVER CONNECTS] Without UseWorkerBridge(bridge), the client 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
.On<JsonElement, object>(":health", (_, __) => ValueTask.FromResult<object?>(new { status = "ok" }));

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

[NOTE!sync/RESERVED EVENT] :health is answered by the bridge before your handler runs. Registering it in a WorkerConfigHandlers router has no effect and only adds dead code.