Table of Contents

Class TimeseriesReader

Namespace
DatabentoDotNet.Historical
Assembly
DatabentoDotNet.Historical.dll

An asynchronous reader over a DBN record stream: what GetRangeAsync(GetRangeParams, CancellationToken) and GetRangeToFileAsync(GetRangeParams, string, CancellationToken) hand back.

public sealed class TimeseriesReader : IAsyncDisposable
Inheritance
TimeseriesReader
Implements
Inherited Members

Examples

await using var reader = await client.Timeseries.GetRangeAsync(request);

// Every DBN stream opens with a metadata block, and it echoes the request rather than describing
// the answer — so it says what was asked for even when nothing came back.
Console.WriteLine($"DBN v{reader.Metadata.Version} {reader.Metadata.Dataset}");

while (true)
{
    while (reader.TryNextRecord(out RecordRef record))
    {
        // `record` is valid until the next call on this reader, and no longer.
        if (record.TryGet(out TradeMsg trade))
        {
            Console.WriteLine($"{DbnTime.ToInstant(trade.IndexTs)} {trade.Price} x {trade.Size}");
        }
    }

    if (await reader.FillBufferAsync() == 0)
    {
        break;
    }
}

ReadRecordsAsync(CancellationToken) is that loop written once, at a copy per record — the right choice whenever a record has to outlive the iteration: 

await foreach (OwnedRecord record in reader.ReadRecordsAsync())
{
    if (record.TryGet(out TradeMsg trade))
    {
        Console.WriteLine(trade.Price);
    }
}

Remarks

Why this exists rather than DbnDecoder. Upstream's get_range returns an AsyncDbnDecoder (timeseries.rs:88-97), and the faithful port of an async decoder in this repo is not a decoder at all — it is the FillBufferAsync(CancellationToken)/TryNextRecord(out RecordRef) pair, for the reason PORTING.md §1 gives and LiveClient already implements: RecordRef is a ref struct, an async method cannot return one, and so there is no Task<RecordRef> and never can be. DbnDecoder is synchronous by design and says so on its own class comment; pointing it at an HTTP response stream would block a thread pool thread for the length of a multi-gigabyte download. This type drives DbnFsm directly, which is what that same class comment says the asynchronous clients do.

The read loop is the whole API. Drain, refill, repeat — the example below.

The inner loop must run to false before each refill. That is not a style preference: a refill may shift the buffer, which is exactly what invalidates a RecordRef the caller is still holding, and it is also the ordering TryNextRecord(out RecordRef)'s truncation check depends on. ReadRecordsAsync(CancellationToken) is this loop written once, at the cost of a copy per record.

Zstandard is not detected here, because it was requested. ToFormParameters() hard-codes compression=zstd on every request, so the caller — TimeseriesClient — knows the framing and unwraps it through Internal/ZstdDecompressor.cs before constructing this type. DbnDecoder sniffs the frame magic because it is handed files of unknown provenance; this type is handed one thing only.

Not thread-safe. One reader is one cursor over one buffer, the same call LiveClient makes about its record loop. Several readers may be read concurrently; one reader may not.

Properties

Metadata

The stream's metadata block, decoded before this object existed and already presented according to the upgrade policy.

public Metadata Metadata { get; }

Property Value

Metadata

Remarks

Never null, unlike Metadata: that one is nullable because it also decodes bare fragments, and timeseries.get_range never returns one. A response with no records still carries a full metadata block — the API answers an empty range with 200, a metadata block, no records, and an X-Warning saying it found nothing.

Methods

DisposeAsync()

Disposes the source stream unless it was left open, and with it the HTTP response it was reading from, when there is one.

public ValueTask DisposeAsync()

Returns

ValueTask

FillBufferAsync(CancellationToken)

Reads more bytes from the source into the decode buffer.

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

Parameters

cancellationToken CancellationToken

Cancels the read.

Returns

ValueTask<int>

How many bytes were read; zero at the end of the source.

Remarks

