25323281489

Committed 04 May 2026 01:58PM UTC coverage: 87.529% (-0.1%) from 87.64%

Build # 25323281489

Build Type

push

github

Committed by

web-flow

Commit Message

feat(subscriptions): widen Subscription<T> public API to SubscriptionItem<T> (closes #487) (#504)

* feat(subscriptions): widen Subscription<T> public API to SubscriptionItem<T>

Public API change (closes #487):

  Sync  next/try_next/next_timeout: Option<T>          -> Option<Result<SubscriptionItem<T>, Error>>
  Async next                       : Option<Result<T, Error>> -> Option<Result<SubscriptionItem<T>, Error>>

* New public type `SubscriptionItem<T> = Data(T) | Notice(Notice)`
  exposed at `ibapi::subscriptions::SubscriptionItem`.
* Sync drops `error()` accessor and the internal `Mutex<Option<Error>>`
  field; errors flow via the `Err` arm of every yielding method.
* New iterator adapters `iter_data` / `try_iter_data` / `timeout_iter_data`
  on sync and `next_data` on both sync + async filter notices and yield
  `Result<T, Error>` for callers that only care about data.
* Existing `iter` / `try_iter` / `timeout_iter` widen to yield
  `Result<SubscriptionItem<T>, Error>`; `IntoIterator` impls match.

Notice arm is structurally present but unreachable until PR 3 emits notices
from the dispatcher.

* refactor(consumers): migrate to SubscriptionItem-aware API

Consumer-side sweep for the widened `Subscription<T>` API:

* Library tests, integration tests, examples, and doc tests now use
  `next_data()` / `iter_data()` (filtering notices to keep legacy-shape
  return) where they previously expected `T` directly.
* Dropped all `subscription.error()` callers — errors flow via `Err` arm
  of `next_data()` / `iter_data()` items.
* Restructured `examples/sync/stream_retry.rs` retry loop with a labeled
  loop so `Err(Error::ConnectionReset)` from the iterator can `continue`
  the outer retry.
* Restructured `examples/sync/bracket_order.rs` to use
  `subscription.try_iter_data().next()` for mutable counter access.
* Doc tests on `accounts`, `client`, `historical`, `realtime`, `news`,
  `orders`, `scanner` rewritten to demonstrate the new shape.

* chore: dr... (continued)

Coverage Stats

48 of 79 new or added lines in 6 files covered. (60.76%)

4 existing lines in 1 file now uncovered.

16978 of 19397 relevant lines covered (87.53%)

26.25 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

63.98

/src/subscriptions/sync.rs

//! Synchronous subscription implementation

use std::marker::PhantomData;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use log::{debug, error, warn};

use super::common::{process_decode_result, DecoderContext, ProcessingResult, SubscriptionItem};
use super::StreamDecoder;
use crate::errors::Error;
use crate::messages::{OutgoingMessages, ResponseMessage};
use crate::transport::{InternalSubscription, MessageBus};

/// A [Subscription] is a stream of responses returned from TWS. A [Subscription] is normally returned when invoking an API that can return more than one value.
///
/// Each call to [next](Subscription::next), [try_next](Subscription::try_next), or
/// [next_timeout](Subscription::next_timeout) returns
/// `Option<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; the stream stays open.
/// * `Some(Err(e))` — terminal error; subsequent calls return `None`.
///
/// When you only care about data, use [`iter_data`](Subscription::iter_data) (or
/// [`next_data`](Subscription::next_data)) which filters notices for you.
#[allow(private_bounds)]
pub struct Subscription<T: StreamDecoder<T>> {
    context: DecoderContext,
    message_bus: Arc<dyn MessageBus>,
    request_id: Option<i32>,
    order_id: Option<i32>,
    message_type: Option<OutgoingMessages>,
    phantom: PhantomData<T>,
    cancelled: AtomicBool,
    snapshot_ended: AtomicBool,
    stream_ended: AtomicBool,
    subscription: InternalSubscription,
}

enum NextAction<T> {
    Return(Option<T>),
    Skip,
}

#[allow(private_bounds)]
impl<T: StreamDecoder<T>> Subscription<T> {
    pub(crate) fn new(message_bus: Arc<dyn MessageBus>, subscription: InternalSubscription, context: DecoderContext) -> Self {
        let request_id = subscription.request_id;
        let order_id = subscription.order_id;
        let message_type = subscription.message_type;

        Subscription {
            context,
            message_bus,
            request_id,
            order_id,
            message_type,
            subscription,
            phantom: PhantomData,
            cancelled: AtomicBool::new(false),
            snapshot_ended: AtomicBool::new(false),
            stream_ended: AtomicBool::new(false),
        }
    }

    /// Cancel the subscription
    pub fn cancel(&self) {
        // Skip on snapshot subscriptions whose data already arrived.
        if self.snapshot_ended.load(Ordering::Relaxed) {
            return;
        }

        if self.cancelled.load(Ordering::Relaxed) {
            return;
        }

        self.cancelled.store(true, Ordering::Relaxed);

        if let Some(request_id) = self.request_id {
            if let Ok(message) = T::cancel_message(self.context.server_version, self.request_id, Some(&self.context)) {
                if let Err(e) = self.message_bus.cancel_subscription(request_id, &message) {
                    warn!("error cancelling subscription: {e}")
                }
                self.subscription.cancel();
            }
        } else if let Some(order_id) = self.order_id {
            if let Ok(message) = T::cancel_message(self.context.server_version, self.request_id, Some(&self.context)) {
                if let Err(e) = self.message_bus.cancel_order_subscription(order_id, &message) {
                    warn!("error cancelling order subscription: {e}")
                }
                self.subscription.cancel();
            }
        } else if let Some(message_type) = self.message_type {
            if let Ok(message) = T::cancel_message(self.context.server_version, self.request_id, Some(&self.context)) {
                if let Err(e) = self.message_bus.cancel_shared_subscription(message_type, &message) {
                    warn!("error cancelling shared subscription: {e}")
                }
                self.subscription.cancel();
            }
        } else {
            debug!("Could not determine cancel method")
        }
    }

    /// Returns the request ID associated with this subscription.
    pub fn request_id(&self) -> Option<i32> {
        self.request_id
    }

    /// Returns the next item, blocking until one is available.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ibapi::client::blocking::Client;
    /// use ibapi::contracts::Contract;
    /// use ibapi::subscriptions::SubscriptionItem;
    ///
    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
    /// let contract = Contract::stock("AAPL").build();
    /// let subscription = client.market_data(&contract)
    ///     .generic_ticks(&["233"])
    ///     .subscribe()
    ///     .expect("market data request failed");
    ///
    /// while let Some(result) = subscription.next() {
    ///     match result {
    ///         Ok(SubscriptionItem::Data(tick))   => println!("tick: {tick:?}"),
    ///         Ok(SubscriptionItem::Notice(n))    => eprintln!("notice: {n}"),
    ///         Err(e)                             => { eprintln!("error: {e}"); break; }
    ///     }
    /// }
    /// ```
    pub fn next(&self) -> Option<Result<SubscriptionItem<T>, Error>> {
        if self.stream_ended.load(Ordering::Relaxed) {
            return None;
        }

        loop {
            match self.handle_response(self.subscription.next()) {
                NextAction::Return(val) => return val,
                NextAction::Skip => continue,
            }
        }
    }

    fn handle_response(&self, response: Option<Result<ResponseMessage, Error>>) -> NextAction<Result<SubscriptionItem<T>, Error>> {
        match response {
            Some(Ok(mut message)) => match process_decode_result(T::decode(&self.context, &mut message)) {
                ProcessingResult::Success(val) => {
                    if val.is_snapshot_end() {
                        self.snapshot_ended.store(true, Ordering::Relaxed);
                    }
                    NextAction::Return(Some(Ok(SubscriptionItem::Data(val))))
                }
                ProcessingResult::Skip => {
                    log::trace!("skipping unexpected message on shared channel");
                    NextAction::Skip
                }
                ProcessingResult::EndOfStream => {
                    self.stream_ended.store(true, Ordering::Relaxed);
                    NextAction::Return(None)
                }
                ProcessingResult::Error(err) => {
                    match &err {
                        Error::Message(code, msg) => warn!("subscription terminated by TWS error [{code}] {msg}"),
                        _ => error!("error decoding message: {err}"),
                    }
                    self.stream_ended.store(true, Ordering::Relaxed);
                    NextAction::Return(Some(Err(err)))
                }
            },
            Some(Err(Error::EndOfStream)) => {
                self.stream_ended.store(true, Ordering::Relaxed);
                NextAction::Return(None)
            }
            Some(Err(e)) => {
                self.stream_ended.store(true, Ordering::Relaxed);
                NextAction::Return(Some(Err(e)))
            }
            None => NextAction::Return(None),
        }
    }

    /// Returns the next item without blocking.
    ///
    /// Returns `None` if no item is available *right now*; check the surrounding
    /// loop or stream state to distinguish from end-of-stream.
    pub fn try_next(&self) -> Option<Result<SubscriptionItem<T>, Error>> {
        if self.stream_ended.load(Ordering::Relaxed) {
            return None;
        }
        loop {
            match self.handle_response(self.subscription.try_next()) {
                NextAction::Return(val) => return val,
                NextAction::Skip => continue,
            }
        }
    }

    /// Returns the next item, blocking up to `timeout`.
    pub fn next_timeout(&self, timeout: Duration) -> Option<Result<SubscriptionItem<T>, Error>> {
        if self.stream_ended.load(Ordering::Relaxed) {
            return None;
        }
        let deadline = Instant::now() + timeout;
        loop {
            let remaining = deadline.saturating_duration_since(Instant::now());
            if remaining.is_zero() {
                return None;
            }
            match self.handle_response(self.subscription.next_timeout(remaining)) {
                NextAction::Return(val) => return val,
                NextAction::Skip => continue,
            }
        }
    }

    /// Convenience: blocking `next` that filters out notices and yields just data.
    /// Equivalent to `iter_data().next()`.
    pub fn next_data(&self) -> Option<Result<T, Error>> {
        self.iter_data().next()
    }

    /// Blocking iterator yielding `Result<SubscriptionItem<T>, Error>`. Use
    /// [`iter_data`](Subscription::iter_data) when you only want data.
    pub fn iter(&self) -> SubscriptionIter<'_, T> {
        SubscriptionIter { subscription: self }
    }

    /// Non-blocking iterator. Returns `None` immediately when nothing is queued.
    pub fn try_iter(&self) -> SubscriptionTryIter<'_, T> {
        SubscriptionTryIter { subscription: self }
    }

    /// Iterator that waits up to `timeout` for each item.
    pub fn timeout_iter(&self, timeout: Duration) -> SubscriptionTimeoutIter<'_, T> {
        SubscriptionTimeoutIter { subscription: self, timeout }
    }

    /// Blocking iterator that filters notices and yields `Result<T, Error>`.
    /// Notices are logged at `warn!` level.
    pub fn iter_data(&self) -> FilterData<SubscriptionIter<'_, T>> {
        self.iter().filter_data()
    }

    /// Non-blocking data iterator (notices filtered).
    pub fn try_iter_data(&self) -> FilterData<SubscriptionTryIter<'_, T>> {
        self.try_iter().filter_data()
    }

    /// Timeout-bounded data iterator (notices filtered).
    pub fn timeout_iter_data(&self, timeout: Duration) -> FilterData<SubscriptionTimeoutIter<'_, T>> {
        self.timeout_iter(timeout).filter_data()
    }
}

