Live streaming

The live path opens the streaming session, subscribes one or more Kairos analytics, and delivers each emission to a typed .on_event callback. No async runtime is needed, and no poll loop is written by hand.

Minimal program

use kairos::{Client, OhlcvcRow};

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Client::connect((
        std::env::var("THETADATA_EMAIL")?,
        std::env::var("THETADATA_PASSWORD")?,
    ))?;

    let bars = client
        .live()
        .ohlcvc(["QQQ"])
        .on_event(|bar: &OhlcvcRow| {
            println!("{} close={} volume={}", bar.bucket_start_ms, bar.close, bar.volume);
        })?;

    // The callback fires on a background worker. Keep the process alive
    // for as long as you want the stream, then stop it:
    std::thread::sleep(std::time::Duration::from_secs(60));
    bars.unsubscribe();
    Ok(())
}

.on_event builds the wire params, calls the engine over the stable C ABI, and registers the callback. Each emission arrives as the analytic's typed <Camel>Row — OhlcvcRow here.

Multiple analytics on one client

The same Client drives any number of concurrent subscriptions. Every .on_event callback in the process shares one background delivery worker, so streaming many analytics at once does not spawn a thread per stream — register each and hold its handle:

use kairos::{Client, OhlcvcRow, GreeksRow, VpinRow};

let client = Client::connect((email, password))?;

let bars = client
    .live()
    .ohlcvc(["QQQ"])
    .on_event(|bar: &OhlcvcRow| { /* … */ })?;

let greeks = client
    .live()
    .greeks(["QQQ"])
    .on_event(|row: &GreeksRow| { /* … */ })?;

let flow = client
    .live()
    .vpin(["QQQ"])
    .on_event(|row: &VpinRow| { /* … */ })?;

The engine multiplexes the underlying upstream session across every subscription, and the shared worker services all three callbacks. Keep callbacks cheap — stash the row, enqueue it, update a widget — since a slow callback delays its siblings on the shared worker; hand heavy work off to your own thread from inside the callback. The wait-strategy choice (Adaptive / Balanced / Efficient) applies once per Client; pick it at Client::builder() before connecting.

Reconnect-replay handling

The live feed's reconnect-replay window covers brief network glitches without caller intervention. On a longer disconnect the affected subscription's stream ends and its handle goes inactive (sub.is_active() returns false); the recommended recovery is to register a fresh .on_event subscription on the same client and resume.

Drop-oldest backpressure

The engine's per-subscription outbound buffer is bounded. On the live feed, drop-oldest is the only correct policy — the feed cannot be slowed without losing the reconnect-replay window. If a callback consumer falls far enough behind that the buffer overflows, the subscription ends rather than silently skipping frames forever, and its handle goes inactive. Keep the callback cheap and move heavy processing off the delivery worker to stay ahead of the feed.

A panic inside a callback is caught, counted on the subscription's panic_count(), and delivery continues with the next emission — one panicking callback never tears down the worker or the sibling subscriptions.

See also

• Consuming events — the callback delivery and panic-isolation model.
• Backpressure — the drop-oldest policy on the live feed.
• Tier quotas — what each tier permits.