25267123189

Committed 03 May 2026 01:53AM UTC coverage: 88.611% (+0.02%) from 88.591%

Build # 25267123189

Build Type

push

github

Committed by

web-flow

Commit Message

fix: surface TWS errors on subscription channels (closes #434) (#490)

* fix: surface TWS errors on subscription channels (closes #434)

Audit StreamDecoders for IncomingMessages::Error handling. When TWS
returns an error tied to a subscription's request_id, decoders that
didn't match on the message type either reproduced the original
"invalid digit found in string" parse failure (PnL, PnLSingle,
WshEventData, ScannerData), terminated with Error::Simple
("unexpected message: Error") losing the IB code/text (option
computations, account summary/position/update variants), or silently
skipped via UnexpectedResponse (option chain, news, wsh metadata,
display groups). 15 decoders now return Err(Error::Message(code, msg))
for type-4 messages.

Also: regression test for realtime_bars Error handling (sync + async)
locks down the prior fix (e1333df3). Sync Subscription log line
distinguishes TWS errors (warn!) from real decode failures (error!).

* test: extract assert_tws_error_message helper

Replace 15 copies of the Error::Message match-and-panic pattern with a
shared helper in common::test_utils. Tightens the new decoder tests
without changing semantics.

Coverage Stats

95 of 105 new or added lines in 8 files covered. (90.48%)

5 existing lines in 3 files now uncovered.

17124 of 19325 relevant lines covered (88.61%)

24.67 hits per line

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

73.74

/src/subscriptions/sync.rs

//! Synchronous subscription implementation

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

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

use super::common::{process_decode_result, should_store_error, DecoderContext, ProcessingResult};
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.
///
/// You can convert subscriptions into blocking or non-blocking iterators using the [iter](Subscription::iter), [try_iter](Subscription::try_iter) or [timeout_iter](Subscription::timeout_iter) methods.
///
/// Alternatively, you may poll subscriptions in a blocking or non-blocking manner using the [next](Subscription::next), [try_next](Subscription::try_next) or [next_timeout](Subscription::next_timeout) methods.
#[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,
    error: Mutex<Option<Error>>,
}

/// Whether a response should be returned or skipped
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),
            error: Mutex::new(None),
        }
    }

    /// Cancel the subscription
    pub fn cancel(&self) {
        // Only cancel if snapshot hasn't ended (for market data snapshots)
        // For streaming subscriptions, snapshot_ended will remain false
        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 available value, blocking if necessary until a value becomes available.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ibapi::client::blocking::Client;
    /// use ibapi::contracts::Contract;
    ///
    /// 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");
    ///
    /// // Process data blocking until the next value is available
    /// while let Some(data) = subscription.next() {
    ///     println!("Received data: {data:?}");
    /// }
    ///
    /// // When the loop exits, check if it was due to an error
    /// if let Some(err) = subscription.error() {
    ///     eprintln!("subscription error: {err}");
    /// }
    /// ```
    /// # Returns
    /// * `Some(T)` - The next available item from the subscription
    /// * `None` - If the subscription has ended or encountered an error
    pub fn next(&self) -> Option<T> {
        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,
            }
        }
    }

    /// Returns the current error state of the subscription.
    ///
    /// This method allows checking if an error occurred during subscription processing.
    /// Errors are stored internally when they occur during `next()`, `try_next()`, or `next_timeout()` calls.
    ///
    /// # Returns
    /// * `Some(Error)` - If an error has occurred
    /// * `None` - If no error has occurred
    pub fn error(&self) -> Option<Error> {
        let mut error = self.error.lock().unwrap();
        error.take()
    }

    fn clear_error(&self) {
        let mut error = self.error.lock().unwrap();
        *error = None;
    }

    fn handle_response(&self, response: Option<Result<ResponseMessage, Error>>) -> NextAction<T> {
        self.clear_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(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}"),
                    }
                    let mut error = self.error.lock().unwrap();
                    *error = Some(err);
                    NextAction::Return(None)
                }
            },
            Some(Err(e)) => {
                if should_store_error(&e) {
                    let mut error = self.error.lock().unwrap();
                    *error = Some(e);
                }
                NextAction::Return(None)
            }
            None => NextAction::Return(None),
        }
    }

    /// Tries to return the next available value without blocking.
    ///
    /// Returns immediately with:
    /// - `Some(value)` if a value is available
    /// - `None` if no data is currently available
    ///
    /// Use this method when you want to poll for data without blocking.
    /// Check `error()` to determine if `None` was returned due to an error.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ibapi::client::blocking::Client;
    /// use ibapi::contracts::Contract;
    /// use std::thread;
    /// use std::time::Duration;
    ///
    /// 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");
    ///
    /// // Poll for data without blocking
    /// loop {
    ///     if let Some(data) = subscription.try_next() {
    ///         println!("{data:?}");
    ///     } else if let Some(err) = subscription.error() {
    ///         eprintln!("Error: {err}");
    ///         break;
    ///     } else {
    ///         // No data available, do other work or sleep
    ///         thread::sleep(Duration::from_millis(100));
    ///     }
    /// }
    /// ```
    pub fn try_next(&self) -> Option<T> {
        loop {
            match self.handle_response(self.subscription.try_next()) {
                NextAction::Return(val) => return val,
                NextAction::Skip => continue,
            }
        }
    }

    /// Waits for the next available value up to the specified timeout duration.
    ///
    /// Returns:
    /// - `Some(value)` if a value becomes available within the timeout
    /// - `None` if the timeout expires before data becomes available
    ///
    /// Check `error()` to determine if `None` was returned due to an error.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ibapi::client::blocking::Client;
    /// use ibapi::contracts::Contract;
    /// use std::time::Duration;
    ///
    /// 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");
    ///
    /// // Wait up to 5 seconds for data
    /// if let Some(data) = subscription.next_timeout(Duration::from_secs(5)) {
    ///     println!("{data:?}");
    /// } else if let Some(err) = subscription.error() {
    ///     eprintln!("Error: {err}");
    /// } else {
    ///     eprintln!("Timeout: no data received within 5 seconds");
    /// }
    /// ```
    pub fn next_timeout(&self, timeout: Duration) -> Option<T> {
        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,
            }
        }
    }

    /// Creates a blocking iterator over the subscription data.
    ///
    /// The iterator will block waiting for the next value if none is immediately available.
    /// The iterator ends when the subscription is cancelled or an unrecoverable error occurs.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ibapi::client::blocking::Client;
    ///
    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
    ///
    /// let subscription = client.positions().expect("positions request failed");
    ///
    /// // Process all positions as they arrive
    /// for position in subscription.iter() {
    ///     println!("{position:?}");
    /// }
    ///
    /// // Check if iteration ended due to an error
    /// if let Some(err) = subscription.error() {
    ///     eprintln!("Subscription error: {err}");
    /// }
    /// ```
    pub fn iter(&self) -> SubscriptionIter<'_, T> {
        SubscriptionIter { subscription: self }
    }

    /// Creates a non-blocking iterator over the subscription data.
    ///
    /// The iterator will return immediately with `None` if no data is available.
    /// Use this when you want to process available data without blocking.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ibapi::client::blocking::Client;
    /// use std::thread;
    /// use std::time::Duration;
    ///
    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
    ///
    /// let subscription = client.positions().expect("positions request failed");
    ///
    /// // Process available positions without blocking
    /// loop {
    ///     let mut data_received = false;
    ///     for position in subscription.try_iter() {
    ///         data_received = true;
    ///         println!("{position:?}");
    ///     }
    ///     
    ///     if let Some(err) = subscription.error() {
    ///         eprintln!("Error: {err}");
    ///         break;
    ///     }
    ///     
    ///     if !data_received {
    ///         // No data available, do other work or sleep
    ///         thread::sleep(Duration::from_millis(100));
    ///     }
    /// }
    /// ```
    pub fn try_iter(&self) -> SubscriptionTryIter<'_, T> {
        SubscriptionTryIter { subscription: self }
    }

    /// Creates an iterator that waits up to the specified timeout for each value.
    ///
    /// The iterator will wait up to `timeout` duration for each value.
    /// If the timeout expires, the iterator ends.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use ibapi::client::blocking::Client;
    /// use std::time::Duration;
    ///
    /// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
    ///
    /// let subscription = client.positions().expect("positions request failed");
    ///
    /// // Process positions with a 5 second timeout per item
    /// for position in subscription.timeout_iter(Duration::from_secs(5)) {
    ///     println!("{position:?}");
    /// }
    ///
    /// if let Some(err) = subscription.error() {
    ///     eprintln!("Error: {err}");
    /// } else {
    ///     println!("No more positions received within timeout");
    /// }
    /// ```
    pub fn timeout_iter(&self, timeout: Duration) -> SubscriptionTimeoutIter<'_, T> {
        SubscriptionTimeoutIter { subscription: self, timeout }
    }
}

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

