Table of Contents

Class LiveSessionRunner

Namespace
DatabentoDotNet.Extensions.Hosting
Assembly
DatabentoDotNet.Extensions.Hosting.dll

Connects, authenticates, subscribes and starts one live session, then drains its records into an ILiveRecordHandler until the stream ends or the caller cancels.

public sealed class LiveSessionRunner : IAsyncDisposable
Inheritance
LiveSessionRunner
Implements
Inherited Members

Remarks

Needs no host and no container. The constructor takes a ResolvedLiveSession, a handler and a ReconnectSupervisor — nothing that only exists inside Microsoft.Extensions.DependencyInjection or Microsoft.Extensions.Hosting. That is what lets MockLiveGateway drive every behaviour here on an ordinary dotnet test, with no IServiceProvider and no IHostedService anywhere near it. The hosted service that wraps this in a BackgroundService is thin enough to be uninteresting precisely because everything interesting already works without it.

Starting is a separate call from running, and that split is deliberate. StartAsync(CancellationToken) only awaits ExecuteAsync until its first await that does not complete synchronously, so a session that connected and authenticated inside the record loop would fail in the background — with the host already reporting itself started. StartSessionAsync(CancellationToken) is the part a hosted service's own StartAsync override awaits, so a rejected API key fails the boot instead of vanishing into a background task nobody is watching.

The drain is synchronous and the fill is the only await, because there is no other shape available. An async method cannot return a ref struct — there is no Task<RecordRef>, and there never can be — so upstream's single-call next_record() does not port. What does port is its fill_buf() / try_next_record() pair: PumpAsync(LiveClient, CancellationToken) awaits FillBufferAsync(CancellationToken) once per pass and hands every record the fill produced to Drain(LiveClient), a plain synchronous method, before awaiting anything else. A RecordRef local inside an async method is fine; only one that survives an await is rejected, by the compiler, as CS4007 — which is the same lifetime rule TryNextRecord(out RecordRef) already imposes.

No System.IO.Pipelines and no Channel<T> between the socket and the handler. Both were considered and rejected: a ReadOnlySequence<T> may be non-contiguous, which breaks the reinterpret cast every record decode depends on, and either one adds a second buffering layer over the decoder's own. A channel would also force a copy per record to cross it — the one thing this library exists not to do.

An exception from the handler, or from the client, ends the session. Swallowing one would lose market data invisibly, which is the failure class this codebase exists to convert into loud ones. Cancelling RunAsync(CancellationToken)'s token is the one way to stop without that being a fault: it moves State to Stopped rather than Faulted, because a host stopping a background service must not be reported as a failed session.

Constructors

LiveSessionRunner(ResolvedLiveSession, ILiveRecordHandler, ReconnectSupervisor, ILogger<LiveSessionRunner>?, LiveSessionMetrics?)

Creates a runner for one resolved session.

public LiveSessionRunner(ResolvedLiveSession session, ILiveRecordHandler handler, ReconnectSupervisor supervisor, ILogger<LiveSessionRunner>? logger = null, LiveSessionMetrics? metrics = null)

Parameters

session ResolvedLiveSession

The session to run.

handler ILiveRecordHandler

Where every record and every flush goes.

supervisor ReconnectSupervisor

The reconnection policy — unused by StartSessionAsync(CancellationToken) and RunAsync(CancellationToken) beyond RecordSuccess() until reconnection lands.

logger ILogger<LiveSessionRunner>

Where ExtensionsLog writes. Defaults to Instance, so a caller who never configures logging pays nothing for it.

metrics LiveSessionMetrics

Where the four instruments are published, or null to publish none. Last and optional for the same reason logger is: a caller who wants neither writes neither, and every existing three- and four-argument call site still compiles.

Properties

CloseTimeout

How long RunAsync(CancellationToken) waits for a courteous close before dropping the socket instead. Defaults to five seconds.

public Duration CloseTimeout { get; init; }

Property Value

Duration

Remarks

Configurable from a host as {section}:Live:{name}:CloseTimeout — see CloseTimeout, which also explains why it is not derived from the host's own ShutdownTimeout. The five seconds here is what both a session that configures nothing and a runner constructed directly get.

Fault

Why the session faulted, or null while it has not.

public Exception? Fault { get; }

Property Value

Exception

Metadata

The DBN metadata the gateway sent when the session started, or null before StartSessionAsync(CancellationToken) has completed.

public Metadata? Metadata { get; }

Property Value

Metadata

RecordsReceived

How many records this session has handed to the handler so far.

public long RecordsReceived { get; }

Property Value

long

Session

The session this runner is running.

public ResolvedLiveSession Session { get; }

Property Value

ResolvedLiveSession

State

Where this session is in its lifecycle.

public LiveSessionState State { get; }

Property Value

LiveSessionState

Methods

AwaitCloseAsync(Task)

Half-closes, so the gateway gets to finish rather than having the socket dropped on it — but bounded, so a gateway that never answers cannot hold the host's shutdown open.

public Task<bool> AwaitCloseAsync(Task closing)

Parameters

closing Task

The close to wait for — normally LiveClient.CloseAsync().

Returns

Task<bool>

true if the close finished inside the ceiling, false if the ceiling expired first.

Remarks

The losing task is left to complete on its own rather than cancelled. It holds a timer and nothing else, it finishes within CloseTimeout, and cancelling it would leave a faulted task nobody awaits — noise, in exchange for reclaiming one timer five seconds early.

The timer is a plain Delay(TimeSpan, CancellationToken), not Delay. It once was; ReconnectSupervisor is documented to hold the reconnect schedule and nothing else, and a caller can now replace that seam to turn it into a synchronisation point for the backoff itself — see LiveSessionReconnectTests. Routing this unrelated shutdown ceiling through the same seam would corrupt that signal, not just be untidy, so this waits on the BCL primitive directly. ToTimeSpan() converts CloseTimeout for the one call that needs it; nothing here stores a TimeSpan.

DisposeAsync()

Disposes the underlying LiveClient, if one was built. Idempotent.

public ValueTask DisposeAsync()

Returns

ValueTask

RunAsync(CancellationToken)

Drains records into the handler until the gateway closes the stream or cancellationToken is cancelled.

public Task RunAsync(CancellationToken cancellationToken)

Parameters

cancellationToken CancellationToken

Stops the loop. Cancelling it ends the session — see FillBufferAsync(CancellationToken) — but is reported as Stopped, not Faulted: a host stopping is not a failure.

Returns

Task

Exceptions

InvalidOperationException

StartSessionAsync(CancellationToken) has not completed.

StartSessionAsync(CancellationToken)

Connects, authenticates, sends every subscription in Session in order, and starts the session.

public Task StartSessionAsync(CancellationToken cancellationToken)

Parameters

cancellationToken CancellationToken

Cancels the handshake. Cancelling any step here leaves the underlying connection unusable — see AuthenticateAsync(CancellationToken) — so there is nothing to resume; construct another runner to try again.

Returns

Task

Remarks

Separate from RunAsync(CancellationToken) on purpose. See the type-level remarks: this is the half a hosted service's own start-up awaits, so a rejected key or an unreachable gateway fails the host's boot rather than a background loop nobody is watching yet.

Exceptions

InvalidOperationException

This runner has already been started.

Edit this page