impl<T: StreamDecoder<T>> Drop for Subscription<T> {
    /// Cancel subscription on drop
    fn drop(&mut self) {
        debug!("dropping subscription");
        self.cancel();
    }
}

/// Adapter that filters `SubscriptionItem::Notice` items (logging them at `warn!`)
/// from any `Iterator<Item = Result<SubscriptionItem<T>, Error>>` and yields the
/// underlying `Result<T, Error>` to the caller.
///
/// Returned by [`SubscriptionItemIterExt::filter_data`].
#[must_use = "iterator adapters are lazy and do nothing unless consumed"]
pub struct FilterData<I> {
    inner: I,
}

impl<I, T> Iterator for FilterData<I>
where
    I: Iterator<Item = Result<SubscriptionItem<T>, Error>>,
{
    type Item = Result<T, Error>;

    fn next(&mut self) -> Option<Self::Item> {
        loop {
            match self.inner.next()? {
                Ok(SubscriptionItem::Data(t)) => return Some(Ok(t)),
                Ok(SubscriptionItem::Notice(n)) => {
                    log::warn!("ib notice on subscription: {n}");
                    continue;
                }
                Err(e) => return Some(Err(e)),
            }
        }
    }
}

/// Extension trait that adds [`filter_data`](SubscriptionItemIterExt::filter_data)
/// to any iterator yielding `Result<SubscriptionItem<T>, Error>`. Use it to compose
/// the data-only flow with iterator combinators that the built-in
/// [`iter_data`](Subscription::iter_data) family doesn't already cover, e.g.
/// `subscription.iter().take(10).filter_data()`.
pub trait SubscriptionItemIterExt: Iterator + Sized {
    /// Wrap `self` in a [`FilterData`] adapter that drops `SubscriptionItem::Notice`
    /// items (logging them) and yields the underlying `Result<T, Error>`.
    fn filter_data<T>(self) -> FilterData<Self>
    where
        Self: Iterator<Item = Result<SubscriptionItem<T>, Error>>,
    {
        FilterData { inner: self }
    }
}

