25258750129

Committed 02 May 2026 06:25PM UTC coverage: 87.582% (-2.3%) from 89.836%

Build # 25258750129

Build Type

push

github

Committed by

web-flow

Commit Message

refactor(test): remove MockGateway, collapse client/{sync,async} (PR 5/5) (#480)

Per todos/eliminate-mock-gateway.md PR 5 and the audit table.

Deletions (~6,400 LOC):
- client/sync/tests.rs (2,555 LOC, MockGateway-driven)
- client/async/tests.rs (2,527 LOC, MockGateway-driven)
- client/test_support/{mocks,scenarios,mod}.rs (~1,300 LOC)

Migrations:
- client/builders/{sync,async}.rs MockGateway test blocks now use
  Client::stubbed + MessageBusStub::default(). The fixtures only ever
  needed a Client instance, not a real gateway.

Layout:
- client/sync/{mod.rs,tests.rs} -> client/sync.rs (flat sibling, per
  CLAUDE.md item 13: a directory module that exists only to host
  tests.rs collapses to flat).
- client/async/{mod.rs,tests.rs} -> client/async.rs (same).
- client/mod.rs drops `mod test_support;`.

Coverage parity preserved:
- 109 of 117 deleted tests were duplicates of per-domain MessageBusStub
  tests (verified in todos/eliminate-mock-gateway-audit.md).
- 6 unique tests (handshake x2, disconnect x4) migrated in PR 4 to
  connection/{sync,async}_tests.rs via MemoryStream.
- 3 parity gaps closed in PR 4.

Coverage Stats

45 of 45 new or added lines in 2 files covered. (100.0%)

326 existing lines in 18 files now uncovered.

16871 of 19263 relevant lines covered (87.58%)

25.46 hits per line

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

71.19

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

wboayue / rust-ibapi / 25258750129

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