Skip to main content

Reconciling Poller

ReconcilingPoller<TItem> is an SDK-provided BackgroundService base that handles the boilerplate every aggregator-backed adapter would otherwise hand-roll: cancellation-aware delay loop, exception isolation per item, diff against the persisted twin, projection into the reported-properties write path.

Pattern​

internal sealed class EnodePoller : ReconcilingPoller<EnodePoller.Item>
{
    private readonly EnodeApiClient _api;
    private readonly IServiceProvider _services;

    public EnodePoller(
        ITwinReader twin,
        IPropertiesClient properties,
        IServiceProvider services,
        EnodeApiClient api,
        ILogger<EnodePoller> logger)
        : base(twin, properties, logger)
    {
        _api = api;
        _services = services;
    }

    protected override TimeSpan Interval => TimeSpan.FromSeconds(60);

    protected override async IAsyncEnumerable<Item> ListAsync(
        [EnumeratorCancellation] CancellationToken ct)
    {
        var page = await _api.ListChargersAsync(ct);
        foreach (var node in page!["data"]!.AsArray())
        {
            yield return new Item(
                ExternalId: node!["id"]!.GetValue<string>(),
                Reported:   (JsonObject)node);
        }
    }

    protected override string? DeviceIdFor(Item item)
    {
        // map upstream id → BT device via the new SDK seam
        using var scope = _services.CreateScope();
        var devices = scope.ServiceProvider.GetRequiredService<IDeviceRepository>();
        var device = devices.FindByExternalIdAsync(item.ExternalId)
            .GetAwaiter().GetResult();
        return device?.DeviceId;
    }

    protected override JsonObject ReportedFor(Item item) => item.Reported;

    public sealed record Item(string ExternalId, JsonObject Reported);
}

Register it:

builder.Services.AddHostedService<EnodePoller>();

What the base class does for you​

  • Initial StartupDelay (default 5s) before the first cycle so SDK background services finish coming up.
  • Interval (default 60s) between cycles.
  • Cancellation-aware delay loop that respects shutdown.
  • Exception isolation: a cycle failure logs and continues; a single item's failure logs at warning and continues with the next item.
  • Diff against the persisted twin via ITwinReader.ReadAsync. Only emits Properties.Report when reported-state actually changed.

What you implement​

OverrideReturns
ListAsync(ct)IAsyncEnumerable<TItem> of upstream items
DeviceIdFor(item)BT deviceId (or null to skip)
ReportedFor(item)Canonical reported-properties JSON

Optional:

OverrideDefault
Interval60s
StartupDelay5s
HasChanged(item, persistedReported, upstreamReported)Equality-by-serialized-JSON

When NOT to use it​

  • For high-throughput telemetry forwarding, write directly via Telemetry.Send — the pipeline owns its own batching + dedup.
  • If the upstream supports reactive subscriptions (websocket, AMQP, webhook firehose), prefer those over polling. Use webhooks for the push path and a sparse poller as a safety net.

See also​