FluxFlow.Engine 1.0.0

FluxFlow.Engine

A protocol-neutral, dataflow-based workflow engine for .NET.

Define graphs of typed processing nodes in JSON, build them into live TPL Dataflow networks, and drive them through a phase-ordered lifecycle — without any protocol-specific dependencies.

┌─────────────────────────────────────────────────────────────────────┐
│  ApplicationDefinition (JSON or code)                               │
│  ┌─────────────┐  ┌────────────────────────────────────────────┐   │
│  │  resources  │  │  workflows.main                             │   │
│  │  ──────────  │  │  ──────────────────────────────────────────│   │
│  │  shared     │  │  source ──→ filter ──→ mapper ──→ sink      │   │
│  └─────────────┘  └────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────┘
         │
         ▼  ApplicationRuntimeBuilder.Build()
┌─────────────────────────────────────────────────────────────────────┐
│  ApplicationRuntime (live TPL Dataflow graph)                       │
│  StateChanges  ──  ISourceBlock<ApplicationStateChanged>            │
│  Events        ──  ISourceBlock<FlowEvent>                          │
└─────────────────────────────────────────────────────────────────────┘

Features

Quick start

1. Implement a node

using FluxFlow.Engine.Components;
using FluxFlow.Engine.Runtime;

public sealed class NumberSource : SourceFlowNode<int>
{
    public static RuntimeNode Create(RuntimeNodeFactoryContext ctx)
    {
        var node = new NumberSource();
        return ctx.CreateNode(node)
            .Output("Output", node.OutputBlock)
            .Build();
    }

    public override async Task StartAsync(CancellationToken ct = default)
    {
        for (var i = 0; i < 10; i++)
            await SendOutputAsync(i, ct);

        CompleteOutput();
    }
}

Use SinkFlowNode<T>, TransformFlowNode<TInput,TOutput>, MapFlowNode<TInput,TOutput>, and EventFlowNodeBase when those shapes fit. Direct IFlowNode implementations still work when a node needs full control.

Nodes derived from FlowNodeBase can also emit operational diagnostics:

TryEmitDiagnostic(
    "demo.node.ready",
    FlowDiagnosticLevel.Information,
    "Node is ready.");

The runtime labels each diagnostic with the node address, id, phase, and type. Use diagnostics for health, status, counters, and live monitoring data. Use FlowEvent for workflow/domain activity.

Subscribe to FlowApplicationHost.Diagnostics before StartAsync when startup diagnostics matter:

var diagnostics = new BufferBlock<RuntimeFlowDiagnostic>();
host.Diagnostics.LinkTo(diagnostics, new DataflowLinkOptions { PropagateCompletion = true });

var result = await host.StartAsync();

2. Define the graph in JSON

{
  "workflows": {
    "main": {
      "source": { "type": "demo.numbers" },
      "printer": {
        "type": "demo.printer",
        "Input": { "from": "source.Output" }
      }
    }
  }
}

If no when expression is set, the link receives every item. If a definition uses when, the host must pass an IFlowExpressionEngine to ApplicationRuntimeBuilder or FlowApplicationHost.Create(...); the engine does not ship a concrete expression language.

3. Register, build, run

var registry = new RuntimeNodeFactoryRegistry();
registry.Register(new NodeType("demo.numbers"), NumberSource.Create);
registry.Register(new NodeType("demo.printer"),  PrinterNode.Create);

var json       = File.ReadAllText("flow.json");
var opts       = ApplicationDefinitionJson.CreateSerializerOptions();
var definition = JsonSerializer.Deserialize<ApplicationDefinition>(json, opts)!;

var host = FlowApplicationHost.Create(definition, registry);

var result = await host.StartAsync();
if (!result.IsSuccess)
    foreach (var e in result.Errors) Console.WriteLine(e.Message);

await host.StopAsync();
await host.DisposeAsync();

Documentation

The package README is the short entrypoint. The focused docs set starts at docs/README.md and covers getting started, definitions, node authoring, package authoring, hosting, observability, workspace projection, and validation, error handling, runtime states, JSON conversion, and expression mapping, package versioning, component composition, and storage host adapters.

Samples

See samples/FluxFlow.SampleApp for a small consumer-style console app. It keeps app-specific workspace metadata outside the engine, projects executable resources and workflows into ApplicationDefinition, registers typed components explicitly, and runs conditional links.

See samples/FluxFlow.MappingControlSample for a broker-free component composition sample. It keeps source and sink nodes in the host, then composes flow.mapper, flow.filter, flow.when, and flow.assert from reusable packages.

See samples/FluxFlow.MqttCompositionSample for an MQTT composition sample backed by an in-memory host adapter. It composes mqtt.subscribe, mapping/control nodes, and mqtt.publish without requiring a live broker.

See samples/FluxFlow.SessionsCompositionSample for a session recording and replay sample. It keeps storage in the host, records messages through session.recorder, then replays them through session.replay.

See samples/FluxFlow.StateCompositionSample for a timer, mapper, state reducer, and counter composition sample. It keeps host-specific expressions and sinks outside the reusable packages.

See samples/FluxFlow.StorageCompositionSample for a logical storage composition sample. It keeps the concrete store in the host while composing storage.put, storage.get, storage.query, and storage.delete.

See samples/FluxFlow.ComponentPackageTemplate for a copyable component package shape with contracts, options, diagnostics, module registration, and tests.

Component Packages

Reusable components live outside FluxFlow.Engine and are released separately.

Building

dotnet build FluxFlow.sln
dotnet test  FluxFlow.sln

License

FluxFlow.Engine is licensed under the MIT License.

Extension boundary

FluxFlow.Engine does not ship transport, storage, web, or designer components. Applications add those behaviors by:

1. Implementing IFlowNode for each external capability
2. Registering node factories with RuntimeNodeFactoryRegistry
3. Composing flows as JSON that FluxFlow.Engine builds and runs

Component packages should own their own options, validation, event names, tests, and documentation. They can expose an IFlowNodeModule or a registry extension that registers one module.

Applications are free to keep richer workspace files with dashboards, tests, connections, or UI state. Project only the executable resources and workflows into ApplicationDefinition before building the engine runtime.