25343054113

Committed 04 May 2026 08:55PM UTC coverage: 87.768% (+0.2%) from 87.56%

Build # 25343054113

Build Type

push

github

Committed by

web-flow

Commit Message

feat(transport): route warnings with request_id as Notice items (PR 3) (#506)

* feat(transport): route warnings with request_id as Notice items (PR 3)

Warnings (codes 2100..=2169) bound to a real request_id now flow to their
owning subscription as non-terminal SubscriptionItem::Notice items instead
of being diverted to the global error log. Hard errors with a real
request_id continue to terminate the subscription as Err.

- routing.rs: add ErrorDelivery struct (Routing × Severity), classify_error_delivery
- Notice: widen with advanced_order_reject_json, add notice_from_decoded
- sync/async dispatchers: classify, deliver_to_request_id (request first, order fallback), log_unrouted
- Subscription<T>: handle_response pattern-matches on RoutedItem; surfaces Notice
- delete log_error_payload, transport/common.rs log import; DecodedError::is_log_only

Tests: classify_error_delivery boundary codes; sync+async dispatcher tests for
owned warning, unrouted warning, hard error, order-channel fallback;
sync+async Subscription<T> end-to-end Notice surface tests.

* review cleanups: ERROR_TIME guard, doc comments, notice_from_decoded tests

- ResponseMessage::advanced_order_reject_json() gains a server_version >= ERROR_TIME guard, mirroring error_time(); old-format messages always return empty.
- routing::extract_text_error migrated to use the accessor, single source of truth.
- /// doc comments on next_routed/try_next_routed/next_timeout_routed in transport/mod.rs.
- New direct unit tests for notice_from_decoded covering rich-payload preservation and missing-optionals path.
- Plan doc: pruned superseded "Scope — classification" / "Scope — dispatcher rewrite" sections; as-shipped notes are now the source of truth.

* review cleanups: From<&DecodedError> impls, log_unrouted free function

Composability cleanups from the duplication/SRP review:

- log_unrouted body extracted to transport::common::log_unrouted_notice as a free function. Per-transport log_unrouted met... (continued)

Coverage Stats

102 of 118 new or added lines in 10 files covered. (86.44%)

3 existing lines in 2 files now uncovered.

17077 of 19457 relevant lines covered (87.77%)

26.29 hits per line

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

64.78

/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::{filter_notice, process_decode_result, DecoderContext, ProcessingResult, RoutedItem, SubscriptionItem};
use super::StreamDecoder;
use crate::errors::Error;
use crate::messages::OutgoingMessages;
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_routed()) {
                NextAction::Return(val) => return val,
                NextAction::Skip => continue,
            }
        }
    }

    fn handle_response(&self, response: Option<RoutedItem>) -> NextAction<Result<SubscriptionItem<T>, Error>> {
        match response {
            Some(RoutedItem::Response(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(RoutedItem::Notice(notice)) => NextAction::Return(Some(Ok(SubscriptionItem::Notice(notice)))),
            Some(RoutedItem::Error(Error::EndOfStream)) => {
                self.stream_ended.store(true, Ordering::Relaxed);
                NextAction::Return(None)
            }
            Some(RoutedItem::Error(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_routed()) {
                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_routed(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 {
            if let Some(out) = filter_notice(self.inner.next()?) {
                return Some(out);
            }
        }
    }
}

/// 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::{filter_notice, process_decode_result, DecoderContext, ProcessingResult, RoutedItem, SubscriptionItem};
11	use super::StreamDecoder;
12	use crate::errors::Error;
13	use crate::messages::OutgoingMessages;
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 {	88✔
51	let request_id = subscription.request_id;	88✔
52	let order_id = subscription.order_id;	88✔
53	let message_type = subscription.message_type;	88✔
54
55	Subscription {	88✔
56	context,	88✔
57	message_bus,	88✔
58	request_id,	88✔
59	order_id,	88✔
60	message_type,	88✔
61	subscription,	88✔
62	phantom: PhantomData,	88✔
63	cancelled: AtomicBool::new(false),	88✔
64	snapshot_ended: AtomicBool::new(false),	88✔
65	stream_ended: AtomicBool::new(false),	88✔
66	}	88✔
67	}	88✔
68
69	/// Cancel the subscription
70	pub fn cancel(&self) {	90✔
71	// Skip on snapshot subscriptions whose data already arrived.
72	if self.snapshot_ended.load(Ordering::Relaxed) {	90✔
73	return;	×
74	}	90✔
75
76	if self.cancelled.load(Ordering::Relaxed) {	90✔
77	return;	2✔
78	}	88✔
79
80	self.cancelled.store(true, Ordering::Relaxed);	88✔
81
82	if let Some(request_id) = self.request_id {	88✔
83	if let Ok(message) = T::cancel_message(self.context.server_version, self.request_id, Some(&self.context)) {	77✔
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	}	14✔
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	}	90✔
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>> {	85✔
138	if self.stream_ended.load(Ordering::Relaxed) {	85✔
139	return None;	3✔
140	}	82✔
141
142	loop {
143	match self.handle_response(self.subscription.next_routed()) {	102✔
144	NextAction::Return(val) => return val,	82✔
145	NextAction::Skip => continue,	20✔
146	}
147	}
148	}	85✔
149
150	fn handle_response(&self, response: Option<RoutedItem>) -> NextAction<Result<SubscriptionItem<T>, Error>> {	102✔
151	match response {	1✔
152	Some(RoutedItem::Response(mut message)) => match process_decode_result(T::decode(&self.context, &mut message)) {	97✔
153	ProcessingResult::Success(val) => {	73✔
154	if val.is_snapshot_end() {	73✔
155	self.snapshot_ended.store(true, Ordering::Relaxed);	×
156	}	73✔
157	NextAction::Return(Some(Ok(SubscriptionItem::Data(val))))	73✔
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(RoutedItem::Notice(notice)) => NextAction::Return(Some(Ok(SubscriptionItem::Notice(notice)))),	1✔
177	Some(RoutedItem::Error(Error::EndOfStream)) => {
178	self.stream_ended.store(true, Ordering::Relaxed);	×
179	NextAction::Return(None)	×
180	}
181	Some(RoutedItem::Error(e)) => {	1✔
182	self.stream_ended.store(true, Ordering::Relaxed);	1✔
183	NextAction::Return(Some(Err(e)))	1✔
184	}
185	None => NextAction::Return(None),	3✔
186	}
187	}	102✔
188
189	/// Returns the next item without blocking.
190	///
191	/// Returns `None` if no item is available right now; check the surrounding
192	/// loop or stream state to distinguish from end-of-stream.
193	pub fn try_next(&self) -> Option<Result<SubscriptionItem<T>, Error>> {	×
194	if self.stream_ended.load(Ordering::Relaxed) {	×
195	return None;	×
196	}	×
197	loop {
NEW 198	match self.handle_response(self.subscription.try_next_routed()) {	×
199	NextAction::Return(val) => return val,	×
200	NextAction::Skip => continue,	×
201	}
202	}
203	}	×
204
205	/// Returns the next item, blocking up to `timeout`.
206	pub fn next_timeout(&self, timeout: Duration) -> Option<Result<SubscriptionItem<T>, Error>> {	×
207	if self.stream_ended.load(Ordering::Relaxed) {	×
208	return None;	×
209	}	×
210	let deadline = Instant::now() + timeout;	×
211	loop {
212	let remaining = deadline.saturating_duration_since(Instant::now());	×
213	if remaining.is_zero() {	×
214	return None;	×
215	}	×
NEW 216	match self.handle_response(self.subscription.next_timeout_routed(remaining)) {	×
217	NextAction::Return(val) => return val,	×
218	NextAction::Skip => continue,	×
219	}
220	}
221	}	×
222
223	/// Convenience: blocking `next` that filters out notices and yields just data.
224	/// Equivalent to `iter_data().next()`.
225	pub fn next_data(&self) -> Option<Result<T, Error>> {	52✔
226	self.iter_data().next()	52✔
227	}	52✔
228
229	/// Blocking iterator yielding `Result<SubscriptionItem<T>, Error>`. Use
230	/// [`iter_data`](Subscription::iter_data) when you only want data.
231	pub fn iter(&self) -> SubscriptionIter<'_, T> {	63✔
232	SubscriptionIter { subscription: self }	63✔
233	}	63✔
234
235	/// Non-blocking iterator. Returns `None` immediately when nothing is queued.
236	pub fn try_iter(&self) -> SubscriptionTryIter<'_, T> {	×
237	SubscriptionTryIter { subscription: self }	×
238	}	×
239
240	/// Iterator that waits up to `timeout` for each item.
241	pub fn timeout_iter(&self, timeout: Duration) -> SubscriptionTimeoutIter<'_, T> {	×
242	SubscriptionTimeoutIter { subscription: self, timeout }	×
243	}	×
244
245	/// Blocking iterator that filters notices and yields `Result<T, Error>`.
246	/// Notices are logged at `warn!` level.
247	pub fn iter_data(&self) -> FilterData<SubscriptionIter<'_, T>> {	63✔
248	self.iter().filter_data()	63✔
249	}	63✔
250
251	/// Non-blocking data iterator (notices filtered).
252	pub fn try_iter_data(&self) -> FilterData<SubscriptionTryIter<'_, T>> {	×
253	self.try_iter().filter_data()	×
254	}	×
255
256	/// Timeout-bounded data iterator (notices filtered).
257	pub fn timeout_iter_data(&self, timeout: Duration) -> FilterData<SubscriptionTimeoutIter<'_, T>> {	×
258	self.timeout_iter(timeout).filter_data()	×
259	}	×
260	}
261
262	impl<T: StreamDecoder<T>> Drop for Subscription<T> {
263	/// Cancel subscription on drop
264	fn drop(&mut self) {	88✔
265	debug!("dropping subscription");	88✔
266	self.cancel();	88✔
267	}	88✔
268	}
269
270	/// Adapter that filters `SubscriptionItem::Notice` items (logging them at `warn!`)
271	/// from any `Iterator<Item = Result<SubscriptionItem<T>, Error>>` and yields the
272	/// underlying `Result<T, Error>` to the caller.
273	///
274	/// Returned by [`SubscriptionItemIterExt::filter_data`].
275	#[must_use = "iterator adapters are lazy and do nothing unless consumed"]
276	pub struct FilterData<I> {
277	inner: I,
278	}
279
280	impl<I, T> Iterator for FilterData<I>
281	where
282	I: Iterator<Item = Result<SubscriptionItem<T>, Error>>,
283	{
284	type Item = Result<T, Error>;
285
286	fn next(&mut self) -> Option<Self::Item> {	76✔
287	loop {
288	if let Some(out) = filter_notice(self.inner.next()?) {	76✔
289	return Some(out);	72✔
290	}	×
291	}
292	}	76✔
293	}
294
295	/// Extension trait that adds [`filter_data`](SubscriptionItemIterExt::filter_data)
296	/// to any iterator yielding `Result<SubscriptionItem<T>, Error>`. Use it to compose
297	/// the data-only flow with iterator combinators that the built-in
298	/// [`iter_data`](Subscription::iter_data) family doesn't already cover, e.g.
299	/// `subscription.iter().take(10).filter_data()`.
300	pub trait SubscriptionItemIterExt: Iterator + Sized {
301	/// Wrap `self` in a [`FilterData`] adapter that drops `SubscriptionItem::Notice`
302	/// items (logging them) and yields the underlying `Result<T, Error>`.
303	fn filter_data<T>(self) -> FilterData<Self>	63✔
304	where	63✔
305	Self: Iterator<Item = Result<SubscriptionItem<T>, Error>>,	63✔
306	{
307	FilterData { inner: self }	63✔
308	}	63✔
309	}
310
311	impl<I: Iterator> SubscriptionItemIterExt for I {}
312
313	/// Blocking iterator over `Result<SubscriptionItem<T>, Error>`.
314	#[allow(private_bounds)]
315	#[must_use = "iterators are lazy and do nothing unless consumed"]
316	pub struct SubscriptionIter<'a, T: StreamDecoder<T>> {
317	subscription: &'a Subscription<T>,
318	}
319
320	impl<T: StreamDecoder<T>> Iterator for SubscriptionIter<'_, T> {
321	type Item = Result<SubscriptionItem<T>, Error>;
322
323	fn next(&mut self) -> Option<Self::Item> {	76✔
324	self.subscription.next()	76✔
325	}	76✔
326	}
327
328	impl<'a, T: StreamDecoder<T>> IntoIterator for &'a Subscription<T> {
329	type Item = Result<SubscriptionItem<T>, Error>;
330	type IntoIter = SubscriptionIter<'a, T>;
331
332	fn into_iter(self) -> Self::IntoIter {	×
333	self.iter()	×
334	}	×
335	}
336
337	/// Owned blocking iterator over `Result<SubscriptionItem<T>, Error>`.
338	#[allow(private_bounds)]
339	#[must_use = "iterators are lazy and do nothing unless consumed"]
340	pub struct SubscriptionOwnedIter<T: StreamDecoder<T>> {
341	subscription: Subscription<T>,
342	}
343
344	impl<T: StreamDecoder<T>> Iterator for SubscriptionOwnedIter<T> {
345	type Item = Result<SubscriptionItem<T>, Error>;
346
347	fn next(&mut self) -> Option<Self::Item> {	2✔
348	self.subscription.next()	2✔
349	}	2✔
350	}
351
352	impl<T: StreamDecoder<T>> IntoIterator for Subscription<T> {
353	type Item = Result<SubscriptionItem<T>, Error>;
354	type IntoIter = SubscriptionOwnedIter<T>;
355
356	fn into_iter(self) -> Self::IntoIter {	1✔
357	SubscriptionOwnedIter { subscription: self }	1✔
358	}	1✔
359	}
360
361	/// Non-blocking iterator.
362	#[allow(private_bounds)]
363	#[must_use = "iterators are lazy and do nothing unless consumed"]
364	pub struct SubscriptionTryIter<'a, T: StreamDecoder<T>> {
365	subscription: &'a Subscription<T>,
366	}
367
368	impl<T: StreamDecoder<T>> Iterator for SubscriptionTryIter<'_, T> {
369	type Item = Result<SubscriptionItem<T>, Error>;
370
371	fn next(&mut self) -> Option<Self::Item> {	×
372	self.subscription.try_next()	×
373	}	×
374	}
375
376	/// Timeout-bounded iterator.
377	#[allow(private_bounds)]
378	#[must_use = "iterators are lazy and do nothing unless consumed"]
379	pub struct SubscriptionTimeoutIter<'a, T: StreamDecoder<T>> {
380	subscription: &'a Subscription<T>,
381	timeout: Duration,
382	}
383
384	impl<T: StreamDecoder<T>> Iterator for SubscriptionTimeoutIter<'_, T> {
385	type Item = Result<SubscriptionItem<T>, Error>;
386
387	fn next(&mut self) -> Option<Self::Item> {	×
388	self.subscription.next_timeout(self.timeout)	×
389	}	×
390	}
391
392	/// Marker trait for subscriptions that share a channel based on message type
393	pub trait SharesChannel {}
394
395	#[cfg(all(test, feature = "sync"))]
396	#[path = "sync_tests.rs"]
397	mod tests;

wboayue / rust-ibapi / 25343054113

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