Class LiveClient

Namespace: DatabentoDotNet.Live

Assembly: DatabentoDotNet.Live.dll

A client for Databento's live gateway: real-time market data, and intraday replay from the same socket.

public sealed class LiveClient : IAsyncDisposable

Inheritance: object

LiveClient

Implements: IAsyncDisposable

Inherited Members: object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.ReferenceEquals(object, object)

object.ToString()

Examples

using DatabentoDotNet;
using DatabentoDotNet.Dbn;
using DatabentoDotNet.Live;

await using var client = new LiveClient
{
    ApiKey = new ApiKey(Environment.GetEnvironmentVariable("DATABENTO_API_KEY")!),
    Dataset = "EQUS.MINI",

    // Gateway left unset: the client derives this dataset's host on lsg.databento.com itself.
};

// Four calls, because the protocol has four steps. Nothing is billed until StartAsync.
await client.ConnectAsync();
await client.AuthenticateAsync();
await client.SubscribeAsync(new Subscription
{
    Schema = Schema.Trades,
    Symbols = Symbols.From(["AAPL", "MSFT"]),
});

Metadata metadata = await client.StartAsync();
Console.WriteLine($"DBN v{metadata.Version}, stype_out {metadata.StypeOut.ToWireString()}");

while (true)
{
    // Drain what the last read produced before asking the socket for more: a refill may move the
    // buffer, and that is exactly what ends a RecordRef's life.
    while (client.TryNextRecord(out RecordRef record))
    {
        if (record.TryGet(out TradeMsg trade))
        {
            Console.WriteLine($"{DbnTime.ToInstant(trade.IndexTs)} {trade.Price} x {trade.Size}");
        }
    }

    if (await client.FillBufferAsync() == 0)
    {
        break;   // the gateway closed the stream
    }
}

// Half-close the socket and let the gateway finish, rather than dropping it.
await client.CloseAsync();

Intraday replay is the same session with a Start on the subscription; the stream transitions to live once it catches up. RecordsAsync(CancellationToken) is the record loop above written once, at two allocations per record.

Remarks

Port of upstream's live::Client (live/client.rs) and of the client half of its live::protocol::Protocol. Upstream's build() connects and authenticates in one call; splitting them is what lets each land with tests of its own against the mock gateway, and it is what makes ConnectTimeoutException and AuthTimeoutException nameable as the separate failures they are.

The order is ConnectAsync(CancellationToken), AuthenticateAsync(CancellationToken), SubscribeAsync(Subscription, CancellationToken), StartAsync(CancellationToken), then records — and StartAsync(CancellationToken) is where billing begins. Nothing before it moves market data: a subscription tells the gateway what to send later, and the gateway sends nothing at all until the session is started. The example below is that order in full.

RecordsAsync(CancellationToken) is the same loop with each record copied onto the heap, for callers who would rather write an await foreach than hold the zero-copy guarantee.

Not thread-safe, and deliberately not made so. One connection is one conversation with the gateway, and the record loop is a single reader by construction — a lock around it would suggest a concurrency the protocol does not have.

No builder. Upstream's ClientBuilder<AK, D> is generic type-state whose only purpose is to make "no API key" and "no dataset" unrepresentable — build() exists only on ClientBuilder<ApiKey, String>. C# 11 required init properties do exactly that natively, checked by the compiler at every construction site. See PORTING.md §2.

Endpoint survives CloseAsync(), on purpose. ReconnectAsync(CancellationToken) reuses the already-resolved address and does not re-resolve DNS, as upstream's reconnect() does not (PORTING.md §4), so the resolved address has to outlive the socket it came from.

The handshake is not cancel-safe, and this type does not pretend otherwise. A partially written authentication line desynchronises the gateway, which closes the connection — so AuthenticateAsync(CancellationToken) cancels by tearing the socket down rather than by abandoning a half-finished write. A caller whose AuthenticateAsync(CancellationToken) throws is disconnected and must connect again; there is no resuming it. PORTING.md §4.

Fields

DefaultAuthTimeout

The handshake budget used when none is set: ten seconds, matching DefaultConnectTimeout. Upstream has no such budget at all — see AuthTimeoutException.

public static readonly Duration DefaultAuthTimeout

Field Value

Duration

DefaultConnectTimeout