/// An iterator that yields items as they become available, blocking if necessary.
#[allow(private_bounds)]
pub struct SubscriptionIter<'a, T: StreamDecoder<T>> {
    subscription: &'a Subscription<T>,
}

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

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

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

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

/// An iterator that takes ownership and yields items as they become available, blocking if necessary.
#[allow(private_bounds)]
pub struct SubscriptionOwnedIter<T: StreamDecoder<T>> {
    subscription: Subscription<T>,
}

impl<T: StreamDecoder<T>> Iterator for SubscriptionOwnedIter<T> {
    type Item = T;

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

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

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

/// An iterator that yields items as they become available without blocking.
#[allow(private_bounds)]
pub struct SubscriptionTryIter<'a, T: StreamDecoder<T>> {
    subscription: &'a Subscription<T>,
}

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

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

/// An iterator that yields items with a timeout.
#[allow(private_bounds)]
pub struct SubscriptionTimeoutIter<'a, T: StreamDecoder<T>> {
    subscription: &'a Subscription<T>,
    timeout: Duration,
}

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

    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"))]
mod tests {
    use super::*;
    use crate::messages::{encode_protobuf_message, OutgoingMessages, ResponseMessage};
    use crate::stubs::MessageBusStub;
    use std::sync::Arc;

    #[derive(Debug)]
    struct EndOfStreamItem;

