Struct Subscription 

pub struct Subscription<T: StreamDecoder<T>> { /* private fields */ }

Asynchronous subscription for streaming data.

Subscription<T> implements futures::Stream with Item = Result<SubscriptionItem<T>, Error>:

• None — the stream has ended.
• Some(Ok(SubscriptionItem::Data(t))) — a decoded value.
• Some(Ok(SubscriptionItem::Notice(n))) — a non-fatal IB notice (a warning code in WARNING_CODE_RANGE or order-cancel code 202) carried on this subscription’s request_id; the stream stays open.
• Some(Err(e)) — terminal error; subsequent calls return None.

Consume via StreamExt:

let client = Client::connect("127.0.0.1:4002", 100).await?;
let contract = Contract::stock("AAPL").build();
let mut subscription = client.market_data(&contract).subscribe().await?;

while let Some(item) = subscription.next().await {
    match item {
        Ok(SubscriptionItem::Data(tick))   => println!("tick: {tick:?}"),
        Ok(SubscriptionItem::Notice(n))    => eprintln!("notice: {n}"),
        Err(e)                             => { eprintln!("error: {e}"); break; }
    }
}

Dropping sends the cancel from a task spawned on the client’s runtime, so a subscription may be dropped on any thread, including one with no Tokio runtime. If the client’s runtime has shut down, nothing is sent: the connection went with it.

Clones share the request: dropping a clone only ends that handle, and the request is cancelled when the last clone is dropped. For a shared stream the clones count as one subscription. cancel on any clone ends the request for every clone; cancel_and_drain writes the cancel while the other clones keep reading up to TWS’s end.

When you only care about data, use the SubscriptionItemStreamExt::filter_data adapter to filter notices (logged at warn!):

let mut data = subscription.filter_data();
while let Some(result) = data.next().await { /* result: Result<TickTypes, _> */ }

Notices that are not tied to a specific subscription — connectivity codes 1100/1101/1102, farm-status 2104/2105/2106/2107/2108, etc. — are not delivered here. Subscribe to them via Client::notice_stream instead.

Implementations

impl<T: StreamDecoder<T>> Subscription<T>

pub fn request_id(&self) -> Option<i32>

Get the request ID associated with this subscription

impl<T: StreamDecoder<T> + Send + 'static> Subscription<T>

pub async fn collect_for(&mut self, timeout: Duration) -> Vec<T>

Collects data items into a Vec, bounded by a total wall-clock timeout.

Drives the subscription until the first of: the timeout elapses, the stream ends, a snapshot-end sentinel arrives (e.g. TickTypes::SnapshotEnd), or a terminal error occurs. Notices are filtered (logged at warn!); the snapshot-end sentinel is not included in the returned Vec. On a terminal error the items collected so far are returned (the error is logged at warn!). A broadcast lag (SUBSCRIPTION_LAG_CODE) counts as one: collection stops at the gap, so the rows have no hole.

This is the one-shot snapshot terminal: combined with MarketDataBuilder::snapshot, the request returns one round of data ending in a snapshot sentinel, so timeout acts only as a safety bound. Equivalent to collect_until with a predicate that never fires.

Examples

use ibapi::prelude::*;
use std::time::Duration;

#[tokio::main]
async fn main() {
    let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
    let contract = Contract::stock("AAPL").build();
    let mut subscription = client.market_data(&contract).snapshot().subscribe().await.expect("request failed");

    let ticks = subscription.collect_for(Duration::from_secs(5)).await;
    println!("collected {} ticks", ticks.len());
}

pub async fn collect_until( &mut self, timeout: Duration, stop: impl FnMut(&[T]) -> bool, ) -> Vec<T>

Collects data items into a Vec, stopping early once stop is satisfied.

Like collect_for, but after each item is appended the stop predicate is called with the full accumulated slice; returning true ends collection (the triggering item is included). Use it to stop as soon as the fields of interest are populated, rather than waiting out the whole timeout. The same timeout / stream-end / snapshot-end / terminal-error bounds as collect_for still apply.

Examples

use ibapi::market_data::realtime::TickTypes;
use ibapi::prelude::*;
use std::time::Duration;

#[tokio::main]
async fn main() {
    let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
    let contract = Contract::stock("AAPL").build();
    let mut subscription = client.market_data(&contract).snapshot().subscribe().await.expect("request failed");

    // Stop as soon as a price tick has arrived.
    let ticks = subscription
        .collect_until(Duration::from_secs(5), |ticks| {
            ticks.iter().any(|t| matches!(t, TickTypes::Price(_) | TickTypes::PriceSize(_)))
        })
        .await;
    println!("collected {} ticks", ticks.len());
}

pub async fn collect_to_end_within( &mut self, timeout: Duration, ) -> Result<Vec<T>, Error>

Collects every data item until TWS’s end marker, or fails once timeout passes.

For requests that end, such as contract details, where a partial result is not a result. Unlike collect_for, running out of time is an error, not a shorter Vec:

Outcome	Result
End marker (or snapshot end)	Ok(rows)
Terminal error, such as a TWS rejection (Error::Notice)	that error
Consumer fell behind and frames were dropped	Error::Notice with SUBSCRIPTION_LAG_CODE
Stream closed without the end marker	Error::UnexpectedEndOfStream
timeout passed	Error::Timeout; rows read so far are dropped

Notices are logged at warn!. A stream with no end marker (market data) always ends in Timeout; use collect_for for those.

After a timeout the request may still be running at TWS. Dropping the subscription sends TWS’s cancel, where the request type has one; to learn whether the request is over before reusing its id, call cancel_and_drain.

