Class OtlpEnvelopeStore

Namespace

Kuestenlogik.Bowire.Protocol.Otlp

Assembly

Kuestenlogik.Bowire.Protocol.Otlp.dll

In-memory ring buffer of received OTLP envelopes plus a fan-out publish/subscribe channel.

public sealed class OtlpEnvelopeStore

Inheritance

object

OtlpEnvelopeStore

Inherited Members

object.GetType()

object.ToString()

object.Equals(object)

object.Equals(object, object)

object.ReferenceEquals(object, object)

object.GetHashCode()

Remarks

Phase 1 keeps the bound deliberately small (1024 envelopes) — the workbench's channel UX surfaces the latest envelopes and operators who want full retention should point their exporters at a proper collector. Phase 2 may grow the bound + add disk spill, but the passive-listener-for-debugging story stays in-memory.

The store is registered as a singleton so every POST /v1/* request and every BowireOtlpProtocol subscriber sees the same envelopes. Subscribers receive only envelopes published after they subscribe — late subscribers can call Snapshot() first to read what's already in the ring.

Constructors

OtlpEnvelopeStore()

public OtlpEnvelopeStore()

OtlpEnvelopeStore(int)

public OtlpEnvelopeStore(int capacity)

Parameters

capacity int

Properties

Capacity

Maximum number of envelopes the ring retains.

public int Capacity { get; }

Property Value

int

Count

Total envelopes currently retained in the ring.

public int Count { get; }

Property Value

int

Methods

Append(OtlpEnvelope)

Add an envelope to the ring and notify every subscriber.

public void Append(OtlpEnvelope envelope)

Parameters

envelope OtlpEnvelope

Clear()

Reset the ring (test helper). Subscribers stay attached and continue receiving new envelopes.

public void Clear()

Latest(OtlpSignalKind)

Returns the most recently appended envelope of the given kind, or null when none retained.

public OtlpEnvelope? Latest(OtlpSignalKind kind)

Parameters

kind OtlpSignalKind

Returns

OtlpEnvelope

ReplayAndSubscribeAsync(OtlpSignalKind, CancellationToken)

Yield the retained envelopes of one signal kind, then everything published from that moment on — with no gap between the two.

public IAsyncEnumerable<OtlpEnvelope> ReplayAndSubscribeAsync(OtlpSignalKind kind, CancellationToken ct = default)

Parameters

kind OtlpSignalKind

ct CancellationToken

Returns

IAsyncEnumerable<OtlpEnvelope>

Remarks

The obvious composition — call Snapshot(OtlpSignalKind), then SubscribeAsync(CancellationToken) — has a hole in it. Between the two calls the subscriber is not yet registered and the ring has already been copied, so anything appended in that window reaches neither and is lost outright. The window is small, which is exactly what makes it nasty: a live OTLP tail silently drops the exporter's first envelopes only under load, and the test that covered it papered over the race with a 50 ms sleep until a more contended runner exposed it.

Registering the writer and copying the ring under ONE lock closes it: an Append(OtlpEnvelope) is then either already in the backlog or delivered to the freshly-registered writer, never neither.

Snapshot()

Returns a snapshot copy of every envelope currently retained.

public IReadOnlyList<OtlpEnvelope> Snapshot()

Returns

IReadOnlyList<OtlpEnvelope>

Snapshot(OtlpSignalKind)

Returns the snapshot filtered to a single signal kind.

public IReadOnlyList<OtlpEnvelope> Snapshot(OtlpSignalKind kind)

Parameters

kind OtlpSignalKind

Returns

IReadOnlyList<OtlpEnvelope>

SubscribeAsync(CancellationToken)

Subscribe to incoming envelopes. The returned reader yields every envelope published after the subscription begins.

public IAsyncEnumerable<OtlpEnvelope> SubscribeAsync(CancellationToken ct = default)

Parameters

ct CancellationToken

Returns

IAsyncEnumerable<OtlpEnvelope>

An IAsyncEnumerable<T> the caller awaits across the lifetime of the subscription. The subscription stays open until the caller cancels via the supplied token.