    impl StreamDecoder<EndOfStreamItem> for EndOfStreamItem {
        fn decode(_context: &DecoderContext, _msg: &mut ResponseMessage) -> Result<EndOfStreamItem, Error> {
            Err(Error::EndOfStream)
        }

        fn cancel_message(_server_version: i32, _id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {
            Ok(encode_protobuf_message(OutgoingMessages::CancelMarketData as i32, &[]))
        }
    }

    #[test]
    fn test_subscription_skips_unexpected_messages_without_limit() {
        use std::sync::atomic::AtomicUsize;

        static CALL_COUNT: AtomicUsize = AtomicUsize::new(0);

        #[derive(Debug)]
        struct SkipThenSuccess;

        impl StreamDecoder<SkipThenSuccess> for SkipThenSuccess {
            fn decode(_context: &DecoderContext, _msg: &mut ResponseMessage) -> Result<SkipThenSuccess, Error> {
                let n = CALL_COUNT.fetch_add(1, Ordering::Relaxed);
                if n < 20 {
                    Err(Error::UnexpectedResponse(ResponseMessage::from("stray\0")))
                } else {
                    Ok(SkipThenSuccess)
                }
            }
        }

        CALL_COUNT.store(0, Ordering::Relaxed);

        // 20 stray messages + 1 valid (more than the old MAX_DECODE_RETRIES=10)
        let mut responses: Vec<String> = (0..21).map(|_| "1|msg".to_string()).collect();
        // Sentinel to avoid blocking on the channel after success
        responses.push("1|done".to_string());

        let stub = MessageBusStub::with_responses(responses);
        let message_bus = Arc::new(stub);

        let sub: Subscription<SkipThenSuccess> = {
            let internal = message_bus.send_request(1, &[]).unwrap();
            Subscription::new(message_bus.clone(), internal, DecoderContext::default())
        };

        let result = sub.next();
        assert!(result.is_some(), "subscription should survive 20 skips and return valid message");
        assert_eq!(CALL_COUNT.load(Ordering::Relaxed), 21);
    }