The connect budget used when none is set: ten seconds, matching upstream.

public static readonly Duration DefaultConnectTimeout

Field Value

Duration

DefaultReadTimeout

The read budget used when neither ReadTimeout nor HeartbeatInterval is set: 35 seconds, matching upstream's heartbeat_timeout fallback.

public static readonly Duration DefaultReadTimeout

Field Value

Duration

MaxHeartbeatInterval

The longest heartbeat interval the gateway accepts.

public static readonly Duration MaxHeartbeatInterval

Field Value

Duration

MinHeartbeatInterval

The shortest heartbeat interval the gateway accepts.

public static readonly Duration MinHeartbeatInterval

Field Value

Duration

ReadTimeoutHeartbeatMargin

How much longer than HeartbeatInterval the derived read budget runs: five seconds, matching upstream. It is the allowance for scheduling and network jitter between the gateway deciding to send a heartbeat and this client seeing one.

public static readonly Duration ReadTimeoutHeartbeatMargin

Field Value

Duration

Properties

ApiKey

The API key to authenticate with. Validated when it is constructed.

public required ApiKey ApiKey { get; init; }

Property Value

ApiKey

AuthTimeout

How long the whole of AuthenticateAsync(CancellationToken) may spend before raising AuthTimeoutException. Defaults to DefaultAuthTimeout.

public Duration AuthTimeout { get; init; }

Property Value

Duration

Remarks

One budget for the exchange rather than one per line: a gateway that sends the greeting and then stalls has spent the caller's time just as surely as one that never speaks at all.

Compression

The compression to negotiate for the record stream. Defaults to None.

public Compression Compression { get; init; }

Property Value

Compression

Remarks

Settled during the handshake and not afterwards: compression= travels on the authentication line, so what this says is what the stream after start_session is framed as. Control lines are always plaintext regardless.

ConnectTimeout

How long ConnectAsync(CancellationToken) may spend before raising ConnectTimeoutException. Defaults to DefaultConnectTimeout.

public Duration ConnectTimeout { get; init; }

Property Value

Duration

Dataset

The dataset to stream, in its wire spelling — GLBX.MDP3. A string rather than the codec's Dataset enum, for the reason given on LiveGateway.

public required string Dataset { get; init; }

Property Value

string

EffectiveReadTimeout

The read budget actually applied: ReadTimeout when it is set, otherwise HeartbeatInterval plus ReadTimeoutHeartbeatMargin, otherwise DefaultReadTimeout.

public Duration EffectiveReadTimeout { get; }

Property Value

Duration

Remarks

Port of upstream's heartbeat_timeout(). Exposed rather than kept private because it is the number a caller has to know to interpret a HeartbeatTimeoutException — and because a derived value that cannot be read back is a setting whose effect can only be discovered by waiting for it.

Endpoint

The address ConnectAsync(CancellationToken) actually reached, once it has. Survives CloseAsync() so a reconnect can reuse it rather than resolving DNS again.

public IPEndPoint? Endpoint { get; }

Property Value

IPEndPoint

Gateway

Connect here instead of at the host LiveGateway derives from Dataset. For tests against a mock gateway, and for the rare deployment that is told a different address.

public EndPoint? Gateway { get; init; }

Property Value

EndPoint

Greeting

The gateway's greeting line — lsg_version=… — kept verbatim for diagnostics, and deliberately not parsed. Set by AuthenticateAsync(CancellationToken), cleared by ConnectAsync(CancellationToken).

public string? Greeting { get; }

Property Value

string

Remarks

Upstream reads it and logs it at debug, and nothing in the protocol depends on it. Keeping the string is what makes "which gateway build did this happen against" answerable from an exception report rather than only from a packet capture.

HeartbeatInterval

How often the gateway should emit a heartbeat when no other record is due, or null to leave it to the gateway's own default.

public Duration? HeartbeatInterval { get; init; }

Property Value

Duration?

Remarks

Heartbeats arrive as ordinary SystemMsg records carrying Heartbeat, not as control frames.

Validated here, where upstream leaves it to the gateway. Upstream's builder documents the 5–1800 second range but only warns about sub-second precision, which it then silently discards (live.rs:133-146). Both are rejected instead: a value out of range costs a round trip and a closed connection to discover, and a silently truncated one means the interval in the caller's code is not the interval on the wire — the confidently-wrong failure this codebase exists to prevent.