impl<I: Iterator> SubscriptionItemIterExt for I {}

/// Blocking iterator over `Result<SubscriptionItem<T>, Error>`.
#[allow(private_bounds)]
#[must_use = "iterators are lazy and do nothing unless consumed"]
pub struct SubscriptionIter<'a, T: StreamDecoder<T>> {
    subscription: &'a Subscription<T>,
}

impl<T: StreamDecoder<T>> Iterator for SubscriptionIter<'_, T> {
    type Item = Result<SubscriptionItem<T>, Error>;

    fn next(&mut self) -> Option<Self::Item> {
        self.subscription.next()
    }
}

impl<'a, T: StreamDecoder<T>> IntoIterator for &'a Subscription<T> {
    type Item = Result<SubscriptionItem<T>, Error>;
    type IntoIter = SubscriptionIter<'a, T>;

    fn into_iter(self) -> Self::IntoIter {
        self.iter()
    }
}

/// Owned blocking iterator over `Result<SubscriptionItem<T>, Error>`.
#[allow(private_bounds)]
#[must_use = "iterators are lazy and do nothing unless consumed"]
pub struct SubscriptionOwnedIter<T: StreamDecoder<T>> {
    subscription: Subscription<T>,
}

impl<T: StreamDecoder<T>> Iterator for SubscriptionOwnedIter<T> {
    type Item = Result<SubscriptionItem<T>, Error>;