    #[test]
    fn test_no_retries_after_end_of_stream() {
        let stub = MessageBusStub::with_responses(vec![
            "1|data".to_string(),  // triggers EndOfStream via decoder
            "1|stray".to_string(), // stray message after stream ended
        ]);
        let message_bus = Arc::new(stub);

        let sub: Subscription<EndOfStreamItem> = {
            let internal = message_bus.send_request(1, &[]).unwrap();
            Subscription::new(message_bus.clone(), internal, DecoderContext::default())
        };

        // First call hits EndOfStream, returns None
        assert!(sub.next().is_none());

        // Second call should return None immediately (stream_ended guard)
        assert!(sub.next().is_none());
        assert!(sub.stream_ended.load(Ordering::Relaxed));
    }
}

1	//! Synchronous subscription implementation
2
3	use std::marker::PhantomData;
4	use std::sync::atomic::{AtomicBool, Ordering};
5	use std::sync::{Arc, Mutex};
6	use std::time::{Duration, Instant};
7
8	use log::{debug, error, warn};
9
10	use super::common::{process_decode_result, should_store_error, DecoderContext, ProcessingResult};
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	/// You can convert subscriptions into blocking or non-blocking iterators using the [iter](Subscription::iter), [try_iter](Subscription::try_iter) or [timeout_iter](Subscription::timeout_iter) methods.
19	///
20	/// Alternatively, you may poll subscriptions in a blocking or non-blocking manner using the [next](Subscription::next), [try_next](Subscription::try_next) or [next_timeout](Subscription::next_timeout) methods.
21	#[allow(private_bounds)]
22	pub struct Subscription<T: StreamDecoder<T>> {
23	context: DecoderContext,
24	message_bus: Arc<dyn MessageBus>,
25	request_id: Option<i32>,
26	order_id: Option<i32>,
27	message_type: Option<OutgoingMessages>,
28	phantom: PhantomData<T>,
29	cancelled: AtomicBool,
30	snapshot_ended: AtomicBool,
31	stream_ended: AtomicBool,
32	subscription: InternalSubscription,
33	error: Mutex<Option<Error>>,
34	}
35
36	/// Whether a response should be returned or skipped
37	enum NextAction<T> {
38	Return(Option<T>),
39	Skip,
40	}
41
42	#[allow(private_bounds)]
43	impl<T: StreamDecoder<T>> Subscription<T> {
44	pub(crate) fn new(message_bus: Arc<dyn MessageBus>, subscription: InternalSubscription, context: DecoderContext) -> Self {	86✔
45	let request_id = subscription.request_id;	86✔
46	let order_id = subscription.order_id;	86✔
47	let message_type = subscription.message_type;	86✔
48
49	Subscription {	86✔
50	context,	86✔
51	message_bus,	86✔
52	request_id,	86✔
53	order_id,	86✔
54	message_type,	86✔
55	subscription,	86✔
56	phantom: PhantomData,	86✔
57	cancelled: AtomicBool::new(false),	86✔
58	snapshot_ended: AtomicBool::new(false),	86✔
59	stream_ended: AtomicBool::new(false),	86✔
60	error: Mutex::new(None),	86✔
61	}	86✔
62	}	86✔
63
64	/// Cancel the subscription
65	pub fn cancel(&self) {	88✔
66	// Only cancel if snapshot hasn't ended (for market data snapshots)
67	// For streaming subscriptions, snapshot_ended will remain false
68	if self.snapshot_ended.load(Ordering::Relaxed) {	88✔
69	return;	×
70	}	88✔
71
72	if self.cancelled.load(Ordering::Relaxed) {	88✔
73	return;	2✔
74	}	86✔
75
76	self.cancelled.store(true, Ordering::Relaxed);	86✔
77
78	if let Some(request_id) = self.request_id {	86✔
79	if let Ok(message) = T::cancel_message(self.context.server_version, self.request_id, Some(&self.context)) {	75✔
80	if let Err(e) = self.message_bus.cancel_subscription(request_id, &message) {	63✔
81	warn!("error cancelling subscription: {e}")	×
82	}	63✔
83	self.subscription.cancel();	63✔
84	}	12✔
85	} else if let Some(order_id) = self.order_id {	11✔
86	if let Ok(message) = T::cancel_message(self.context.server_version, self.request_id, Some(&self.context)) {	×
87	if let Err(e) = self.message_bus.cancel_order_subscription(order_id, &message) {	×
88	warn!("error cancelling order subscription: {e}")	×
89	}	×
90	self.subscription.cancel();	×
91	}	×
92	} else if let Some(message_type) = self.message_type {	11✔
93	if let Ok(message) = T::cancel_message(self.context.server_version, self.request_id, Some(&self.context)) {	9✔
94	if let Err(e) = self.message_bus.cancel_shared_subscription(message_type, &message) {	5✔
95	warn!("error cancelling shared subscription: {e}")	×
96	}	5✔
97	self.subscription.cancel();	5✔
98	}	4✔
99	} else {
100	debug!("Could not determine cancel method")	2✔
101	}
102	}	88✔
103
104	/// Returns the request ID associated with this subscription.
105	pub fn request_id(&self) -> Option<i32> {	1✔
106	self.request_id	1✔
107	}	1✔
108
109	/// Returns the next available value, blocking if necessary until a value becomes available.
110	///
111	/// # Examples
112	///
113	/// ```no_run
114	/// use ibapi::client::blocking::Client;
115	/// use ibapi::contracts::Contract;
116	///
117	/// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
118	///
119	/// let contract = Contract::stock("AAPL").build();
120	/// let subscription = client.market_data(&contract)
121	/// .generic_ticks(&["233"])
122	/// .subscribe()
123	/// .expect("market data request failed");
124	///
125	/// // Process data blocking until the next value is available
126	/// while let Some(data) = subscription.next() {
127	/// println!("Received data: {data:?}");
128	/// }
129	///
130	/// // When the loop exits, check if it was due to an error
131	/// if let Some(err) = subscription.error() {
132	/// eprintln!("subscription error: {err}");
133	/// }
134	/// ```
135	/// # Returns
136	/// * `Some(T)` - The next available item from the subscription
137	/// * `None` - If the subscription has ended or encountered an error
138	pub fn next(&self) -> Option<T> {	80✔
139	if self.stream_ended.load(Ordering::Relaxed) {	80✔
140	return None;	1✔
141	}	79✔
142
143	loop {
144	match self.handle_response(self.subscription.next()) {	99✔
145	NextAction::Return(val) => return val,	79✔
146	NextAction::Skip => continue,	20✔
147	}
148	}
149	}	80✔
150
151	/// Returns the current error state of the subscription.
152	///
153	/// This method allows checking if an error occurred during subscription processing.
154	/// Errors are stored internally when they occur during `next()`, `try_next()`, or `next_timeout()` calls.
155	///
156	/// # Returns
157	/// * `Some(Error)` - If an error has occurred
158	/// * `None` - If no error has occurred
159	pub fn error(&self) -> Option<Error> {	3✔
160	let mut error = self.error.lock().unwrap();	3✔
161	error.take()	3✔
162	}	3✔
163
164	fn clear_error(&self) {	99✔
165	let mut error = self.error.lock().unwrap();	99✔
166	*error = None;	99✔
167	}	99✔
168
169	fn handle_response(&self, response: Option<Result<ResponseMessage, Error>>) -> NextAction<T> {	99✔
170	self.clear_error();	99✔
171
172	match response {	96✔
173	Some(Ok(mut message)) => match process_decode_result(T::decode(&self.context, &mut message)) {	96✔
174	ProcessingResult::Success(val) => {	72✔
175	if val.is_snapshot_end() {	72✔
176	self.snapshot_ended.store(true, Ordering::Relaxed);	×
177	}	72✔
178	NextAction::Return(Some(val))	72✔
179	}
180	ProcessingResult::Skip => {
181	log::trace!("skipping unexpected message on shared channel");	20✔
182	NextAction::Skip	20✔
183	}
184	ProcessingResult::EndOfStream => {
185	self.stream_ended.store(true, Ordering::Relaxed);	2✔
186	NextAction::Return(None)	2✔
187	}
188	ProcessingResult::Error(err) => {	2✔
189	match &err {	2✔
190	Error::Message(code, msg) => warn!("subscription terminated by TWS error [{code}] {msg}"),	2✔
NEW 191	_ => error!("error decoding message: {err}"),	×
192	}
193	let mut error = self.error.lock().unwrap();	2✔
194	*error = Some(err);	2✔
195	NextAction::Return(None)	2✔
196	}
197	},
198	Some(Err(e)) => {	×
199	if should_store_error(&e) {	×
200	let mut error = self.error.lock().unwrap();	×
201	*error = Some(e);	×
202	}	×
203	NextAction::Return(None)	×
204	}
205	None => NextAction::Return(None),	3✔
206	}
207	}	99✔
208
209	/// Tries to return the next available value without blocking.
210	///
211	/// Returns immediately with:
212	/// - `Some(value)` if a value is available
213	/// - `None` if no data is currently available
214	///
215	/// Use this method when you want to poll for data without blocking.
216	/// Check `error()` to determine if `None` was returned due to an error.
217	///
218	/// # Examples
219	///
220	/// ```no_run
221	/// use ibapi::client::blocking::Client;
222	/// use ibapi::contracts::Contract;
223	/// use std::thread;
224	/// use std::time::Duration;
225	///
226	/// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
227	///
228	/// let contract = Contract::stock("AAPL").build();
229	/// let subscription = client.market_data(&contract)
230	/// .generic_ticks(&["233"])
231	/// .subscribe()
232	/// .expect("market data request failed");
233	///
234	/// // Poll for data without blocking
235	/// loop {
236	/// if let Some(data) = subscription.try_next() {
237	/// println!("{data:?}");
238	/// } else if let Some(err) = subscription.error() {
239	/// eprintln!("Error: {err}");
240	/// break;
241	/// } else {
242	/// // No data available, do other work or sleep
243	/// thread::sleep(Duration::from_millis(100));
244	/// }
245	/// }
246	/// ```
247	pub fn try_next(&self) -> Option<T> {	×
248	loop {
249	match self.handle_response(self.subscription.try_next()) {	×
250	NextAction::Return(val) => return val,	×
251	NextAction::Skip => continue,	×
252	}
253	}
254	}	×
255
256	/// Waits for the next available value up to the specified timeout duration.
257	///
258	/// Returns:
259	/// - `Some(value)` if a value becomes available within the timeout
260	/// - `None` if the timeout expires before data becomes available
261	///
262	/// Check `error()` to determine if `None` was returned due to an error.
263	///
264	/// # Examples
265	///
266	/// ```no_run
267	/// use ibapi::client::blocking::Client;
268	/// use ibapi::contracts::Contract;
269	/// use std::time::Duration;
270	///
271	/// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
272	///
273	/// let contract = Contract::stock("AAPL").build();
274	/// let subscription = client.market_data(&contract)
275	/// .generic_ticks(&["233"])
276	/// .subscribe()
277	/// .expect("market data request failed");
278	///
279	/// // Wait up to 5 seconds for data
280	/// if let Some(data) = subscription.next_timeout(Duration::from_secs(5)) {
281	/// println!("{data:?}");
282	/// } else if let Some(err) = subscription.error() {
283	/// eprintln!("Error: {err}");
284	/// } else {
285	/// eprintln!("Timeout: no data received within 5 seconds");
286	/// }
287	/// ```
288	pub fn next_timeout(&self, timeout: Duration) -> Option<T> {	×
289	let deadline = Instant::now() + timeout;	×
290	loop {
291	let remaining = deadline.saturating_duration_since(Instant::now());	×
292	if remaining.is_zero() {	×
293	return None;	×
294	}	×
295	match self.handle_response(self.subscription.next_timeout(remaining)) {	×
296	NextAction::Return(val) => return val,	×
297	NextAction::Skip => continue,	×
298	}
299	}
300	}	×
301
302	/// Creates a blocking iterator over the subscription data.
303	///
304	/// The iterator will block waiting for the next value if none is immediately available.
305	/// The iterator ends when the subscription is cancelled or an unrecoverable error occurs.
306	///
307	/// # Examples
308	///
309	/// ```no_run
310	/// use ibapi::client::blocking::Client;
311	///
312	/// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
313	///
314	/// let subscription = client.positions().expect("positions request failed");
315	///
316	/// // Process all positions as they arrive
317	/// for position in subscription.iter() {
318	/// println!("{position:?}");
319	/// }
320	///
321	/// // Check if iteration ended due to an error
322	/// if let Some(err) = subscription.error() {
323	/// eprintln!("Subscription error: {err}");
324	/// }
325	/// ```
326	pub fn iter(&self) -> SubscriptionIter<'_, T> {	10✔
327	SubscriptionIter { subscription: self }	10✔
328	}	10✔
329
330	/// Creates a non-blocking iterator over the subscription data.
331	///
332	/// The iterator will return immediately with `None` if no data is available.
333	/// Use this when you want to process available data without blocking.
334	///
335	/// # Examples
336	///
337	/// ```no_run
338	/// use ibapi::client::blocking::Client;
339	/// use std::thread;
340	/// use std::time::Duration;
341	///
342	/// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
343	///
344	/// let subscription = client.positions().expect("positions request failed");
345	///
346	/// // Process available positions without blocking
347	/// loop {
348	/// let mut data_received = false;
349	/// for position in subscription.try_iter() {
350	/// data_received = true;
351	/// println!("{position:?}");
352	/// }
353	///
354	/// if let Some(err) = subscription.error() {
355	/// eprintln!("Error: {err}");
356	/// break;
357	/// }
358	///
359	/// if !data_received {
360	/// // No data available, do other work or sleep
361	/// thread::sleep(Duration::from_millis(100));
362	/// }
363	/// }
364	/// ```
365	pub fn try_iter(&self) -> SubscriptionTryIter<'_, T> {	×
366	SubscriptionTryIter { subscription: self }	×
367	}	×
368
369	/// Creates an iterator that waits up to the specified timeout for each value.
370	///
371	/// The iterator will wait up to `timeout` duration for each value.
372	/// If the timeout expires, the iterator ends.
373	///
374	/// # Examples
375	///
376	/// ```no_run
377	/// use ibapi::client::blocking::Client;
378	/// use std::time::Duration;
379	///
380	/// let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
381	///
382	/// let subscription = client.positions().expect("positions request failed");
383	///
384	/// // Process positions with a 5 second timeout per item
385	/// for position in subscription.timeout_iter(Duration::from_secs(5)) {
386	/// println!("{position:?}");
387	/// }
388	///
389	/// if let Some(err) = subscription.error() {
390	/// eprintln!("Error: {err}");
391	/// } else {
392	/// println!("No more positions received within timeout");
393	/// }
394	/// ```
395	pub fn timeout_iter(&self, timeout: Duration) -> SubscriptionTimeoutIter<'_, T> {	×
396	SubscriptionTimeoutIter { subscription: self, timeout }	×
397	}	×
398	}
399
400	impl<T: StreamDecoder<T>> Drop for Subscription<T> {
401	/// Cancel subscription on drop
402	fn drop(&mut self) {	86✔
403	debug!("dropping subscription");	86✔
404	self.cancel();	86✔
405	}	86✔
406	}
407
408	/// An iterator that yields items as they become available, blocking if necessary.
409	#[allow(private_bounds)]
410	pub struct SubscriptionIter<'a, T: StreamDecoder<T>> {
411	subscription: &'a Subscription<T>,
412	}
413
414	impl<T: StreamDecoder<T>> Iterator for SubscriptionIter<'_, T> {
415	type Item = T;
416
417	fn next(&mut self) -> Option<Self::Item> {	19✔
418	self.subscription.next()	19✔
419	}	19✔
420	}
421
422	impl<'a, T: StreamDecoder<T>> IntoIterator for &'a Subscription<T> {
423	type Item = T;
424	type IntoIter = SubscriptionIter<'a, T>;
425
426	fn into_iter(self) -> Self::IntoIter {	×
427	self.iter()	×
428	}	×
429	}
430
431	/// An iterator that takes ownership and yields items as they become available, blocking if necessary.
432	#[allow(private_bounds)]
433	pub struct SubscriptionOwnedIter<T: StreamDecoder<T>> {
434	subscription: Subscription<T>,
435	}
436
437	impl<T: StreamDecoder<T>> Iterator for SubscriptionOwnedIter<T> {
438	type Item = T;
439
440	fn next(&mut self) -> Option<Self::Item> {	6✔
441	self.subscription.next()	6✔
442	}	6✔
443	}
444
445	impl<T: StreamDecoder<T>> IntoIterator for Subscription<T> {
446	type Item = T;
447	type IntoIter = SubscriptionOwnedIter<T>;
448
449	fn into_iter(self) -> Self::IntoIter {	2✔
450	SubscriptionOwnedIter { subscription: self }	2✔
451	}	2✔
452	}
453
454	/// An iterator that yields items as they become available without blocking.
455	#[allow(private_bounds)]
456	pub struct SubscriptionTryIter<'a, T: StreamDecoder<T>> {
457	subscription: &'a Subscription<T>,
458	}
459
460	impl<T: StreamDecoder<T>> Iterator for SubscriptionTryIter<'_, T> {
461	type Item = T;
462
463	fn next(&mut self) -> Option<Self::Item> {	×
464	self.subscription.try_next()	×
465	}	×
466	}
467
468	/// An iterator that yields items with a timeout.
469	#[allow(private_bounds)]
470	pub struct SubscriptionTimeoutIter<'a, T: StreamDecoder<T>> {
471	subscription: &'a Subscription<T>,
472	timeout: Duration,
473	}
474
475	impl<T: StreamDecoder<T>> Iterator for SubscriptionTimeoutIter<'_, T> {
476	type Item = T;
477
478	fn next(&mut self) -> Option<Self::Item> {	×
479	self.subscription.next_timeout(self.timeout)	×
480	}	×
481	}
482
483	/// Marker trait for subscriptions that share a channel based on message type
484	pub trait SharesChannel {}
485
486	#[cfg(all(test, feature = "sync"))]
487	mod tests {
488	use super::*;
489	use crate::messages::{encode_protobuf_message, OutgoingMessages, ResponseMessage};
490	use crate::stubs::MessageBusStub;
491	use std::sync::Arc;
492
493	#[derive(Debug)]
494	struct EndOfStreamItem;
495
496	impl StreamDecoder<EndOfStreamItem> for EndOfStreamItem {
497	fn decode(_context: &DecoderContext, _msg: &mut ResponseMessage) -> Result<EndOfStreamItem, Error> {	1✔
498	Err(Error::EndOfStream)	1✔
499	}	1✔
500
501	fn cancel_message(_server_version: i32, _id: Option<i32>, _context: Option<&DecoderContext>) -> Result<Vec<u8>, Error> {	1✔
502	Ok(encode_protobuf_message(OutgoingMessages::CancelMarketData as i32, &[]))	1✔
503	}	1✔
504	}
505
506	#[test]
507	fn test_subscription_skips_unexpected_messages_without_limit() {	1✔
508	use std::sync::atomic::AtomicUsize;
509
510	static CALL_COUNT: AtomicUsize = AtomicUsize::new(0);
511
512	#[derive(Debug)]
513	struct SkipThenSuccess;
514
515	impl StreamDecoder<SkipThenSuccess> for SkipThenSuccess {
516	fn decode(_context: &DecoderContext, _msg: &mut ResponseMessage) -> Result<SkipThenSuccess, Error> {	21✔
517	let n = CALL_COUNT.fetch_add(1, Ordering::Relaxed);	21✔
518	if n < 20 {	21✔
519	Err(Error::UnexpectedResponse(ResponseMessage::from("stray\0")))	20✔
520	} else {
521	Ok(SkipThenSuccess)	1✔
522	}
523	}	21✔
524	}
525
526	CALL_COUNT.store(0, Ordering::Relaxed);	1✔
527
528	// 20 stray messages + 1 valid (more than the old MAX_DECODE_RETRIES=10)
529	let mut responses: Vec<String> = (0..21).map(\|_\| "1\|msg".to_string()).collect();	21✔
530	// Sentinel to avoid blocking on the channel after success
531	responses.push("1\|done".to_string());	1✔
532
533	let stub = MessageBusStub::with_responses(responses);	1✔
534	let message_bus = Arc::new(stub);	1✔
535
536	let sub: Subscription<SkipThenSuccess> = {	1✔
537	let internal = message_bus.send_request(1, &[]).unwrap();	1✔
538	Subscription::new(message_bus.clone(), internal, DecoderContext::default())	1✔
539	};
540
541	let result = sub.next();	1✔
542	assert!(result.is_some(), "subscription should survive 20 skips and return valid message");	1✔
543	assert_eq!(CALL_COUNT.load(Ordering::Relaxed), 21);	1✔
544	}	1✔
545
546	#[test]
547	fn test_no_retries_after_end_of_stream() {	1✔
548	let stub = MessageBusStub::with_responses(vec![	1✔
549	"1\|data".to_string(), // triggers EndOfStream via decoder	1✔
550	"1\|stray".to_string(), // stray message after stream ended	1✔
551	]);
552	let message_bus = Arc::new(stub);	1✔
553
554	let sub: Subscription<EndOfStreamItem> = {	1✔
555	let internal = message_bus.send_request(1, &[]).unwrap();	1✔
556	Subscription::new(message_bus.clone(), internal, DecoderContext::default())	1✔
557	};
558
559	// First call hits EndOfStream, returns None
560	assert!(sub.next().is_none());	1✔
561
562	// Second call should return None immediately (stream_ended guard)
563	assert!(sub.next().is_none());	1✔
564	assert!(sub.stream_ended.load(Ordering::Relaxed));	1✔
565	}	1✔
566	}

wboayue / rust-ibapi / 25267123189

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