Exceptions

ArgumentOutOfRangeException: The interval is outside MinHeartbeatInterval..MaxHeartbeatInterval.
ArgumentException: The interval is not a whole number of seconds.

IsAuthenticated

Whether the handshake on the current connection has succeeded.

public bool IsAuthenticated { get; }

Property Value

bool

IsClosed

Whether the record stream has ended: the gateway closed it cleanly, the read budget elapsed, or CloseAsync() was called.

public bool IsClosed { get; }

Property Value

bool

Remarks

Port of upstream's is_closed(), including its starting value: a client that has never connected reports false, because this answers "did the stream end", not "is there a stream". IsConnected and IsSessionStarted answer the other two questions.

A clean close is not an error. It surfaces as false from TryNextRecord(out RecordRef) and 0 from FillBufferAsync(CancellationToken) — the same values a merely-empty buffer produces — which is what makes this property the way to tell "no records right now" from "no records ever again". PORTING.md §2.

IsConnected

Whether a socket is currently open.

public bool IsConnected { get; }

Property Value

bool

IsSessionStarted

Whether StartAsync(CancellationToken) has run on the current connection, so the record stream is open and FillBufferAsync(CancellationToken) and TryNextRecord(out RecordRef) may be called.

public bool IsSessionStarted { get; }

Property Value

bool

Metadata

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

public Metadata? Metadata { get; }

Property Value

Metadata

Remarks

The same object StartAsync(CancellationToken) returns, kept because it is what TsOut and the symbol mappings are read from long after the call that produced it. It survives CloseAsync() and is cleared by ConnectAsync(CancellationToken), exactly as Greeting and SessionId are: what the last session said is a diagnostic, and a reconnect is the only thing that can replace it.

ReadTimeout

How long the record stream may go silent before FillBufferAsync(CancellationToken) raises HeartbeatTimeoutException, or null to derive it from HeartbeatInterval. See EffectiveReadTimeout.

public Duration? ReadTimeout { get; init; }

Property Value

Duration?

Remarks

Upstream has no such setting — its heartbeat_timeout() is always heartbeat_interval + 5s, or 35 seconds when no interval was requested, with no way to override it. That derivation is the right default and is what EffectiveReadTimeout computes; it is a poor *only* option, because the budget that matters is a property of the deployment — a replay of a quiet overnight session and a busy equities open are the same code reading very different streams.

Setting this shorter than the gateway's heartbeat interval will time out a healthy connection, since a heartbeat is the only traffic guaranteed on a quiet feed. Nothing here rejects that combination: HeartbeatInterval may be left unset, in which case the gateway picks its own interval and this client has no way to know what it is.

Exceptions

ArgumentOutOfRangeException: The budget is zero or negative.

SendTsOut

Whether to ask the gateway to append its send timestamp to every record. When set, every record on the stream is eight bytes longer and decodes as WithTsOut<T>.

public bool SendTsOut { get; init; }

Property Value

bool

SessionId

The session_id the gateway assigned, once AuthenticateAsync(CancellationToken) has succeeded, or null when it did not send one.

public string? SessionId { get; }

Property Value

string

Remarks

Upstream maps an absent session_id to the empty string (protocol.rs, unwrap_or_default), which makes "the gateway sent no id" and "the gateway sent an empty id" the same value. They are kept apart here: this is the identifier a support request is answered against, so whether it exists is worth being able to tell.

SlowReaderBehavior

What the gateway should do when this client falls behind real time, or null to leave it to the gateway's default.

public SlowReaderBehavior? SlowReaderBehavior { get; init; }

Property Value

SlowReaderBehavior?

Subscriptions

Every subscription sent on this client, in the order it was sent, with Id filled in.

public IReadOnlyList<Subscription> Subscriptions { get; }

Property Value

IReadOnlyList<Subscription>

Remarks

Upstream's subscriptions(). It survives CloseAsync() for the same reason Endpoint does: a reconnect has to replay them, and a list cleared on disconnect would leave nothing for ResubscribeAsync(CancellationToken) to replay.

Read-only, where upstream also exposes subscriptions_mut(). The one thing that mutation is for upstream — clearing each start before a resubscribe, so a reconnect does not replay the same history twice — belongs to the resubscribe itself rather than to callers.

