Struct SubscriptionBuffer

Source

pub struct SubscriptionBuffer { /* private fields */ }

Expand description

Egress buffer of one subscription. It applies the subscription’s [Backpressure] policy.

The buffer never blocks and never waits. It is a plain in-memory queue (or single slot), not thread-safe by itself. BackpressureManager owns many of them, and the emitter wraps the manager in a lock.

§Examples

use brokapi_core::{
    Backpressure, Event, InstrumentId, InstrumentKey, NativeId, SubscriptionId,
};
use brokapi_runtime::{PushResult, SubscriptionBuffer};

let sber = InstrumentId::new(
    InstrumentKey::new("SBER", "TQBR"),
    NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
);
let mut buf = SubscriptionBuffer::new(SubscriptionId::new(1), sber, Backpressure::DropNewest { cap: 1 });

assert_eq!(buf.try_push(Event::Balances(vec![])), PushResult::Delivered);
// The buffer holds one event. The next one is thrown away and counted.
assert_eq!(buf.try_push(Event::Balances(vec![])), PushResult::DroppedNewest);
assert_eq!(buf.dropped(), 1);
assert_eq!(buf.drain().len(), 1);

Implementations§

Source §

impl SubscriptionBuffer

Source

pub fn new( sub: SubscriptionId, instrument: InstrumentId, policy: Backpressure, ) -> Self

Creates an empty buffer for a subscription with the given policy.

A cap of 0 in Lossless or DropNewest makes a buffer that accepts nothing: the first event already overflows (or is dropped).

Source

pub fn sub_id(&self) -> SubscriptionId

The subscription this buffer serves.

Source

pub fn instrument(&self) -> &InstrumentId

The instrument of the subscription, as given to new.

Source

pub fn is_overflowed(&self) -> bool

Whether a Lossless buffer has overflowed and not been reset yet.

Source

pub fn conflated(&self) -> u64

How many events a Conflate buffer replaced before they were taken.

Source

pub fn dropped(&self) -> u64

How many events a DropNewest buffer threw away.

Source

pub fn len(&self) -> usize

How many events wait in the buffer. For Conflate this is 0 or 1.

Source

pub fn is_empty(&self) -> bool

Whether the buffer holds no event.

Source

pub fn try_push(&mut self, event: Event) -> PushResult

Offers an event to the buffer without blocking.

Conflate: the event becomes the latest value of its kind (an order book, a quote, a portfolio and so on: one slot per kind). A value of the same kind that was still waiting is replaced and counted in conflated. Result: PushResult::Conflated.
Lossless { cap }: stored while fewer than cap events wait. The first event that finds the buffer full is not stored, the buffer is flagged overflowed and the result is PushResult::Overflow. After that every event gets Overflow and is not stored.
DropNewest { cap }: stored while fewer than cap events wait. Otherwise the new event is thrown away and counted in dropped.

Why Lossless stops for good: it is used for events that must not have a hole (orders, fills). Once one is lost, later events would hide the gap, so the consumer must be told and re-read the state.

§Examples

use brokapi_core::{Backpressure, Event, InstrumentId, InstrumentKey, NativeId, SubscriptionId};
use brokapi_runtime::{PushResult, SubscriptionBuffer};

let sber = InstrumentId::new(
    InstrumentKey::new("SBER", "TQBR"),
    NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
);
let mut buf = SubscriptionBuffer::new(SubscriptionId::new(1), sber, Backpressure::Lossless { cap: 1 });

assert_eq!(buf.try_push(Event::Balances(vec![])), PushResult::Delivered);
assert_eq!(buf.try_push(Event::Balances(vec![])), PushResult::Overflow);
assert!(buf.is_overflowed());
// Even after the consumer takes the stored event, the buffer stays overflowed.
assert_eq!(buf.drain().len(), 1);
assert_eq!(buf.try_push(Event::Balances(vec![])), PushResult::Overflow);

Source

pub fn overflow_event(&self, lost_from: UnixNanos) -> Event

Builds the [Event::Overflow] notice for this subscription.

lost_from is the wall-clock time from which events may be missing. The buffer does not send the notice itself, the owner does.

§Examples

use brokapi_core::{Backpressure, Event, InstrumentId, InstrumentKey, NativeId, SubscriptionId, UnixNanos};
use brokapi_runtime::SubscriptionBuffer;

let sber = InstrumentId::new(
    InstrumentKey::new("SBER", "TQBR"),
    NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
);
let buf = SubscriptionBuffer::new(SubscriptionId::new(7), sber, Backpressure::Lossless { cap: 1 });
let notice = buf.overflow_event(UnixNanos::from_nanos(5));
assert!(matches!(notice, Event::Overflow { sub, .. } if sub == SubscriptionId::new(7)));