Function run_pump 

pub async fn run_pump<F, H>(
    io: PumpIo<F>,
    handle: H,
    on_overflow: Arc<dyn Fn(SubscriptionId) + Send + Sync>,
)
where
    F: Send + 'static,
    H: FnMut(F, Instant) -> Vec<CoreEvent> + Send,

Runs until cancelled, the frame channel closes, or the consumer drops the receiver.

handle turns one frame into core events (it owns the core, so parsing happens right here in the task that received the frame). on_overflow is called once when a Lossless buffer overflows. The user data connector uses it to mark itself degraded.

Delivery order: control events (Resync, Overflow) first, then buffered data, one subscription after another. A Resync does not overtake the data of its own subscription that is already buffered: that data goes out just before it. The pump never waits for a slow consumer inside the core: events wait in the subscription’s buffer, and what the buffer’s policy says happens when it is full (merge for Conflate, overflow for Lossless).

When the pump is cancelled it makes one last pass and hands over what fits into out without waiting. The rest is dropped.

Examples

A frame goes in, an envelope with the source label comes out:

use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use brokapi_core::{Event, ResyncReason, SubscriptionId};
use brokapi_finam::core::{MdFrame, MdEvent};
use brokapi_finam::pump::{PumpIo, run_pump};
use brokapi_runtime::BackpressureManager;
use brokapi_sansio::CoreEvent;
use tokio::sync::mpsc;
use tokio_util::sync::CancellationToken;

let (frames, rx) = mpsc::channel::<MdFrame>(8);
let (out, mut envelopes) = mpsc::channel(8);
let cancel = CancellationToken::new();
let io = PumpIo {
    rx,
    out,
    source: Arc::from("finam-md"),
    backpressure: Arc::new(Mutex::new(BackpressureManager::new())),
    cancel: cancel.clone(),
};
// The handler stands in for the connector's core: every frame becomes one control event.
let sub = SubscriptionId::new(1);
let handler = move |_frame: MdFrame, _now: Instant| {
    vec![CoreEvent {
        sub: Some(sub),
        payload: Event::Resync {
            sub,
            instrument: brokapi_finam::util::instrument_from_symbol("SBER@MISX"),
            reason: ResyncReason::SubscriptionReset,
        },
    }]
};
let pump = tokio::spawn(run_pump(io, handler, Arc::new(|_| {})));

frames.send(MdFrame { sub, event: MdEvent::Connected }).await.unwrap();
let envelope = tokio::time::timeout(Duration::from_secs(5), envelopes.recv()).await.unwrap().unwrap();
assert_eq!(&*envelope.source, "finam-md");
assert!(matches!(envelope.payload, Event::Resync { .. }));

cancel.cancel();
pump.await.unwrap();