UpgradePolicy

How records from earlier DBN versions are handled while decoding. Defaults to UpgradeToV3, as upstream does.

public VersionUpgradePolicy UpgradePolicy { get; init; }

Property Value

VersionUpgradePolicy

Methods

AuthenticateAsync(CancellationToken)

Runs the CRAM handshake: reads the greeting and the challenge, sends the authentication request, and reads the response.

public Task AuthenticateAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken: Cancels the handshake by closing the connection. See the remarks.

Returns

Task

Remarks

Port of upstream's Protocol::authenticate (live/protocol.rs). The request is one line:

auth={sha256_hex}-{bucket_id}|dataset={ds}|encoding=dbn|compression={c}|ts_out={0|1}|client={ua}

where the digest is SHA256(challenge + "|" + apiKey) in lowercase hex and the bucket id is the last five characters of the key — the only part of it that goes on the wire. heartbeat_interval_s and slow_reader_behavior follow when they are set, and are omitted entirely when they are not, so the gateway applies its own defaults rather than ours.

Not cancel-safe, and it cancels by disconnecting. The gateway reads a control message as a whole line; half of one desynchronises it and it closes the connection. So neither cancellationToken nor AuthTimeout is threaded into the middle of a write — both abort by tearing the socket down, which fails the pending read or write outright. Every failure here leaves the client disconnected: there is nothing left to retry the handshake on. PORTING.md §4.

Exceptions

InvalidOperationException: No connection is open, or this connection has already authenticated.
DatabentoAuthenticationException: The gateway rejected the credentials.
LiveProtocolException: The gateway sent something that is not a handshake, or stopped sending mid-way.
AuthTimeoutException: The exchange outlived AuthTimeout.

CloseAsync()

Closes the connection, keeping Endpoint. A no-op when nothing is open.

public Task CloseAsync()

Returns

Task

ConnectAsync(CancellationToken)

Opens a TCP connection to the gateway. Sends nothing: the handshake is a separate step.

public Task ConnectAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken: Cancels the attempt.

Returns

Task

Exceptions

InvalidOperationException: A connection is already open.
ArgumentException: Gateway is unset and Dataset does not produce a usable host name. See For(string).
ConnectTimeoutException: The attempt outlived ConnectTimeout.
LiveConnectException: The attempt failed — refused, unreachable, unresolvable.

DisposeAsync()

Closes the connection.

public ValueTask DisposeAsync()

Returns

ValueTask

FillBufferAsync(CancellationToken)

Reads whatever the gateway has sent into the decoder's buffer, ready for TryNextRecord(out RecordRef).

public ValueTask<int> FillBufferAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken: Stops the loop. See the remarks.

Returns

ValueTask<int>: How many bytes were read, or 0 when the gateway closed the stream cleanly — after which IsClosed is set and every later call returns 0 without touching the socket.

Examples

while (true)
{
    // The inner loop must run to false before each refill. Not a style preference: a refill may
    // shift the buffer, which is what invalidates a RecordRef the caller is still holding.
    while (client.TryNextRecord(out RecordRef record))
    {
        if (record.TryGet(out TradeMsg trade))
        {
            Console.WriteLine($"{DbnTime.ToInstant(trade.IndexTs)} {trade.Price} x {trade.Size}");
        }
    }

    if (await client.FillBufferAsync() == 0)
    {
        break;   // the gateway closed the stream
    }
}

Remarks

Port of upstream's fill_buf(), and the half of the zero-copy pair that does the I/O. The canonical loop is:

while (true)
{
    while (client.TryNextRecord(out var record)) { Process(record); }
    if (await client.FillBufferAsync(ct) == 0) { break; }
}

Zero is the end of the stream, not an error — PORTING.md §2. It reads directly into the state machine's own buffer through SpaceMemory(), so no byte is copied on the way in; that, and TryNextRecord(out RecordRef) handing back a reference into the same buffer, is what makes the pair allocation-free per record.

A read the socket can already satisfy costs nothing, and that is deliberate. The read is started before any timeout machinery exists, and when it completes synchronously — the ordinary case on a stream with bytes waiting — this returns without building a CancellationTokenSource, without registering a callback, and without boxing an async state machine. Those three allocations are what a naive shape would pay on every call, and they are what the allocation assertion in the test suite would find. The read budget therefore applies only to a read that actually waits, which is the only read it was ever describing.