Call this only when TryNextRecord(out RecordRef) has answered false. It may shift the buffer, which invalidates every RecordRef handed out since the last call — the same contract LiveClient.FillBufferAsync carries, and the reason neither takes a record as a parameter it could invalidate.

This is where a truncated download is reported, because this is where the read loop ends. The natural loop breaks the moment this returns zero and never calls TryNextRecord(out RecordRef) again, so a check that lived only there would be unreachable from the one shape every caller writes.

The condition is exact rather than heuristic, and fires only when all three hold: TryNextRecord(out RecordRef) has already answered false at least once since the last refill, the source has just reported end of data, and BufferedByteCount is non-zero. Those bytes are the front of a record whose tail never arrived; there is no other way to reach that state. A caller who refills without draining first has not established that the leftover bytes are incomplete, so nothing is thrown — the check declines rather than guesses.

Exceptions

ObjectDisposedException

This reader has been disposed.

DbnDecodeException

The source ended part-way through a record — the download was truncated.

IOException

The connection dropped mid-body. A chunked response whose terminating chunk never arrives fails here rather than ending quietly, which is the other half of "a truncated download is an exception".

OpenAsync(Stream, VersionUpgradePolicy, bool, int, CancellationToken)

Reads the metadata block, then hands back a stream positioned at the first record.

public static ValueTask<TimeseriesReader> OpenAsync(Stream source, VersionUpgradePolicy upgradePolicy = VersionUpgradePolicy.UpgradeToV3, bool leaveOpen = false, int bufferSize = 65536, CancellationToken cancellationToken = default)

Parameters

source Stream

The decompressed DBN stream, positioned at its first byte. Read forward only; never seeked.

upgradePolicy VersionUpgradePolicy

How to present records from an older DBN version.

leaveOpen bool

true to leave source open when this reader is disposed.

bufferSize int

The read buffer's size in bytes.

cancellationToken CancellationToken

Cancels the metadata read.

Returns

ValueTask<TimeseriesReader>

A reader positioned at the first record.

Exceptions

ArgumentNullException

source is null.

DbnDecodeException

The stream does not begin with valid DBN metadata, or ends part-way through it.

ReadRecordsAsync(CancellationToken)

Every record in the stream, as an IAsyncEnumerable<T>.

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

Parameters

cancellationToken CancellationToken

Stops the enumeration.

Returns

IAsyncEnumerable<OwnedRecord>

Every record, in order, until the stream ends.

Remarks

The convenient surface, and it copies — necessarily. yield return carries the same restriction await does, so a ref struct cannot leave an iterator. Each record arrives as an OwnedRecord, whose own documentation states the cost. 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 — including its truncation check. The same split LiveClient offers.

Exceptions

DbnDecodeException

The stream was truncated, or its bytes are not valid DBN.

TryNextRecord(out RecordRef)

Decodes the next record already sitting in the buffer.

public bool TryNextRecord(out RecordRef record)

Parameters

record RecordRef

Receives the decoded record. Valid only until the next call on this stream — the bytes it points at live in the decode buffer, which the next FillBufferAsync(CancellationToken) may move.

Returns

bool

true when a record was decoded; false when the buffer holds no complete record, which means either "refill and ask again" or "the stream is finished" depending on whether FillBufferAsync(CancellationToken) has since returned zero.

Remarks

Unlike TryNextRecord(out RecordRef), this throws on a truncated stream. That decoder documents a trailing partial record as not an error and silently drops it, which is the right call for a local file that may legitimately be a fragment. It is the wrong call for a download: bytes the server promised and did not deliver are a failed request, and silently returning the records that did arrive turns a network fault into a short answer the caller has no way to distinguish from a complete one. #31 drew the same line on the metadata path.

The check is exact rather than heuristic. It fires only when all three hold: the machine has no complete record, FillBufferAsync(CancellationToken) has reported end of source, and BufferedByteCount is non-zero. Those bytes are the front of a record whose tail never arrived; there is no other way to reach that state.

Exceptions

ObjectDisposedException

This reader has been disposed.

DbnDecodeException

The source ended part-way through a record, or its bytes are not valid DBN.

Edit this page