refactor(historical): use Subscription<HistoricalBarUpdate> (closes #431) (#488)

* refactor(historical): replace HistoricalDataStreamingSubscription with generic Subscription

Custom struct duplicated cancel-on-drop, AtomicBool dedup, message_bus
storage, tokio::spawn in async Drop, and inline message dispatch — all
already provided by the generic Subscription<T>. Replaced with a 30-line
StreamDecoder impl (mirrors MarketDepths / Bar / Trade pattern) and a
3-line factory body. Net -271 lines.

Async API breaks: next() now returns Option<Result<HistoricalBarUpdate,
Error>> and there is no separate error() accessor. Sync API unchanged
in this PR; see #487 for the follow-up.

Drop tests rewired to drive through the public factory rather than the
deleted internal constructor, so they exercise the real cancel-on-drop
path end-to-end.

Closes #431.

* comment: collapse keepUpToDate constraint to inline

* simplify: shorten Subscription import path; drop narration comments

Author: wboayue

examples/async/historical_data.rs

@@ -263,7 +263,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     while let Some(update) = subscription.next().await {
         match update {
-            HistoricalBarUpdate::Historical(data) => {
+            Ok(HistoricalBarUpdate::Historical(data)) => {
                 println!("Received {} initial historical bars", data.bars.len());
                 if let Some(bar) = data.bars.last() {
                     println!(
@@ -277,7 +277,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 }
                 println!("Now streaming updates...");
             }
-            HistoricalBarUpdate::Update(bar) => {
+            Ok(HistoricalBarUpdate::Update(bar)) => {
                 println!(
                     "UPDATE: {} - O: ${:.2}, H: ${:.2}, L: ${:.2}, C: ${:.2}, V: {:.0}",
                     format!("{:02}:{:02}:{:02}", bar.date.hour(), bar.date.minute(), bar.date.second()),
@@ -288,10 +288,14 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                     bar.volume
                 );
             }
-            HistoricalBarUpdate::End { start, end } => {
+            Ok(HistoricalBarUpdate::End { start, end }) => {
                 println!("Stream ended: {} - {}", start, end);
                 break;
             }
+            Err(e) => {
+                eprintln!("Stream error: {e}");
+                break;
+            }
         }
     }
 

src/market_data/historical/async.rs

@@ -4,12 +4,12 @@ use std::sync::Arc;
 
 use log::{debug, error, warn};
 use time::OffsetDateTime;
-use time_tz::Tz;
 
 use crate::client::ClientRequestBuilders;
 use crate::contracts::Contract;
 use crate::messages::IncomingMessages;
 use crate::protocol::{check_version, Features};
+use crate::subscriptions::r#async::Subscription;
 use crate::transport::{AsyncInternalSubscription, AsyncMessageBus};
 use crate::{Client, Error, MAX_RETRIES};
 
@@ -250,17 +250,16 @@ impl Client {
         what_to_show: Option<WhatToShow>,
         trading_hours: TradingHours,
         keep_up_to_date: bool,
-    ) -> Result<HistoricalDataStreamingSubscription, Error> {
+    ) -> Result<Subscription<HistoricalBarUpdate>, Error> {
         if !contract.trading_class.is_empty() || contract.contract_id > 0 {
             check_version(self.server_version(), Features::TRADING_CLASS)?;
         }
 
-        // Note: end_date must be None when keepUpToDate=true (IBKR requirement)
         let builder = self.request();
         let request = encoders::encode_request_historical_data(
             builder.request_id(),
             contract,
-            None, // end_date must be None for keepUpToDate
+            None, // end_date must be None when keepUpToDate=true (IBKR requirement)
             duration,
             bar_size,
             what_to_show,
@@ -269,21 +268,7 @@ impl Client {
             &Vec::<crate::contracts::TagValue>::default(),
         )?;
 
-        let request_id = builder.request_id();
-        let subscription = builder.send_raw(request).await?;
-
-        let tz: &'static Tz = self.time_zone.unwrap_or_else(|| {
-            warn!("server timezone unknown. assuming UTC, but that may be incorrect!");
-            time_tz::timezones::db::UTC
-        });
-
-        Ok(HistoricalDataStreamingSubscription::new(
-            subscription,
-            self.server_version(),
-            tz,
-            request_id,
-            self.message_bus.clone(),
-        ))
+        builder.send::<HistoricalBarUpdate>(request).await
     }
 }
 
@@ -407,143 +392,6 @@ impl<T: TickDecoder<T> + Send> Drop for TickSubscription<T> {
     }
 }
 
-// === Historical Data Streaming with keepUpToDate ===
-
-/// Async subscription for streaming historical data with keepUpToDate=true.
-///
-/// This subscription first yields the initial historical bars as a `Historical` variant,
-/// then continues to yield streaming updates for the current bar as `Update` variants.
-pub struct HistoricalDataStreamingSubscription {
-    messages: AsyncInternalSubscription,
-    server_version: i32,
-    time_zone: &'static Tz,
-    error: Option<Error>,
-    request_id: i32,
-    message_bus: Arc<dyn AsyncMessageBus>,
-    cancelled: AtomicBool,
-}
-
-impl HistoricalDataStreamingSubscription {
-    fn new(
-        messages: AsyncInternalSubscription,
-        server_version: i32,
-        time_zone: &'static Tz,
-        request_id: i32,
-        message_bus: Arc<dyn AsyncMessageBus>,
-    ) -> Self {
-        Self {
-            messages,
-            server_version,
-            time_zone,
-            error: None,
-            request_id,
-            message_bus,
-            cancelled: AtomicBool::new(false),
-        }
-    }
-
-    /// Cancel the streaming subscription, sending CancelHistoricalData to the server.
-    pub async fn cancel(&self) {
-        if self.cancelled.swap(true, Ordering::Relaxed) {
-            return;
-        }
-        if let Ok(message) = encoders::encode_cancel_historical_data(self.request_id) {
-            if let Err(e) = self.message_bus.cancel_subscription(self.request_id, message).await {
-                warn!("error sending cancel historical data: {e}");
-            }
-        }
-    }
-
-    /// Get the next update from the streaming subscription.
-    ///
-    /// Returns:
-    /// - `Some(HistoricalBarUpdate::Historical(data))` - Initial batch of historical bars (always first)
-    /// - `Some(HistoricalBarUpdate::Update(bar))` - Streaming bar update
-    /// - `None` - Subscription ended (connection closed or error)
-    pub async fn next(&mut self) -> Option<HistoricalBarUpdate> {
-        loop {
-            match self.messages.next().await {
-                Some(Ok(mut message)) => {
-                    match message.message_type() {
-                        IncomingMessages::HistoricalData => {
-                            // Initial historical data batch
-                            match decoders::decode_historical_data(self.server_version, self.time_zone, &mut message) {
-                                Ok(data) => {
-                                    return Some(HistoricalBarUpdate::Historical(data));
-                                }
-                                Err(e) => {
-                                    self.error = Some(e);
-                                    return None;
-                                }
-                            }
-                        }
-                        IncomingMessages::HistoricalDataUpdate => {
-                            // Streaming bar update
-                            match decoders::decode_historical_data_update(self.time_zone, &mut message) {
-                                Ok(bar) => {
-                                    return Some(HistoricalBarUpdate::Update(bar));
-                                }
-                                Err(e) => {
-                                    self.error = Some(e);
-                                    return None;
-                                }
-                            }
-                        }
-                        IncomingMessages::HistoricalDataEnd => {
-                            match decoders::decode_historical_data_end(self.server_version, self.time_zone, &mut message) {
-                                Ok((start, end)) => return Some(HistoricalBarUpdate::End { start, end }),
-                                Err(e) => {
-                                    self.error = Some(e);
-                                    return None;
-                                }
-                            }
-                        }
-                        IncomingMessages::Error => {
-                            self.error = Some(Error::from(message));
-                            return None;
-                        }
-                        _ => {
-                            // Skip unexpected messages
-                            debug!("unexpected message in streaming subscription: {:?}", message.message_type());
-                            continue;
-                        }
-                    }
-                }
-                Some(Err(e)) => {
-                    self.error = Some(e);
-                    return None;
-                }
-                None => {
-                    // Channel closed
-                    return None;
-                }
-            }
-        }
-    }
-
-    /// Returns the last error that occurred, if any.
-    pub fn error(&self) -> Option<&Error> {
-        self.error.as_ref()
-    }
-}
-
-impl Drop for HistoricalDataStreamingSubscription {
-    fn drop(&mut self) {
-        if self.cancelled.swap(true, Ordering::Relaxed) {
-            return;
-        }
-        let request_id = self.request_id;
-        let message_bus = self.message_bus.clone();
-        if let Ok(message) = encoders::encode_cancel_historical_data(request_id) {
-            tokio::spawn(async move {
-                if let Err(e) = message_bus.cancel_subscription(request_id, message).await {
-                    warn!("error sending cancel historical data in drop: {e}");
-                }
-            });
-        }
-    }
-}
-
 #[cfg(test)]
 #[path = "async_tests.rs"]
 mod tests;

src/market_data/historical/async_tests.rs

@@ -1,5 +1,5 @@
 use super::*;
-use crate::common::test_utils::helpers::assert_proto_msg_id;
+use crate::common::test_utils::helpers::{assert_proto_msg_id, count_proto_msgs};
 use crate::contracts::{Contract, Currency, Exchange, SecurityType, Symbol};
 use crate::messages::OutgoingMessages;
 use crate::server_versions;
@@ -579,9 +579,8 @@ async fn test_historical_data_streaming_with_updates() {
         .expect("streaming request should succeed");
 
     // First: receive initial historical data
-    let update1 = subscription.next().await;
-    assert!(update1.is_some(), "Should receive initial historical data");
-    match update1.unwrap() {
+    let update1 = subscription.next().await.expect("Should receive initial historical data");
+    match update1.expect("decode should succeed") {
         HistoricalBarUpdate::Historical(data) => {
             assert_eq!(data.bars.len(), 1, "Should have 1 initial bar");
             assert_eq!(data.bars[0].open, 185.50, "Wrong open price");
@@ -590,9 +589,8 @@ async fn test_historical_data_streaming_with_updates() {
     }
 
     // Second: receive streaming update
-    let update2 = subscription.next().await;
-    assert!(update2.is_some(), "Should receive streaming update");
-    match update2.unwrap() {
+    let update2 = subscription.next().await.expect("Should receive streaming update");
+    match update2.expect("decode should succeed") {
         HistoricalBarUpdate::Update(bar) => {
             assert_eq!(bar.open, 185.80, "Wrong open price in update");
             assert_eq!(bar.high, 186.10, "Wrong high price in update");
@@ -635,9 +633,8 @@ async fn test_historical_data_streaming_keep_up_to_date_false() {
         .expect("streaming request should succeed");
 
     // Receive initial historical data
-    let update1 = subscription.next().await;
-    assert!(update1.is_some(), "Should receive initial historical data");
-    match update1.unwrap() {
+    let update1 = subscription.next().await.expect("Should receive initial historical data");
+    match update1.expect("decode should succeed") {
         HistoricalBarUpdate::Historical(data) => {
             assert_eq!(data.bars.len(), 1, "Should have 1 initial bar");
         }
@@ -677,17 +674,10 @@ async fn test_historical_data_streaming_error_response() {
         .await
         .expect("streaming request should succeed");
 
-    // Should return None due to error
-    let update = subscription.next().await;
-    assert!(update.is_none(), "Should return None on error");
-
-    // Error should be accessible
-    let error = subscription.error();
-    assert!(error.is_some(), "Error should be stored");
-    assert!(
-        error.unwrap().to_string().contains("No market data permissions"),
-        "Error should contain the message"
-    );
+    // Should yield Some(Err(_)) — Subscription<T> surfaces errors through next().
+    let update = subscription.next().await.expect("error should arrive as Some(Err(_))");
+    let err = update.expect_err("Should yield error result");
+    assert!(err.to_string().contains("No market data permissions"), "Error should contain the message");
 }
 
 #[tokio::test]
@@ -762,27 +752,32 @@ async fn test_streaming_subscription_sends_cancel_on_drop() {
         response_messages: vec![],
     });
 
-    let (_tx, rx) = tokio::sync::broadcast::channel(16);
-    let internal = AsyncInternalSubscription::new(rx);
-    let request_id = 9000;
+    let mut client = Client::stubbed(message_bus.clone(), server_versions::SIZE_RULES);
+    client.time_zone = Some(time_tz::timezones::db::UTC);
+    let contract = Contract::stock("SPY").build();
 
     {
-        let _subscription = HistoricalDataStreamingSubscription::new(
-            internal,
-            server_versions::SIZE_RULES,
-            time_tz::timezones::db::UTC,
-            request_id,
-            message_bus.clone(),
-        );
-        // subscription dropped here
+        let _subscription = client
+            .historical_data_streaming(
+                &contract,
+                Duration::days(1),
+                BarSize::Hour,
+                Some(WhatToShow::Trades),
+                TradingHours::Regular,
+                true,
+            )
+            .await
+            .expect("streaming request should succeed");
     }
 
-    // Give tokio::spawn time to execute
     tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
 
     let messages = message_bus.request_messages.read().unwrap();
-    assert_eq!(messages.len(), 1, "should send cancel message on drop");
-    assert_proto_msg_id(&messages[0], OutgoingMessages::CancelHistoricalData);
+    assert_eq!(
+        count_proto_msgs(&messages, OutgoingMessages::CancelHistoricalData),
+        1,
+        "should send exactly one cancel message on drop"
+    );
 }
 
 #[tokio::test]
@@ -792,27 +787,32 @@ async fn test_streaming_subscription_cancel_prevents_duplicate_on_drop() {
         response_messages: vec![],
     });
 
-    let (_tx, rx) = tokio::sync::broadcast::channel(16);
-    let internal = AsyncInternalSubscription::new(rx);
-    let request_id = 9001;
+    let mut client = Client::stubbed(message_bus.clone(), server_versions::SIZE_RULES);
+    client.time_zone = Some(time_tz::timezones::db::UTC);
+    let contract = Contract::stock("SPY").build();
 
     {
-        let subscription = HistoricalDataStreamingSubscription::new(
-            internal,
-            server_versions::SIZE_RULES,
-            time_tz::timezones::db::UTC,
-            request_id,
-            message_bus.clone(),
-        );
-
-        // Explicit cancel
-        subscription.cancel().await;
+        let subscription = client
+            .historical_data_streaming(
+                &contract,
+                Duration::days(1),
+                BarSize::Hour,
+                Some(WhatToShow::Trades),
+                TradingHours::Regular,
+                true,
+            )
+            .await
+            .expect("streaming request should succeed");
 
-        // Drop should not send a second cancel
+        subscription.cancel().await;
     }
 
     tokio::time::sleep(tokio::time::Duration::from_millis(50)).await;
 
     let messages = message_bus.request_messages.read().unwrap();
-    assert_eq!(messages.len(), 1, "should send cancel only once");
+    assert_eq!(
+        count_proto_msgs(&messages, OutgoingMessages::CancelHistoricalData),
+        1,
+        "should send cancel only once"
+    );
 }