Cancellation ends the session here, where upstream's is cancel-safe. Upstream's fill_buf can be dropped mid-read inside a tokio::select! and lose nothing, because tokio's AsyncRead guarantees a cancelled read consumed nothing. .NET makes no such guarantee about a socket read, and bytes taken off the socket but not handed back are not a lost read — they are a decoder that silently resumes mid-record. So a cancelled fill marks the client IsClosed rather than pretending the stream is still intact. Use the token to stop the loop, not to pause it.

The obvious repair — race the read against the token and keep the pending Task for the next call — was rejected: the buffer that read is writing into belongs to the state machine, and the next SpaceMemory() may shift it underneath an in-flight read. That trades a detectable failure for a data race.

Exceptions

InvalidOperationException: The session has not been started. A session that has ended returns 0 instead — see IsClosed.
HeartbeatTimeoutException: Nothing arrived within EffectiveReadTimeout. The connection is torn down.

ReconnectAsync(CancellationToken)

Closes the current connection and opens a fresh one to the same address, running the handshake again. Subscriptions are kept but not replayed — see ResubscribeAsync(CancellationToken).

public Task ReconnectAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken: Cancels the connect, then the handshake. See the remarks on AuthenticateAsync(CancellationToken) for what cancelling a handshake costs.

Returns

Task

Remarks

Port of upstream's reconnect(). It reuses Endpoint rather than resolving the host again, which is deliberate upstream and is why that property survives CloseAsync(): a reconnect should reach the same gateway instance, and DNS may have moved on. Neither Gateway nor Dataset is consulted here at all.

What it replaces: the socket, the handshake, and therefore Greeting and SessionId — a reconnect is a new session and the gateway issues a new id for it. Metadata is cleared, and IsClosed goes back to false.

What it does not do: replay subscriptions, or start the session. Both are separate calls because both are the caller's decision — upstream keeps reconnect and resubscribe apart for exactly that reason, and fusing them into an auto-reconnect would replay subscriptions a caller may no longer want. The full sequence after a stream ends is ReconnectAsync(CancellationToken), ResubscribeAsync(CancellationToken), StartAsync(CancellationToken).

A close that fails does not stop the reconnect, matching upstream, which logs a warning and carries on. The connection being replaced is by definition the broken one, and refusing to replace it because it would not shut down politely is strictly worse than replacing it.

The subscription id counter is not reset, where upstream sets it back to zero. Upstream's resubscribe then raises it to the highest id it replayed, so in the ordinary reconnect-then-resubscribe sequence the two agree exactly. They differ only when a caller reconnects and subscribes to something new without resubscribing: upstream hands out id 1 again while its retained list still holds a different subscription with that id, so Subscriptions would carry two entries the gateway cannot tell apart in an error. A monotonic counter costs nothing on the wire — the id is a correlation handle, not a sequence the gateway checks — and it cannot produce that pair. See PORTING.md §4.

Exceptions

InvalidOperationException: This client has never connected, so there is no address to reuse.
ConnectTimeoutException: The attempt outlived ConnectTimeout.
LiveConnectException: The attempt failed — refused, or unreachable.
DatabentoAuthenticationException: The gateway rejected the credentials.
LiveProtocolException: The gateway sent something that is not a handshake.
AuthTimeoutException: The handshake outlived AuthTimeout.

RecordsAsync(CancellationToken)

The record stream as an IAsyncEnumerable<T>: the same loop as FillBufferAsync(CancellationToken) and TryNextRecord(out RecordRef), with each record copied so it can cross the yield.

public IAsyncEnumerable<OwnedRecord> RecordsAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken: Stops the enumeration and ends the session.

Returns

IAsyncEnumerable<OwnedRecord>: Every record the gateway sends, in order, until the stream ends.

Remarks

This is the convenient surface, and it copies — necessarily. yield return carries the same restriction await does, so a ref struct cannot leave an iterator at all. Each record therefore arrives as an OwnedRecord: two allocations, stated on that type rather than hidden here. Callers who need the zero-copy guarantee want the FillBufferAsync(CancellationToken)/TryNextRecord(out RecordRef) pair, which this is written in terms of and does not bypass.