Examples

use ibapi::prelude::*;
use ibapi::Error;
use std::time::Duration;

#[tokio::main]
async fn main() {
    let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
    let contract = Contract::stock("AAPL").build();
    let mut subscription = client.contract_details_stream(&contract).subscribe().await.expect("request failed");

    match subscription.collect_to_end_within(Duration::from_secs(30)).await {
        Ok(details) => println!("{} contracts", details.len()),
        Err(Error::Timeout) => eprintln!("no answer in 30s"), // drop sends the cancel
        Err(e) => eprintln!("request failed: {e}"),
    }
}

pub async fn next_batch( &mut self, quiet: Duration, ) -> Option<Result<Vec<T>, Error>>

Returns the next batch of data items: those that arrive together, closed by quiet without an item or by a batch-end marker.

For streams that send an initial dump ending in an end marker, then push updates without one: account summary, account updates, positions and their multi variants. Each push closes once quiet passes without a row; choose quiet longer than the gap between the rows of one push. A gap longer than quiet inside the initial dump splits it: the batch ending in the marker completes it, so check batch.last().

On a stream that never pauses for quiet, such as market data, a batch never closes.

Waits without a time limit until the first item arrives.

Outcome	Result
Batch-end marker	Some(Ok(rows)), the marker last
quiet passed after a row	Some(Ok(rows))
Stream ended	Some(Ok(rows)) if any were read, else None
Terminal error	Some(Err(_)); rows of the partial batch are dropped
Consumer fell behind and frames were dropped	Some(Err(Error::Notice)) with SUBSCRIPTION_LAG_CODE; the stream stays open for the next batch

Notices are logged at warn!.

Not cancel-safe: dropping the future, as tokio::select! does with the branches that lose, drops the rows of the partial batch.

Examples

use ibapi::prelude::*;
use ibapi::accounts::PositionUpdate;
use std::time::Duration;

#[tokio::main]
async fn main() {
    let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");
    let mut subscription = client.positions().await.expect("request failed");

    while let Some(batch) = subscription.next_batch(Duration::from_secs(1)).await {
        let batch = batch.expect("positions error");
        let initial = matches!(batch.last(), Some(PositionUpdate::PositionEnd));
        println!("{} updates (initial dump: {initial})", batch.len());
    }
}

pub async fn cancel_and_drain(self, deadline: Instant) -> Result<Drained, Error>

Cancel the request and wait, up to deadline, for TWS to confirm it is over, discarding anything that arrives meanwhile.

Use it when the request id must be known finished before it’s reused, or before the next request under strict pacing. Dropping the subscription also cancels, but doesn’t wait for TWS.

Outcome	Meaning
Drained::Ended	TWS sent the end marker, before or after the cancel. Nothing is written if it (or a snapshot’s end) had already arrived.
Drained::Rejected	TWS answered with an error for the request, before or during the drain.
Drained::Unconfirmed	The deadline passed, or the stream had already ended with an error on our side (such as Error::BufferLimitExceeded). Treat the id as possibly live.
Err	The connection reset or the client shut down meanwhile.

The cancel is TWS’s native one, where the request type has one (contract details: server 215+); otherwise nothing is written and the drain waits for a natural end. Observed live (server 225), TWS kept sending contract details after the cancel: the rest of a prepared 714-row result, then the end marker; and 8,000+ rows over 150 s of an unfiltered option query with no end marker yet. So the drain usually waits out the whole result: size deadline for it, and prefer narrow queries. A stream with no end marker (market data) always ends Unconfirmed. Subscriptions without a request id (shared, order and order-update streams) are cancelled as by cancel and return Unconfirmed immediately.

Dropping the returned future mid-drain is safe: the cancel write runs in its own task and completes, the subscription drops, and nothing is written twice.

Examples

use ibapi::prelude::*;
use tokio::time::{Duration, Instant};

#[tokio::main]
async fn main() {
    let client = Client::connect("127.0.0.1:4002", 100).await.expect("connection failed");

    let contract = Contract::stock("AAPL").build();
    let mut subscription = client.contract_details_stream(&contract).subscribe().await.expect("request failed");
    let first = subscription.next().await;
    println!("first row: {first:?}");

    match subscription.cancel_and_drain(Instant::now() + Duration::from_secs(10)).await {
        Ok(Drained::Ended) | Ok(Drained::Rejected(_)) => println!("request finished at TWS"),
        Ok(Drained::Unconfirmed) => println!("not confirmed; don't reuse the request slot yet"),
        Err(e) => eprintln!("session error: {e}"),
    }
}

impl<T: StreamDecoder<T>> Subscription<T>

pub async fn cancel(&self)

Cancel the subscription.

Writes TWS’s cancel unless the request has already finished (end marker, snapshot end, or a TWS error its type knows ends the request) or its type has none. Either way the stream stops, for every clone: the next poll yields Err(Error::Cancelled), frames not yet read are skipped, and the stream ends. A clone already waiting in next() is woken to yield it. Idempotent.

Trait Implementations

impl<T: StreamDecoder<T>> Clone for Subscription<T>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T: StreamDecoder<T> + Send + 'static> Stream for Subscription<T>

type Item = Result<SubscriptionItem<T>, Error>

Values yielded by the stream.

fn poll_next( self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll<Option<Self::Item>>

Attempt to pull out the next value of this stream, registering the current task for wakeup if the value is not yet available, and returning None if the stream is exhausted.

fn size_hint(&self) -> (usize, Option<usize>)

Returns the bounds on the remaining length of the stream.