    fn next(&mut self) -> Option<Self::Item> {
        self.subscription.next()
    }
}

impl<T: StreamDecoder<T>> IntoIterator for Subscription<T> {
    type Item = Result<SubscriptionItem<T>, Error>;
    type IntoIter = SubscriptionOwnedIter<T>;

    fn into_iter(self) -> Self::IntoIter {
        SubscriptionOwnedIter { subscription: self }
    }
}

/// Non-blocking iterator.
#[allow(private_bounds)]
#[must_use = "iterators are lazy and do nothing unless consumed"]
pub struct SubscriptionTryIter<'a, T: StreamDecoder<T>> {
    subscription: &'a Subscription<T>,
}

impl<T: StreamDecoder<T>> Iterator for SubscriptionTryIter<'_, T> {
    type Item = Result<SubscriptionItem<T>, Error>;

    fn next(&mut self) -> Option<Self::Item> {
        self.subscription.try_next()
    }
}

/// Timeout-bounded iterator.
#[allow(private_bounds)]
#[must_use = "iterators are lazy and do nothing unless consumed"]
pub struct SubscriptionTimeoutIter<'a, T: StreamDecoder<T>> {
    subscription: &'a Subscription<T>,
    timeout: Duration,
}

impl<T: StreamDecoder<T>> Iterator for SubscriptionTimeoutIter<'_, T> {
    type Item = Result<SubscriptionItem<T>, Error>;

    fn next(&mut self) -> Option<Self::Item> {
        self.subscription.next_timeout(self.timeout)
    }
}

/// Marker trait for subscriptions that share a channel based on message type
pub trait SharesChannel {}

#[cfg(all(test, feature = "sync"))]
#[path = "sync_tests.rs"]
mod tests;