The enumeration ends when the gateway closes the stream cleanly. Cancelling it ends the session, for the reason given on FillBufferAsync(CancellationToken).

Exceptions

InvalidOperationException: The session has not been started. Raised from the first enumeration step, not from this call — an iterator method's body does not run until it is enumerated.

ResubscribeAsync(CancellationToken)

Sends every subscription this client has made again, each without its replay Start. Usually the call after ReconnectAsync(CancellationToken).

public Task ResubscribeAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken: Cancels by closing the connection. See the remarks.

Returns

Task

Remarks

Port of upstream's resubscribe(). Clearing Start is the whole point of it. A reconnect that replayed the original subscriptions verbatim would ask the gateway for the same intraday history a second time, and the symptom — duplicated records after a reconnect — looks like a gateway fault and is not one. PORTING.md §4.

The retained subscriptions are cleared too, not just the lines on the wire. Subscriptions reports what was last sent, so after this every entry has a null Start — which is also what stops a second reconnect from replaying a start this one already dropped. Upstream mutates its stored subscriptions in place for the same reason; Subscription is immutable, so the entry is replaced rather than edited.

Ids are kept, not reassigned. A replayed subscription is the same subscription, and the id is what the gateway quotes when it raises an error about one. Nothing is appended to Subscriptions either — this replays the list, it does not grow it.

Not cancel-safe, and it cancels by disconnecting, exactly as SubscribeAsync(Subscription, CancellationToken) is not. A resubscribe that fails part way through has left some subscriptions sent and some not, on a socket the gateway has stopped reading; the repair is another ReconnectAsync(CancellationToken) and another call to this, which by then has no starts left to drop.

Exceptions

InvalidOperationException: The client is not connected or has not authenticated.
LiveProtocolException: The gateway closed the connection mid-write.

StartAsync(CancellationToken)

Starts the session: sends start_session and reads the DBN metadata the gateway answers with, after which records flow.

public Task<Metadata> StartAsync(CancellationToken cancellationToken = default)

Parameters

cancellationToken CancellationToken: Cancels by closing the connection. See the remarks.

Returns

Task<Metadata>: The session's DBN metadata, also kept in Metadata.

Remarks

Port of upstream's Client::start. This is the line that begins billing. Everything before it — the handshake, every subscription — moves no market data and costs nothing; the gateway sends nothing at all until this call, and then sends everything the subscriptions asked for.

Compression is settled here, not negotiated here. Compression travels on the authentication line, so by the time this runs the gateway has already decided how the stream after this point is framed. A Zstd session gets a decompressor between the socket and the state machine from the metadata block onwards; control lines were and remain plaintext, which is why DatabentoDotNet.Live.Internal.ControlChannel reads the socket directly and one byte at a time — a buffered reader would have swallowed the front of this metadata while reading the authentication response.

ts_out comes from the metadata, not from SendTsOut. What the client asked for and what the stream carries are two different facts, and only the second one determines whether each record is eight bytes longer. The state machine is built without a tsOut hint precisely so the metadata block is the only thing that can set it; a client that asked and was refused then decodes correctly rather than confidently misreading every record by eight bytes.

Not cancel-safe, and it cancels by disconnecting, for the same reason AuthenticateAsync(CancellationToken) is not: start_session is a control line, and half of one desynchronises the gateway. The wait for the metadata is bounded by EffectiveReadTimeout.

Exceptions

InvalidOperationException: The client is not connected, has not authenticated, or has already started a session.
LiveProtocolException: The gateway closed the connection before sending the metadata block.
HeartbeatTimeoutException: The metadata did not arrive within EffectiveReadTimeout.
DbnDecodeException: What the gateway sent is not valid DBN metadata.

SubscribeAsync(Subscription, CancellationToken)

Sends a subscription, splitting it across as many messages as its symbol count needs.

public Task<Subscription> SubscribeAsync(Subscription subscription, CancellationToken cancellationToken = default)

Parameters

subscription Subscription: What to subscribe to.
cancellationToken CancellationToken: Cancels by closing the connection. See the remarks.

Returns

Task<Subscription>: The subscription as sent, with Id filled in.

Remarks

