Struct BackpressureManager 

pub struct BackpressureManager { /* private fields */ }

The buffers of all active subscriptions of one connector, plus the connector’s degraded flag.

Used from the parser task: ingest puts an event into the buffer of its subscription without blocking, and pop_ready takes the next one out for delivery. When any Lossless buffer overflows, the manager marks the whole connector degraded. A degraded connector refuses trading commands with [brokapi_core::Error::Degraded] (the caller applies that, for example through CommandGate) until a successful reconcile calls clear_degraded.

Most connectors use Emitter, which owns a manager and also moves events into the output channel.

Examples

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

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

assert_eq!(mgr.ingest(sub, Event::Balances(vec![]), UnixNanos::ZERO), PushResult::Delivered);
assert_eq!(mgr.ingest(sub, Event::Balances(vec![]), UnixNanos::ZERO), PushResult::Overflow);
assert!(mgr.is_degraded());

// The reconcile that follows an overflow clears the flag.
mgr.clear_degraded();
assert!(!mgr.is_degraded());

Implementations

impl BackpressureManager

pub fn new() -> Self

Creates a manager with no subscriptions, not degraded.

pub fn register( &mut self, sub: SubscriptionId, instrument: InstrumentId, policy: Backpressure, )

Adds a buffer for a subscription.

Registering the same sub twice creates two buffers. Only the first one is ever used, so register each subscription id once.

pub fn unregister(&mut self, sub: SubscriptionId)

Removes the buffer of a subscription. Events still waiting in it are discarded. An unknown sub is ignored.

pub fn is_degraded(&self) -> bool

Whether some Lossless buffer overflowed since the last clear_degraded.

pub fn clear_degraded(&mut self)

Leaves the degraded state and reopens every overflowed buffer.

Call it after a successful reconcile, when the consumer has re-read the state it lost. Events that were refused during the overflow are not recovered.

pub fn ingest( &mut self, sub: SubscriptionId, event: Event, lost_from: UnixNanos, ) -> PushResult

Offers an event to the buffer of sub without blocking.

An PushResult::Overflow also sets the degraded flag of the manager. The lost_from argument is accepted for the caller’s convenience and is not used by the manager: the caller builds the [Event::Overflow] notice itself (see SubscriptionBuffer::overflow_event).

An event for a sub that is not registered is discarded and the result is still PushResult::Delivered. Register the subscription before its first event can arrive.

pub fn pop_ready(&mut self, sub: SubscriptionId) -> Option<Event>

Takes the next event of sub that is ready for delivery, if any.

A Conflate subscription yields its latest value of each kind, one per call. Others yield their queue front, oldest first. Returns None for an empty buffer and for an unknown sub.

Examples

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

let sber = InstrumentId::new(
    InstrumentKey::new("SBER", "TQBR"),
    NativeId::TickerClass { ticker: "SBER".into(), class_code: "TQBR".into() },
);
let sub = SubscriptionId::new(1);
let mut mgr = BackpressureManager::new();
mgr.register(sub, sber, Backpressure::Conflate);

// Three snapshots arrive before the consumer looks. Only the last one is kept.
for _ in 0..3 {
    mgr.ingest(sub, Event::Balances(vec![]), UnixNanos::ZERO);
}
assert!(mgr.pop_ready(sub).is_some());
assert!(mgr.pop_ready(sub).is_none());

Trait Implementations

impl Debug for BackpressureManager

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for BackpressureManager

fn default() -> BackpressureManager

Returns the “default value” for a type.