C# / .NET SDK (`Industream.FlowMaker.Sdk`)

Package: sdk/dotnet/Industream.FlowMaker.Sdk.

Overview

The C# SDK provides the runtime infrastructure for building FlowBox workers, stateless or stateful components that process data streams in the FlowMaker platform. It targets .NET 10 and uses NetMQ, System.Reactive, and MessagePack.

[INFO!hub/SDK SCOPE] This SDK handles the runtime loop, serialization, and logging. Your FlowBox implementations focus on business logic: transforming data, calling external APIs, or managing state.

Quick Start

Create a minimal worker with a single pipe:

using Industream.FlowMaker.Sdk.Clients;
using Industream.FlowMaker.Sdk.Elements.Advanced;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;

var services = new ServiceCollection();
services.AddLogging();
services.AddSingleton<IFlowMakerLogger, FlowMakerLogger>();
services.AddSingleton<SocketIOFactory>();
var serviceProvider = services.BuildServiceProvider();

var clientBuilder = new FlowMakerClientBuilder(serviceProvider)
    .SetRouterTransportAddress("tcp://*:5560")
    .SetRuntimeHttpAddress("http://localhost:3120")
    .SetLoggerAddress("http://localhost:3040");

var registration = new RegistrationInfoBuilder(FlowElementType.Pipe)
    .SetId("myorg/message-counter")
    .SetDisplayName("Message Counter")
    .SetCurrentVersion("1.0.0")
    .SetInputs(new[] { new TypeDefinition("input") })
    .SetOutputs(new[] { new TypeDefinition("output") })
    .Build((initParams, sp) => new MessageCounter(initParams, sp));

clientBuilder.DeclareFlowElement(registration);
var client = clientBuilder.Build();
await client.Run();

public class MessageCounter : FlowBoxCore
{
    private int _counter;
    
    public MessageCounter(FlowBoxInitParams initParams, IServiceProvider serviceProvider)
        : base(initParams, serviceProvider) { }
    
    protected override async Task OnInput(byte[] inputId, Header header, byte[] data)
    {
        _counter++;
        var serializer = Serialization.GetSerializer(header);
        var input = serializer.Deserialize<Dictionary<string, object>>(data);
        input["counter"] = _counter;
        await Push("output", header, serializer.Serialize(input));
    }
}

[NOTE!lightbulb/START HERE] This is the minimal working example. Copy it, modify the registration and OnInput() logic, and you have a custom worker.

Documentation Pages

FlowBox Implementations — Build Source, Pipe, and Sink nodes with real examples
FlowBoxCore and FlowBoxRaw — Base classes, hierarchy, and implementation patterns
FlowBoxLogger API — Logging and socket management
FlowBoxSerializer API — Serialization with MessagePack/JSON and header handling

Runtime Architecture & Data Flow

[NOTE!lightbulb/DIAGRAM CONTEXT] This diagram shows a typical three-stage pipeline. Data flows down (Source → Pipe → Sink) via NetMQ frames with headers. Logs flow right to the Logger via Socket.IO. Control events flow to the Scheduler via NetMQ.

NetMQ/Frames

Socket.IO Events

Key points:

Source: Emits data via OnOutputReady() when downstream signals CAN_SEND_NEXT
Pipe: Receives via OnInputReceived(), transforms, pushes downstream
Sink: Terminal node, receives data without pushing
Logger: Socket.IO transport for all log messages
Scheduler: Manages flow lifecycle, sends control events

Data Shape Convention

{
  "temperature": 21.7,
  "pressure": 1.8,
  "$$meta": {
    "temperature": { "entryId": "dc-entry-temp" },
    "pressure": { "entryId": "dc-entry-pressure" }
  },
  "$$timestamp": "2026-04-10T16:00:00.000Z"
}

Root keys: measured tag values
$$meta: per-tag metadata (e.g., DataCatalog entry IDs)
$$timestamp: event timestamp in ISO 8601 format

Registration Protocol

FlowBox registrations are sent to the runtime as MessagePack maps. The SDK uses the protocol field names shown below; do not rely on CLR member names or numeric enum values.

Registration area	MessagePack fields
Identity	`id`, `displayName`, `currentVersion`, `type`, `icon`, `stateless`, `agent`
Worker and UI	`workerId`, `workerDataConnectionString`, `uiConfig` (`type`, `defaultOptions`, `implementation`)
I/O	`ioInterfaces` (`inputs`, `outputs`, `allowInputOverride`, `allowOutputOverride`)
Extension metadata	`documentation` (`summary`, `brief`, `detail`), `codePlug`

FlowElementType and SerializationMethods use lowercase wire strings: sink, pipe, source, msgpack, and json. SetJavaScriptBundle() sets uiConfig.type to js-bundle and base64-encodes the bundle; SetCodePlug() base64-encodes its source. Override fields and TypeDefinition.DisallowOverride are optional and remain unset unless configured.

var registration = new RegistrationInfoBuilder(FlowElementType.Pipe)
    .SetJavaScriptBundle("export default {}")
    .SetCodePlug("console.log('ready')")
    .SetAllowInputOverride()
    .SetAllowOutputOverride(false)
    .Build((initParams, services) => new MessageCounter(initParams, services));

Lifecycle And Optional Health

After the runtime creates a FlowBox, it awaits InitializeAsync() before acknowledging the initialization request. Override it for asynchronous setup such as opening a device connection. The base FlowBoxRaw.InitializeAsync() implementation completes immediately, so it is safe to omit when no setup is needed.

public override async Task InitializeAsync()
{
    await _device.ConnectAsync();
}

The SDK does not start HTTP health endpoints, metrics, hosting, configuration loading, or a dependency-injection container. Workers that need the retained health feature explicitly register it while composing their own services:

services.AddFlowMakerHealthChecks("my-worker", "My FlowMaker Worker");

This opt-in feature reads FM_HEALTH_CHECK_PORT (or HEALTH_CHECK_PORT) when configured through its environment-based overload. It is separate from the FlowMaker runtime registration and lifecycle.

Build and packaging

From sdk/dotnet:

dotnet restore Industream.FlowMaker.Sdk.slnx
dotnet build Industream.FlowMaker.Sdk.slnx -c Release
dotnet test Industream.FlowMaker.Sdk.slnx -c Release

Target framework and language version:

TargetFramework: net10.0
LangVersion: 13.0
Solution: Industream.FlowMaker.Sdk.slnx

References

sdk/dotnet/Industream.FlowMaker.Sdk/Industream.FlowMaker.Sdk.csproj
sdk/dotnet/Industream.FlowMaker.Sdk/Clients/FlowMakerClient.cs
sdk/dotnet/Industream.FlowMaker.Sdk/Clients/FlowBoxRegistrationInfo.cs
sdk/dotnet/Industream.FlowMaker.Sdk/Clients/Header.cs
sdk/dotnet/Industream.FlowMaker.Sdk/Serializations/Serialization.cs
sdk/dotnet/Industream.FlowMaker.Sdk/Elements/Advanced/FlowBoxCore.cs
sdk/dotnet/Industream.FlowMaker.Sdk/Elements/Simple/FlowBoxCore.cs
sdk/dotnet/release-notes.md

C# / .NET SDK (Industream.FlowMaker.Sdk)