Port of upstream's Client::subscribe and the Protocol::subscribe it calls. Each message is one line:

schema={s}|stype_in={t}|symbols={csv}|snapshot={0|1}|is_last={0|1}[|start={unix_nanos}]|id={n}

with at most ChunkSize symbols per line and is_last=1 on the final one only. A gateway that sees is_last=1 treats the subscription as complete, so the flag is what makes a chunked subscription one subscription rather than several partial ones.

Returning what was sent, rather than nothing. Upstream mutates the caller's Subscription to record the id it assigned; Subscription is immutable, so the sent form comes back instead. It is also appended to Subscriptions, which is what ResubscribeAsync(CancellationToken) replays.

Subscribing is legal before and after the session starts. Both are the same code path on the same socket — the gateway distinguishes them, this client does not need to.

Not cancel-safe, and it cancels by disconnecting, for the same reason AuthenticateAsync(CancellationToken) is not: a half-written subscription line desynchronises the gateway, which closes the connection. Cancelling tears the socket down rather than abandoning a partial write, and any failure here leaves the client disconnected. PORTING.md §4.

Exceptions

ArgumentNullException: subscription is null.
ArgumentException: The subscription combines a snapshot with a replay start, asks for a snapshot on a schema other than Mbo, or names no symbols. Nothing is written to the socket.
InvalidOperationException: The client is not connected or has not authenticated, or every subscription id has been used.
LiveProtocolException: The gateway closed the connection mid-write.

TryNextRecord(out RecordRef)

Decodes the next record already in the buffer, without touching the socket.

public bool TryNextRecord(out RecordRef record)

Parameters

record RecordRef: Receives the decoded record. Valid only until the next call on this client — it points into the decoder's buffer, which the next FillBufferAsync(CancellationToken) may move. Copy what you need out of it, or use RecordsAsync(CancellationToken), which does that for you.

Returns

bool: true when a record was decoded. false means the buffer holds no complete record — call FillBufferAsync(CancellationToken) and try again, and check IsClosed to tell "not yet" from "not ever".

Examples

// Synchronous, and it has to be: an async method cannot return a ref struct, so there is no
// Task<RecordRef> and never can be. The await lives in FillBufferAsync, one level out.
while (client.TryNextRecord(out RecordRef record))
{
    if (record.TryGet(out TradeMsg trade))
    {
        Console.WriteLine($"{DbnTime.ToInstant(trade.IndexTs)} {trade.Price} x {trade.Size}");
    }
}

Remarks

Port of upstream's try_next_record(). Its single-call next_record() has no .NET equivalent and never can: an async method cannot return a ref struct, so there is no Task<RecordRef>. A RecordRef local inside an async method is fine — only one that survives an await is rejected, as CS4007, which is the lifetime rule the sentence above states by hand. PORTING.md §1.

Exceptions

InvalidOperationException: The session has not been started. A session that has ended returns false instead — see IsClosed.
DbnDecodeException: The buffered bytes are not valid DBN.

Table of Contents

Class LiveClient

Examples

Remarks

Fields

DefaultAuthTimeout

Field Value

DefaultConnectTimeout

Field Value

DefaultReadTimeout

Field Value

MaxHeartbeatInterval

Field Value

MinHeartbeatInterval

Field Value

ReadTimeoutHeartbeatMargin

Field Value

Properties

ApiKey

Property Value

AuthTimeout

Property Value

Remarks

Compression

Property Value

Remarks

ConnectTimeout

Property Value

Dataset

Property Value

EffectiveReadTimeout

Property Value

Remarks

Endpoint

Property Value

Gateway

Property Value

Greeting

Property Value

Remarks

HeartbeatInterval

Property Value

Remarks

Exceptions

IsAuthenticated

Property Value

IsClosed

Property Value

Remarks

IsConnected

Property Value

IsSessionStarted

Property Value

Metadata

Property Value

Remarks

ReadTimeout

Property Value

Remarks

Exceptions

SendTsOut

Property Value

SessionId

Property Value

Remarks

SlowReaderBehavior

Property Value

Subscriptions

Property Value

Remarks

UpgradePolicy

Property Value

Methods

AuthenticateAsync(CancellationToken)

Parameters

Returns

Remarks

Exceptions

CloseAsync()

Returns