FluxFlow.Components.Mqtt 1.2.1

FluxFlow.Components.Mqtt

MQTT component package for FluxFlow.

This package keeps protocol-specific nodes outside FluxFlow.Engine while preserving the same runtime registration model.

Nodes

The package does not include a concrete network client. Applications provide an adapter through RegisterMqttComponents.

var registry = new RuntimeNodeFactoryRegistry()
    .RegisterMqttComponents((context, cancellationToken) =>
        ValueTask.FromResult(MqttClientLease.Owned(
            new AppMqttClientAdapter(context.Profile))));

Options are static node settings. Requests and messages are per-item data.

mqtt.publish bounds each adapter publish with publishTimeoutMilliseconds (default 30000). A publish that exceeds the timeout is reported as a per-message error and the node continues with later requests, so a hung adapter cannot wedge the node.

MqttClientFactoryContext includes the runtime node address, connection name, and connection profile. Hosts can use ConnectionName to resolve app-owned resources instead of placing all broker settings inline.

Return MqttClientLease.Owned(adapter) when the node should dispose the adapter. Return MqttClientLease.Shared(adapter) when the host owns a shared client lifetime.

Subscriptions return IMqttSubscription; once SubscribeAsync returns, startup is considered successful. Each subscription should expose an independent message stream so shared clients can safely serve multiple nodes.

Runtime Timing

Use MqttComponentOptions.UseClock(...) when tests or hosts need deterministic package-owned timestamps.

var registry = new RuntimeNodeFactoryRegistry()
    .RegisterMqttComponents(options => options
        .UseClientFactory(factory)
        .UseClock(testClock));

mqtt.publish uses the configured clock for MqttPublishResult.Timestamp. MQTT publish, subscribe, and connection health events also use that clock for their FlowEvent.Timestamp values.

Incoming MqttReceivedMessage.Timestamp values stay adapter-owned because they represent when the adapter observed the broker message. Adapter-provided MqttClientHealthEvent.Timestamp values also stay adapter-owned, while the emitted workflow event timestamp uses the configured package clock.

Adapter Health

Adapters that also implement IMqttClientHealthSource can expose connection health transitions. mqtt.publish and mqtt.subscribe forward those values as diagnostics and events named mqtt.connection.healthChanged.

The package does not own reconnect policy. Adapters decide how to connect, reconnect, and report state.

Reconnect Policy Hints

Nodes can optionally set reconnect to pass policy hints to the host adapter through MqttClientFactoryContext.Reconnect.

{
  "type": "mqtt.publish",
  "name": "publisher",
  "defaultTopic": "devices/state",
  "reconnect": {
    "enabled": true,
    "maxAttempts": 5,
    "initialDelayMilliseconds": 100,
    "maxDelayMilliseconds": 5000,
    "backoffMultiplier": 2,
    "useJitter": true
  }
}

These values are advisory. The package validates them and passes them to the adapter factory. The adapter still owns connection state, retry loops, shared client behavior, and broker-specific recovery.

Topic validation

Use MqttTopicValidator.ValidatePublishTopic and MqttTopicValidator.ValidateSubscriptionFilter when projecting host settings or building requests.

Publish topics must be present and cannot contain MQTT wildcards. Subscription filters may use + as a complete level and # only as the final complete level. Both helpers also reject null characters and oversized encoded topics.

Design Metadata

This package exposes a package-owned IComponentDesignMetadataProvider for its node types. Hosts can compose it through ComponentDesignMetadataCatalog to populate palettes, editors, validation views, and documentation without duplicating package descriptors.

Composition Guidance

Use this package as one part of a host-composed graph. See Component Composition for recommended host boundaries, package boundaries, and extraction timing.