Adapter Interface

IDeviceAdapter is the single extension point your provider implements. It defines how your protocol handles device lifecycle, message delivery, and command execution. The canonical contract lives in the adapter-contract spec.

IDeviceAdapter

public interface IDeviceAdapter : IAsyncDisposable
{
    AdapterCapabilities Capabilities { get; }

    Task StartAsync(CancellationToken cancellationToken);
    Task StopAsync(CancellationToken cancellationToken);

    Task<ProvisionResult> OnProvisionAsync(ProvisionContext context, CancellationToken ct);
    Task<ProvisionResult> OnUnprovisionAsync(string deviceId, CancellationToken ct);

    Task SendDesiredPropertiesAsync(string deviceId, JsonObject properties, CancellationToken ct);
    Task<CommandResult> SendCommandAsync(AdapterCommand command, CancellationToken ct);

    bool IsDeviceConnected(string deviceId);
}

Your adapter should extend BaseAdapter rather than implementing IDeviceAdapter directly. BaseAdapter injects AdapterContext and exposes the following clients to subclasses:

Property	Interface	Use
Telemetry	ITelemetryClient	Push device-to-cloud telemetry into the SDK pipeline.
Properties	IPropertiesClient	Push reported-property updates.
Diagnostics	IDiagnosticsClient	Report connection lifecycle events (connected/disconnected/auth-fail).
Twin	ITwinReader	Read the current persisted twin (desired/reported/tags). See Twin Reader.
AsyncActions	IAsyncActionHelper	Bridge an async upstream API into the synchronous direct-method timeout window. Use only for sub-30s actions; longer ones use property convergence.

BaseAdapter also provides a default DisposeAsync that calls StopAsync.

Capabilities

The SDK reads Capabilities once at startup and uses the flags to decide whether it must manage behaviours itself or can rely on the transport's native semantics.

public sealed record AdapterCapabilities
{
    public bool SupportsRetain { get; init; }          // e.g. MQTT retain flag
    public bool RetainSurvivesRestart { get; init; }   // retained messages persist across broker restart
    public bool SupportsCommands { get; init; }        // request/response with timeout
}

If SupportsRetain == false (or RetainSurvivesRestart == false), the SDK's IDesiredDeliveryTracker is the source of truth for desired-property delivery and re-fires desired updates on reconnect.

Lifecycle

StartAsync / StopAsync

Called by the SDK's AdapterLifecycleService during application startup and shutdown. Use these to:

• Connect to the protocol broker (e.g., MQTT connect, WebSocket listen).
• Subscribe to device topics or open the listener.
• Register disconnect/reconnect handlers.
• Clean up connections on shutdown.

The SDK ensures StartAsync runs after DI is built and DbUp migrations have completed. Bootstrap that must finish before any other hosted service starts (e.g. the MQTT superuser seeder) belongs in Program.cs between Build() and RunAsync(), not in StartAsync.

OnProvisionAsync

Called inside the provisioning database transaction. The SDK has already inserted the DeviceEntity and DeviceTwinEntity row. Your adapter creates any protocol-specific records.

public override async Task<ProvisionResult> OnProvisionAsync(
    ProvisionContext context, CancellationToken ct)
{
    // context.DeviceId        — device identifier
    // context.AuthMethod      — DeviceAuthMethod.SymmetricKey | Certificate
    // context.PrimaryKey      — plaintext primary key (SymmetricKey only)
    // context.SecondaryKey    — plaintext secondary key (nullable)
    // context.PrimaryKeyHash  — pre-computed bcrypt hash (for broker auth DBs)
    // context.SecondaryKeyHash — pre-computed bcrypt hash (nullable)

    var user = MqttUserEntity.Create(
        context.DeviceId, context.PrimaryKeyHash!, context.DeviceId);
    await _userRepo.CreateAsync(user, ct);

    return ProvisionResult.Ok();
}

If your adapter returns ProvisionResult.Failed(error), the entire transaction is rolled back — no device, twin, or protocol records are created. The error string surfaces back to the caller.

OnUnprovisionAsync

Called inside the unprovisioning transaction. Clean up protocol-specific records. The bus-api unprovision is idempotent — receiving an OnUnprovisionAsync for a device that has no protocol-side state is normal; treat missing rows as success.

public override async Task<ProvisionResult> OnUnprovisionAsync(
    string deviceId, CancellationToken ct)
{
    await _userRepo.DeleteByDeviceIdAsync(deviceId, ct: ct);
    await _aclService.DeleteDeviceAclAsync(deviceId, ct: ct);
    return ProvisionResult.Ok();
}

Reporting from the device side

Telemetry

Adapters forward bytes from the protocol transport into the SDK pipeline via BaseAdapter.Telemetry (an ITelemetryClient):

public interface ITelemetryClient
{
    void Send(string deviceId, TelemetryMessage message);
}

public record TelemetryMessage
{
    public required ReadOnlyMemory<byte> Payload { get; init; }
    public bool IsReportedProperties { get; init; }
}

Send is fire-and-forget — it writes to a bounded Channel<T> and returns immediately. It's safe to call from any thread, including synchronous protocol event handlers. If the channel is full the message is dropped and pipeline.messages.dropped is incremented (warning is rate-limited).

Reported properties

Reported properties have two equivalent surfaces. Use the structured one if you already have a JsonObject:

public interface IPropertiesClient
{
    void Report(string deviceId, JsonObject properties);
}

If you only have bytes from the transport, send them through Telemetry.Send with IsReportedProperties = true — the pipeline parses them on the reported branch.

Both go through the same channel; they end up in device_twin_entity.reported (jsonb_set + version increment) and a NATS publish with messageType: ProviderClientPropertiesChanged.

Diagnostics

