Struct Emitter 

pub struct Emitter { /* private fields */ }

Moves events from a connector’s reader task into the consumer’s channel, without ever blocking.

This is the egress half of a connector. Each subscription has a buffer with its [Backpressure] policy. emit puts events into the buffers and moves as many as the channel accepts into it with try_send. When the consumer is slow the channel fills and the buffers take the strain:

Policy	A slow consumer causes
[Backpressure::Conflate]	only the latest value is kept, older ones are replaced
[Backpressure::DropNewest]	new events are thrown away once the buffer is full
[Backpressure::Lossless]	the buffer overflows: an [Event::Overflow] notice is queued once, the event is lost, the emitter becomes degraded

Degraded means events were lost that must not be lost. The connector copies the flag (is_degraded) into its command gate, so trading commands fail with [brokapi_core::Error::Degraded] until a successful reconcile calls clear_degraded. Why: after a hole in the order or fill stream the client’s picture of its orders is wrong, and trading on it could be harmful.

Things to know:

• Events with no subscription (sub is None) skip the buffers. If the channel is full or closed they are dropped and counted in dropped_unrouted.
• Resync notices ([Event::Resync], made by the protocol cores after a sequence gap or a reconnect) travel like any other event of their subscription. The emitter does not create them and does not treat them specially.
• Each envelope is stamped with the time it is built, which is when it leaves the buffer, not when the event was parsed. For a slow consumer the stamp is later than the arrival.
• The [Event::Overflow] notice has its own queue, which is sent before the buffers. So it can reach the consumer before older events of the same subscription that are still waiting. Treat it as “events of this subscription are missing, re-read the state”, not as an ordered marker.
• Events for a subscription that was not registered are discarded.
• All methods take &self and are safe to call from several tasks. Share the emitter as an Arc.

Examples

use brokapi_core::{Backpressure, Event, SubscriptionId};
use brokapi_runtime::Emitter;
use brokapi_sansio::CoreEvent;
use tokio::sync::mpsc;

// A channel that holds one envelope, and a subscription whose buffer holds one event.
let (tx, mut rx) = mpsc::channel(1);
let emitter = Emitter::new("demo".into(), tx);
let sub = SubscriptionId::new(1);
emitter.register(sub, None, Backpressure::Lossless { cap: 1 });

let event = || CoreEvent { sub: Some(sub), payload: Event::Balances(vec![]) };
// The consumer does not read. The first event fills the channel, the second waits in a
// slot, the third fills the buffer, the fourth finds no room.
for _ in 0..4 {
    emitter.emit(vec![event()]);
}
assert!(emitter.is_degraded());
assert!(emitter.has_backlog());

// The consumer wakes up. Flushing moves what is left into the channel.
let mut overflow_notices = 0;
let mut events = 0;
for _ in 0..10 {
    while let Ok(envelope) = rx.try_recv() {
        assert_eq!(&*envelope.source, "demo");
        assert_eq!(envelope.sub, Some(sub));
        match envelope.payload {
            Event::Overflow { .. } => overflow_notices += 1,
            _ => events += 1,
        }
    }
    emitter.flush();
}
assert_eq!(overflow_notices, 1);
assert_eq!(events, 3); // the fourth event was lost

// A successful reconcile ends the degraded state.
emitter.clear_degraded();
assert!(!emitter.is_degraded());

Implementations

impl Emitter

pub fn new(source: Arc<str>, out: Sender<Envelope>) -> Arc<Emitter>

Creates an emitter that writes into out.

source is the connector label from its config (for example "bcs-ud"). It is put into every [Envelope] so the consumer can tell connectors apart.

pub fn source(&self) -> &Arc<str>

The source label given to new.

pub fn register( &self, sub: SubscriptionId, instrument: Option<InstrumentId>, policy: Backpressure, )

Starts buffering a subscription with the given policy.

Call it before the first event of sub can arrive. instrument is the subscription’s instrument, or None for account-level streams such as order state or portfolio.

pub fn unregister(&self, sub: SubscriptionId)

Stops buffering a subscription. Events still waiting for it are discarded.

An [Event::Overflow] notice already queued for the subscription is still sent.

pub fn is_degraded(&self) -> bool

Whether the connector lost events and no reconcile has cleared it yet: a Lossless subscription overflowed, or a stream of orders or fills was cut and reopened ([Event::Reconnected]).

pub fn mark_degraded(&self)

Enters the degraded state by hand: the owner knows that events were lost for a reason the emitter cannot see (a message that could not be processed). The trading gate closes until the next successful reconcile, like after an overflow. Send an [Event::Overflow] for the subscription to tell the consumer.

pub fn clear_degraded(&self)

Leaves the degraded state. Call it from a successful reconcile.

Overflowed Lossless buffers accept events again and a new overflow sends a new [Event::Overflow]. Events lost during the overflow are not recovered, the reconcile is what makes up for them. Events already waiting in the buffers are still delivered.

pub fn has_backlog(&self) -> bool

Whether some events or notices are waiting because the channel was full.

While it is true, call flush again later. emit also retries on every call.

pub fn dropped_unrouted(&self) -> u64

How many events with no subscription were dropped because the channel was full or closed.

pub fn emit(&self, events: Vec<CoreEvent>)

Buffers events and moves what the channel accepts. Never blocks.

Events with a subscription go through that subscription’s buffer and policy. A Lossless overflow queues one [Event::Overflow] (with lost_from set to the current wall clock) per subscription and marks the emitter degraded. Events with no subscription are sent straight to the channel or dropped, see the type docs. Nothing is returned: the outcome is visible through is_degraded, has_backlog and dropped_unrouted.

pub fn flush(&self)

Retries delivery of buffered events and notices. Never blocks.

Call it periodically while has_backlog is true, for example when the consumer has read from the channel.