1	//! Synchronous subscription implementation
2
3	use std::marker::PhantomData;
4	use std::sync::atomic::{AtomicBool, Ordering};
5	use std::sync::Arc;
6	use std::time::{Duration, Instant};
7
8	use log::{debug, error, warn};
9
10	use super::common::{process_decode_result, DecoderContext, ProcessingResult, SubscriptionItem};
11	use super::StreamDecoder;
12	use crate::errors::Error;
13	use crate::messages::{OutgoingMessages, ResponseMessage};
14	use crate::transport::{InternalSubscription, MessageBus};
15
16	/// A [Subscription] is a stream of responses returned from TWS. A [Subscription] is normally returned when invoking an API that can return more than one value.
17	///
18	/// Each call to [next](Subscription::next), [try_next](Subscription::try_next), or
19	/// [next_timeout](Subscription::next_timeout) returns
20	/// `Option<Result<SubscriptionItem<T>, Error>>`:
21	///
22	/// * `None` — the stream has ended.
23	/// * `Some(Ok(SubscriptionItem::Data(t)))` — a decoded value.
24	/// * `Some(Ok(SubscriptionItem::Notice(n)))` — a non-fatal IB notice; the stream stays open.
25	/// * `Some(Err(e))` — terminal error; subsequent calls return `None`.
26	///
27	/// When you only care about data, use [`iter_data`](Subscription::iter_data) (or
28	/// [`next_data`](Subscription::next_data)) which filters notices for you.
29	#[allow(private_bounds)]
30	pub struct Subscription<T: StreamDecoder<T>> {
31	context: DecoderContext,
32	message_bus: Arc<dyn MessageBus>,
33	request_id: Option<i32>,
34	order_id: Option<i32>,
35	message_type: Option<OutgoingMessages>,
36	phantom: PhantomData<T>,
37	cancelled: AtomicBool,
38	snapshot_ended: AtomicBool,
39	stream_ended: AtomicBool,
40	subscription: InternalSubscription,
41	}
42
43	enum NextAction<T> {
44	Return(Option<T>),
45	Skip,
46	}
47
48	#[allow(private_bounds)]
49	impl<T: StreamDecoder<T>> Subscription<T> {
50	pub(crate) fn new(message_bus: Arc<dyn MessageBus>, subscription: InternalSubscription, context: DecoderContext) -> Self {	87✔
51	let request_id = subscription.request_id;	87✔
52	let order_id = subscription.order_id;	87✔
53	let message_type = subscription.message_type;	87✔
54
55	Subscription {	87✔
56	context,	87✔
57	message_bus,	87✔
58	request_id,	87✔
59	order_id,	87✔
60	message_type,	87✔
61	subscription,	87✔
62	phantom: PhantomData,	87✔
63	cancelled: AtomicBool::new(false),	87✔
64	snapshot_ended: AtomicBool::new(false),	87✔
65	stream_ended: AtomicBool::new(false),	87✔
66	}	87✔
67	}	87✔
68
69	/// Cancel the subscription
70	pub fn cancel(&self) {	89✔
71	// Skip on snapshot subscriptions whose data already arrived.
72	if self.snapshot_ended.load(Ordering::Relaxed) {	89✔
73	return;	×
74	}	89✔
75
76	if self.cancelled.load(Ordering::Relaxed) {	89✔
77	return;	2✔
78	}	87✔
79
80	self.cancelled.store(true, Ordering::Relaxed);	87✔
81
82	if let Some(request_id) = self.request_id {	87✔
83	if let Ok(message) = T::cancel_message(self.context.server_version, self.request_id, Some(&self.context)) {	76✔
84	if let Err(e) = self.message_bus.cancel_subscription(request_id, &message) {	63✔
85	warn!("error cancelling subscription: {e}")	×
86	}	63✔
87	self.subscription.cancel();	63✔
88	}	13✔
89	} else if let Some(order_id) = self.order_id {	11✔
90	if let Ok(message) = T::cancel_message(self.context.server_version, self.request_id, Some(&self.context)) {	×
91	if let Err(e) = self.message_bus.cancel_order_subscription(order_id, &message) {	×
92	warn!("error cancelling order subscription: {e}")	×
93	}	×
94	self.subscription.cancel();	×
95	}	×
96	} else if let Some(message_type) = self.message_type {	11✔
97	if let Ok(message) = T::cancel_message(self.context.server_version, self.request_id, Some(&self.context)) {	9✔
98	if let Err(e) = self.message_bus.cancel_shared_subscription(message_type, &message) {	5✔
99	warn!("error cancelling shared subscription: {e}")	×
100	}	5✔
101	self.subscription.cancel();	5✔
102	}	4✔
103	} else {
104	debug!("Could not determine cancel method")	2✔
105	}
106	}	89✔
107
108	/// Returns the request ID associated with this subscription.
109	pub fn request_id(&self) -> Option<i32> {	1✔
110	self.request_id	1✔
111	}	1✔
112
113	/// Returns the next item, blocking until one is available.
114	///
115	/// # Examples
116	///
117	/// ```no_run
118	/// use ibapi::client::blocking::Client;
119	/// use ibapi::contracts::Contract;
120	/// use ibapi::subscriptions::SubscriptionItem;
121	///
122	/// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
123	/// let contract = Contract::stock("AAPL").build();
124	/// let subscription = client.market_data(&contract)
125	/// .generic_ticks(&["233"])
126	/// .subscribe()
127	/// .expect("market data request failed");
128	///
129	/// while let Some(result) = subscription.next() {
130	/// match result {
131	/// Ok(SubscriptionItem::Data(tick)) => println!("tick: {tick:?}"),
132	/// Ok(SubscriptionItem::Notice(n)) => eprintln!("notice: {n}"),
133	/// Err(e) => { eprintln!("error: {e}"); break; }
134	/// }
135	/// }
136	/// ```
137	pub fn next(&self) -> Option<Result<SubscriptionItem<T>, Error>> {	83✔
138	if self.stream_ended.load(Ordering::Relaxed) {	83✔
139	return None;	3✔
140	}	80✔
141
142	loop {
143	match self.handle_response(self.subscription.next()) {	100✔
144	NextAction::Return(val) => return val,	80✔
145	NextAction::Skip => continue,	20✔
146	}
147	}
148	}	83✔
149
150	fn handle_response(&self, response: Option<Result<ResponseMessage, Error>>) -> NextAction<Result<SubscriptionItem<T>, Error>> {	100✔
151	match response {	1✔
152	Some(Ok(mut message)) => match process_decode_result(T::decode(&self.context, &mut message)) {	96✔
153	ProcessingResult::Success(val) => {	72✔
154	if val.is_snapshot_end() {	72✔
155	self.snapshot_ended.store(true, Ordering::Relaxed);	×
156	}	72✔
157	NextAction::Return(Some(Ok(SubscriptionItem::Data(val))))	72✔
158	}
159	ProcessingResult::Skip => {
160	log::trace!("skipping unexpected message on shared channel");	20✔
161	NextAction::Skip	20✔
162	}
163	ProcessingResult::EndOfStream => {
164	self.stream_ended.store(true, Ordering::Relaxed);	2✔
165	NextAction::Return(None)	2✔
166	}
167	ProcessingResult::Error(err) => {	2✔
168	match &err {	2✔
169	Error::Message(code, msg) => warn!("subscription terminated by TWS error [{code}] {msg}"),	2✔
170	_ => error!("error decoding message: {err}"),	×
171	}
172	self.stream_ended.store(true, Ordering::Relaxed);	2✔
173	NextAction::Return(Some(Err(err)))	2✔
174	}
175	},
176	Some(Err(Error::EndOfStream)) => {
NEW 177	self.stream_ended.store(true, Ordering::Relaxed);	×
UNCOV 178	NextAction::Return(None)	×
179	}
180	Some(Err(e)) => {	1✔
181	self.stream_ended.store(true, Ordering::Relaxed);	1✔
182	NextAction::Return(Some(Err(e)))	1✔
183	}
184	None => NextAction::Return(None),	3✔
185	}
186	}	100✔
187
188	/// Returns the next item without blocking.
189	///
190	/// Returns `None` if no item is available right now; check the surrounding
191	/// loop or stream state to distinguish from end-of-stream.
NEW 192	pub fn try_next(&self) -> Option<Result<SubscriptionItem<T>, Error>> {	×
NEW 193	if self.stream_ended.load(Ordering::Relaxed) {	×
NEW 194	return None;	×
NEW 195	}	×
196	loop {
197	match self.handle_response(self.subscription.try_next()) {	×
198	NextAction::Return(val) => return val,	×
199	NextAction::Skip => continue,	×
200	}
201	}
202	}	×
203
204	/// Returns the next item, blocking up to `timeout`.
NEW 205	pub fn next_timeout(&self, timeout: Duration) -> Option<Result<SubscriptionItem<T>, Error>> {	×
NEW 206	if self.stream_ended.load(Ordering::Relaxed) {	×
NEW 207	return None;	×
NEW 208	}	×
UNCOV 209	let deadline = Instant::now() + timeout;	×
210	loop {
211	let remaining = deadline.saturating_duration_since(Instant::now());	×
212	if remaining.is_zero() {	×
213	return None;	×
214	}	×
215	match self.handle_response(self.subscription.next_timeout(remaining)) {	×
216	NextAction::Return(val) => return val,	×
217	NextAction::Skip => continue,	×
218	}
219	}
220	}	×
221
222	/// Convenience: blocking `next` that filters out notices and yields just data.
223	/// Equivalent to `iter_data().next()`.
224	pub fn next_data(&self) -> Option<Result<T, Error>> {	52✔
225	self.iter_data().next()	52✔
226	}	52✔
227
228	/// Blocking iterator yielding `Result<SubscriptionItem<T>, Error>`. Use
229	/// [`iter_data`](Subscription::iter_data) when you only want data.
230	pub fn iter(&self) -> SubscriptionIter<'_, T> {	63✔
231	SubscriptionIter { subscription: self }	63✔
232	}	63✔
233
234	/// Non-blocking iterator. Returns `None` immediately when nothing is queued.
UNCOV 235	pub fn try_iter(&self) -> SubscriptionTryIter<'_, T> {	×
236	SubscriptionTryIter { subscription: self }	×
237	}	×
238
239	/// Iterator that waits up to `timeout` for each item.
UNCOV 240	pub fn timeout_iter(&self, timeout: Duration) -> SubscriptionTimeoutIter<'_, T> {	×
241	SubscriptionTimeoutIter { subscription: self, timeout }	×
242	}	×
243
244	/// Blocking iterator that filters notices and yields `Result<T, Error>`.
245	/// Notices are logged at `warn!` level.
246	pub fn iter_data(&self) -> FilterData<SubscriptionIter<'_, T>> {	63✔
247	self.iter().filter_data()	63✔
248	}	63✔
249
250	/// Non-blocking data iterator (notices filtered).
NEW 251	pub fn try_iter_data(&self) -> FilterData<SubscriptionTryIter<'_, T>> {	×
NEW 252	self.try_iter().filter_data()	×
NEW 253	}	×
254
255	/// Timeout-bounded data iterator (notices filtered).
NEW 256	pub fn timeout_iter_data(&self, timeout: Duration) -> FilterData<SubscriptionTimeoutIter<'_, T>> {	×
NEW 257	self.timeout_iter(timeout).filter_data()	×
NEW 258	}	×
259	}
260
261	impl<T: StreamDecoder<T>> Drop for Subscription<T> {
262	/// Cancel subscription on drop
263	fn drop(&mut self) {	87✔
264	debug!("dropping subscription");	87✔
265	self.cancel();	87✔
266	}	87✔
267	}
268
269	/// Adapter that filters `SubscriptionItem::Notice` items (logging them at `warn!`)
270	/// from any `Iterator<Item = Result<SubscriptionItem<T>, Error>>` and yields the
271	/// underlying `Result<T, Error>` to the caller.
272	///
273	/// Returned by [`SubscriptionItemIterExt::filter_data`].
274	#[must_use = "iterator adapters are lazy and do nothing unless consumed"]
275	pub struct FilterData<I> {
276	inner: I,
277	}
278
279	impl<I, T> Iterator for FilterData<I>
280	where
281	I: Iterator<Item = Result<SubscriptionItem<T>, Error>>,
282	{
283	type Item = Result<T, Error>;
284
285	fn next(&mut self) -> Option<Self::Item> {	76✔
286	loop {
287	match self.inner.next()? {	76✔
288	Ok(SubscriptionItem::Data(t)) => return Some(Ok(t)),	70✔
NEW 289	Ok(SubscriptionItem::Notice(n)) => {	×
NEW 290	log::warn!("ib notice on subscription: {n}");	×
NEW 291	continue;	×
292	}
293	Err(e) => return Some(Err(e)),	2✔
294	}
295	}
296	}	76✔
297	}
298
299	/// Extension trait that adds [`filter_data`](SubscriptionItemIterExt::filter_data)
300	/// to any iterator yielding `Result<SubscriptionItem<T>, Error>`. Use it to compose
301	/// the data-only flow with iterator combinators that the built-in
302	/// [`iter_data`](Subscription::iter_data) family doesn't already cover, e.g.
303	/// `subscription.iter().take(10).filter_data()`.
304	pub trait SubscriptionItemIterExt: Iterator + Sized {
305	/// Wrap `self` in a [`FilterData`] adapter that drops `SubscriptionItem::Notice`
306	/// items (logging them) and yields the underlying `Result<T, Error>`.
307	fn filter_data<T>(self) -> FilterData<Self>	63✔
308	where	63✔
309	Self: Iterator<Item = Result<SubscriptionItem<T>, Error>>,	63✔
310	{
311	FilterData { inner: self }	63✔
312	}	63✔
313	}
314
315	impl<I: Iterator> SubscriptionItemIterExt for I {}
316
317	/// Blocking iterator over `Result<SubscriptionItem<T>, Error>`.
318	#[allow(private_bounds)]
319	#[must_use = "iterators are lazy and do nothing unless consumed"]
320	pub struct SubscriptionIter<'a, T: StreamDecoder<T>> {
321	subscription: &'a Subscription<T>,
322	}
323
324	impl<T: StreamDecoder<T>> Iterator for SubscriptionIter<'_, T> {
325	type Item = Result<SubscriptionItem<T>, Error>;
326
327	fn next(&mut self) -> Option<Self::Item> {	76✔
328	self.subscription.next()	76✔
329	}	76✔
330	}
331
332	impl<'a, T: StreamDecoder<T>> IntoIterator for &'a Subscription<T> {
333	type Item = Result<SubscriptionItem<T>, Error>;
334	type IntoIter = SubscriptionIter<'a, T>;
335
336	fn into_iter(self) -> Self::IntoIter {	×
337	self.iter()	×
338	}	×
339	}
340
341	/// Owned blocking iterator over `Result<SubscriptionItem<T>, Error>`.
342	#[allow(private_bounds)]
343	#[must_use = "iterators are lazy and do nothing unless consumed"]
344	pub struct SubscriptionOwnedIter<T: StreamDecoder<T>> {
345	subscription: Subscription<T>,
346	}
347
348	impl<T: StreamDecoder<T>> Iterator for SubscriptionOwnedIter<T> {
349	type Item = Result<SubscriptionItem<T>, Error>;
350
351	fn next(&mut self) -> Option<Self::Item> {	2✔
352	self.subscription.next()	2✔
353	}	2✔
354	}
355
356	impl<T: StreamDecoder<T>> IntoIterator for Subscription<T> {
357	type Item = Result<SubscriptionItem<T>, Error>;
358	type IntoIter = SubscriptionOwnedIter<T>;
359
360	fn into_iter(self) -> Self::IntoIter {	1✔
361	SubscriptionOwnedIter { subscription: self }	1✔
362	}	1✔
363	}
364
365	/// Non-blocking iterator.
366	#[allow(private_bounds)]
367	#[must_use = "iterators are lazy and do nothing unless consumed"]
368	pub struct SubscriptionTryIter<'a, T: StreamDecoder<T>> {
369	subscription: &'a Subscription<T>,
370	}
371
372	impl<T: StreamDecoder<T>> Iterator for SubscriptionTryIter<'_, T> {
373	type Item = Result<SubscriptionItem<T>, Error>;
374
375	fn next(&mut self) -> Option<Self::Item> {	×
376	self.subscription.try_next()	×
377	}	×
378	}
379
380	/// Timeout-bounded iterator.
381	#[allow(private_bounds)]
382	#[must_use = "iterators are lazy and do nothing unless consumed"]
383	pub struct SubscriptionTimeoutIter<'a, T: StreamDecoder<T>> {
384	subscription: &'a Subscription<T>,
385	timeout: Duration,
386	}
387
388	impl<T: StreamDecoder<T>> Iterator for SubscriptionTimeoutIter<'_, T> {
389	type Item = Result<SubscriptionItem<T>, Error>;
390
391	fn next(&mut self) -> Option<Self::Item> {	×
392	self.subscription.next_timeout(self.timeout)	×
393	}	×
394	}
395
396	/// Marker trait for subscriptions that share a channel based on message type
397	pub trait SharesChannel {}
398
399	#[cfg(all(test, feature = "sync"))]
400	#[path = "sync_tests.rs"]
401	mod tests;

wboayue / rust-ibapi / 25323281489

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous