Developing flow nodes

Nodes live in MultiShock.PluginSdk.Flow. Implement IFlowNodeProvider on your plugin class and return one node object per type from GetNodeTypes(). The host registers those objects in its node registry; do not create a new object for each event or flow.

Node types and instances

IFlowNode describes the type: name, category, icon, ports, properties, and the CreateInstance factory. The host stores plugin node type IDs as <plugin-id>:<TypeId>. Keep TypeId, port IDs, and property keys stable after publishing: saved flows reference them.

One node type object can serve several nodes and several flows. Put runtime state in IFlowNodeInstance.State or a store keyed by the actual instance, not ordinary fields such as _currentFlow on the node type. An InstanceId is unique within a flow; it is not an independent global identity.

A complete process node

This node adds a configured prefix to a text input. It has no device side effects, so it can be tested with Manual Trigger and Display.

using MultiShock.PluginSdk.Flow;
 
namespace MyPlugin;
 
public sealed class PrefixNode : IFlowProcessNode
{
    public string TypeId => "myplugin.prefix";
    public string DisplayName => "Add Prefix";
    public string Category => FlowNodeCategory.String;
    public string? Description => "Adds a prefix to text.";
    public string Icon => "text-cursor-input";
    public string? Color => null;
 
    public IReadOnlyList<FlowPort> InputPorts { get; } =
    [
        FlowPort.FlowIn(),
        FlowPort.String("text", "Text")
    ];
 
    public IReadOnlyList<FlowPort> OutputPorts { get; } =
    [
        FlowPort.FlowOut(),
        FlowPort.String("result", "Result")
    ];
 
    public IReadOnlyDictionary<string, FlowProperty> Properties { get; } =
        new Dictionary<string, FlowProperty>
        {
            ["prefix"] = FlowProperty.String("Prefix", "Message: ")
        };
 
    public IFlowNodeInstance CreateInstance(string instanceId, Dictionary<string, object?> config)
        => new FlowNodeInstance(instanceId, this, config);
 
    public Task<FlowNodeResult> ExecuteAsync(
        IFlowNodeInstance instance,
        FlowExecutionContext context,
        CancellationToken cancellationToken)
    {
        cancellationToken.ThrowIfCancellationRequested();
        var prefix = instance.GetConfig<string>("prefix", "Message: ");
        var text = context.GetInput<string>("text", string.Empty);
        var result = prefix + text;
 
        return Task.FromResult(FlowNodeResult.Success(
            new Dictionary<string, object?> { ["result"] = result })
            .WithDisplayValue(result));
    }
}

In your plugin's GetNodeTypes(), include new PrefixNode(). Restart MultiShock after changing node implementations. Wire Manual Trigger → Add Prefix → Display using the green pins, and connect Result to Display's Value. With no text input, the output is the configured prefix.

Inputs and properties

Data inputs use the IDs in FlowPort, not their display labels. The context contains inputs collected from the graph, but an unwired input has no entry. If a pin should override a property, check for the key explicitly:

var value = context.Inputs.ContainsKey("amount")
    ? context.GetInput<int>("amount")
    : instance.GetConfig<int>("amount", 1);

This preserves a legitimate zero input. Decide separately what null or an invalid converted value should mean. GetInput<T> is a convenience conversion, not validation of a required external value.

Use instance.GetConfig<T> to get saved configuration with property-default fallback. Match T to the property: FlowProperty.Int uses int, Double uses double, and Bool uses bool. context.GetConfig<T> reads context config and the explicit fallback; it does not look up node property defaults.

FlowProperty.String, Int, Double, Bool, and Select describe editor controls. Min/max values and Required help the editor, but your execution code must still validate imported or externally supplied values.

Results and errors

Output dictionary keys must match output port IDs. WithDisplayValue(text) shows debug text beneath the node; do not expose tokens or other private data there. Expected failures should be actionable. Let cancellation propagate as OperationCanceledException; do not turn a normal flow stop into a failure badge.

A complete trigger node

A trigger must return quickly from StartAsync. Start a cancellable listener, then stop and await it in StopAsync. Its Flow output must use ID triggered. The host listens to Triggered and filters events by the same runtime instance you pass back.

This example emits an increasing number every second. It stores listener ownership per instance and does not send device commands.

using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using MultiShock.PluginSdk.Flow;
 
namespace MyPlugin;
 
public sealed class HeartbeatNode : IFlowTriggerNode
{
    private sealed record Listener(CancellationTokenSource Cancellation, Task Task);
 
    public string TypeId => "myplugin.heartbeat";
    public string DisplayName => "Example Heartbeat";
    public string Category => FlowNodeCategory.Trigger;
    public string? Description => "Emits a count once a second while running.";
    public string Icon => "timer";
    public string? Color => null;
 
    public IReadOnlyList<FlowPort> InputPorts { get; } = [];
    public IReadOnlyList<FlowPort> OutputPorts { get; } =
    [
        new FlowPort { Id = "triggered", Name = "Triggered", Type = FlowPortType.Flow },
        FlowPort.Number("count", "Count")
    ];
    public IReadOnlyDictionary<string, FlowProperty> Properties { get; } =
        new Dictionary<string, FlowProperty>();
 
    public event Func<IFlowNodeInstance, Dictionary<string, object?>, Task>? Triggered;
 
    public IFlowNodeInstance CreateInstance(string instanceId, Dictionary<string, object?> config)
        => new FlowNodeInstance(instanceId, this, config);
 
    public async Task StartAsync(
        IFlowNodeInstance instance,
        IServiceProvider services,
        CancellationToken cancellationToken)
    {
        await StopAsync(instance);
        var logger = services.GetRequiredService<ILoggerFactory>()
            .CreateLogger("MyPlugin.Plugin.Heartbeat");
        var cancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
        var task = ListenAsync(instance, logger, cancellation.Token);
        instance.SetState("listener", new Listener(cancellation, task));
    }
 
    private async Task ListenAsync(
        IFlowNodeInstance instance,
        ILogger logger,
        CancellationToken cancellationToken)
    {
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(1));
        var count = 0;
        try
        {
            while (await timer.WaitForNextTickAsync(cancellationToken))
            {
                var outputs = new Dictionary<string, object?> { ["count"] = ++count };
                var handlers = Triggered;
                if (handlers is null)
                    continue;
 
                foreach (Func<IFlowNodeInstance, Dictionary<string, object?>, Task> handler
                    in handlers.GetInvocationList())
                {
                    cancellationToken.ThrowIfCancellationRequested();
                    await handler(instance, outputs);
                }
            }
        }
        catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
        {
            logger.LogDebug("Heartbeat stopped for {InstanceId}", instance.InstanceId);
        }
        catch (Exception exception)
        {
            logger.LogError(exception, "Heartbeat listener failed for {InstanceId}", instance.InstanceId);
        }
    }
 
    public async Task StopAsync(IFlowNodeInstance instance)
    {
        if (instance.GetState<Listener>("listener") is not { } listener)
            return;
 
        instance.State.Remove("listener");
        try
        {
            listener.Cancellation.Cancel();
            await listener.Task;
        }
        finally
        {
            listener.Cancellation.Dispose();
        }
    }
}

Add new HeartbeatNode() to the plugin's GetNodeTypes(), restart, and test with Example Heartbeat → Display, wiring both Triggered → Execute and Count → Value. Stop and restart the flow: counts should stop, then begin at 1 again. Test two flows at once to verify that stopping one does not stop the other.

The sample iterates the event invocation list so every asynchronous subscriber is awaited. For external event sources, keep callback tasks observable too; avoid async void listeners that lose errors or outlive the flow.

Dynamic definitions and concurrency

IDynamicPortNode.GetInputPorts(config) and GetOutputPorts(config) replace the static port definitions for that configuration. Generate stable IDs for ports that persist between edits, and ensure execution returns matching IDs.

IDynamicPropertyNode.GetProperties(config, serviceProvider) supplies runtime options. The service provider can be absent; support that case. Return cached options quickly instead of doing network I/O while the property panel renders.

Default flow execution is serialized, but users can enable concurrent executions and parallel branches. instance.State is an ordinary dictionary; shared mutable objects in it are not automatically thread-safe. Use appropriate synchronization or design nodes to avoid shared mutation. Flow-level variables also need care for compound updates. See flow execution.

Test your contract

• Exercise default properties, absent inputs, explicit zero/false/empty values, invalid input, and a saved flow loaded back from JSON.
• Verify output IDs, branch selection, cancellation during an awaited operation, and useful failure messages.
• For triggers, verify repeated start/stop, two instances in one flow, two flows, and cleanup after a source disconnects.
• Restart after node changes. Development page reload does not replace the flow subsystem's existing node registrations.

Keep process-node tests independent of hardware. Device delivery and real external events need separate integration checks; a successful node result alone does not prove either one happened.

Contributors

Thank you to everyone who has helped improve this documentation.

bwmp