public interface IDiagnosticsClient
{
    void ReportConnected(string deviceId, ConnectionMetadata metadata);
    void ReportDisconnected(string deviceId, string reason, ConnectionMetadata? metadata = null);
    void ReportAuthFailure(string deviceId, string detail);
    void ReportEvent(string deviceId, DiagnosticEvent evt);
}

ConnectionMetadata carries optional RemoteAddress, ProtocolVersion, KeepaliveSeconds, CleanSession, plus a free-form Extra dictionary. DiagnosticEvent carries EventType, Level, optional Detail and Data.

These are fire-and-forget. The SDK batches status updates (StatusFlushIntervalSeconds, default 5 s) and connection log entries (LogBatchSize, default 100) before flushing to PostgreSQL.

Cloud → device paths

SendDesiredPropertiesAsync

Called when desired properties change (via the bus-api set-desired or HTTP PATCH /api/devices/{id}/desired, both RFC 7396 JSON Merge Patch). Push the merged desired document to the connected device.

public override async Task SendDesiredPropertiesAsync(
    string deviceId, JsonObject properties, CancellationToken ct)
{
    var msg = new MqttApplicationMessageBuilder()
        .WithTopic($"devices/{deviceId}/twin/desired")
        .WithPayload(JsonSerializer.SerializeToUtf8Bytes(properties))
        .WithRetainFlag(true)
        .Build();

    await _mqttClient.PublishAsync(msg, ct);
}

If your transport doesn't support push (e.g. HTTP polling), return Task.CompletedTask. The SDK's IDesiredDeliveryTracker records the pending update; the device gets it on the next pull.

Convergence — keeping desired and reported in sync

For aggregator/REST adapters where the upstream may reject (4xx), throttle (429/5xx), or take time to apply changes, register a convergence policy so the SDK keeps pushing with exponential backoff until reported matches desired. The result of each round-trip is recorded in desiredSync ({ ac, av, ad, ... }) so callers can observe whether the write actually stuck.

Adapters opt into convergence by:

1. Throwing UpstreamRejectedException(statusCode, reason) for 4xx-class definitive rejections (the SDK stops retrying and stamps the envelope with the exact code).
2. Letting any other exception (5xx, network, timeout) bubble — the SDK retries per OnError (default Retry).

SendCommandAsync

Called when a cloud-to-device method is invoked. Deliver the command and return a CommandResult. CommandStatus is a closed enum — adapter implementations MUST NOT introduce new values without amending the spec.

public sealed record CommandResult
{
    public required string CorrelationId { get; init; }
    public required CommandStatus Status { get; init; }
    public int? StatusCode { get; init; }      // device-side HTTP-like code (Ok/DeviceError only)
    public JsonObject? Response { get; init; } // device response body, null when no body
    public string? ErrorCode { get; init; }    // = nameof(Status) for non-Ok/non-DeviceError
    public string? ErrorMessage { get; init; } // human-readable, no PII
    public required bool Retryable { get; init; }
}

public enum CommandStatus
{
    Ok,                       // device replied 2xx
    DeviceError,              // device replied 4xx/5xx
    DeviceNotConnected,       // adapter knows the device has no live session
    MethodTimeout,            // no device reply within deadline
    DuplicateCorrelationId,   // caller reused an in-flight correlation id
    PendingCapacityExceeded,  // adapter pending-command registry full
    PayloadTypeMismatch,      // request or response not parseable JSON
    TransportError,           // publish failed, broker disconnect, generic adapter fault
    Cancelled,                // caller cancellation or shutdown
}

Construct results explicitly — there are no Ok()/Failed() factories. Always echo the request CorrelationId, even on transport-level failures where the request never reached the device.

return new CommandResult
{
    CorrelationId = command.CorrelationId ?? string.Empty,
    Status = CommandStatus.Ok,
    StatusCode = 200,
    Response = response,
    Retryable = false,
};

catch (OperationCanceledException) when (!ct.IsCancellationRequested)
{
    return new CommandResult
    {
        CorrelationId = command.CorrelationId ?? string.Empty,
        Status = CommandStatus.MethodTimeout,
        ErrorCode = nameof(CommandStatus.MethodTimeout),
        Retryable = true,
    };
}

StatusCode is set only for Ok and DeviceError. Response is null for transport-level statuses; a device response with no body is also null (not an empty JsonObject).

IsDeviceConnected

Returns whether a device is currently connected via your protocol on this replica. The bus-api method-invocation handler is broadcast (no queue group) — every replica receives the request, but only the one whose IsDeviceConnected(deviceId) returns true answers. If no replica owns the device, the NATS request times out.

BaseAdapter

BaseAdapter exposes the SDK clients to subclasses:

protected ITelemetryClient   Telemetry   => _context.Telemetry;
protected IPropertiesClient  Properties  => _context.Properties;
protected IDiagnosticsClient Diagnostics => _context.Diagnostics;

AdapterContext is a DI-managed singleton constructed by AddBeaconTowerProvider<TAdapter>:

public sealed class AdapterContext
{
    public required ITelemetryClient   Telemetry   { get; init; }
    public required IPropertiesClient  Properties  { get; init; }
    public required IDiagnosticsClient Diagnostics { get; init; }
}

TelemetryPipeline implements both ITelemetryClient and IPropertiesClient. DeviceStatusProjection implements IDiagnosticsClient.

Message transformers

The SDK supports per-device payload transformation through IMessageTransformer, resolved by IMessageTransformerResolver on the device's parser twin tag. The default resolver returns JsonPassthroughTransformer.Instance for every device until a host swaps it.

A per-device transformer-error circuit breaker (TransformerOptions) opens after 10 failures in 60 s with a 30 s cool-down. While the circuit is open the device's messages bypass the transformer and pass through unchanged, with a metric increment.