Refactor subscriptions (#122)

Refactor subscriptions

examples/breakout.rs

@@ -15,7 +15,7 @@ fn main() {
 
     let mut channel = BreakoutChannel::new(30);
 
-    for bar in bars {
+    for bar in bars.iter() {
         channel.add_bar(&bar);
 
         // Ensure enough bars and no open positions.

examples/pnl.rs

@@ -15,7 +15,7 @@ fn main() {
 
     let client = Client::connect(&gateway_url, 919).expect("connection failed");
 
-    let mut subscription = client.pnl(&account, None).expect("pnl request failed");
+    let subscription = client.pnl(&account, None).expect("pnl request failed");
 
     // Get next item non-blocking
     if let Some(pnl) = subscription.try_next() {

examples/pnl_single.rs

@@ -18,7 +18,7 @@ fn main() {
 
     let client = Client::connect(&gateway_url, 919).expect("connection failed");
 
-    let mut subscription = client.pnl_single(&account, contract_id, None).expect("pnl single request failed");
+    let subscription = client.pnl_single(&account, contract_id, None).expect("pnl single request failed");
 
     // Get next item non-blocking
     if let Some(pnl) = subscription.try_next() {

examples/positions.rs

@@ -3,7 +3,7 @@ use ibapi::{accounts::PositionUpdate, Client};
 fn main() {
     let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
 
-    let mut positions = client.positions().expect("request failed");
+    let positions = client.positions().expect("request failed");
     while let Some(position_update) = positions.next() {
         match position_update {
             PositionUpdate::Position(position) => {

examples/positions_multi.rs

@@ -8,7 +8,7 @@ pub fn main() {
     let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
 
     let subscription = client.positions_multi(Some(&account), None).expect("error requesting positions by model");
-    for position in subscription {
+    for position in subscription.iter() {
         println!("{position:?}")
     }
 }

examples/readme_realtime_data_1.rs

@@ -8,7 +8,7 @@ fn main() {
 
     // Request real-time bars data for AAPL with 5-second intervals
     let contract = Contract::stock("AAPL");
-    let mut subscription = client
+    let subscription = client
         .realtime_bars(&contract, BarSize::Sec5, WhatToShow::Trades, false)
         .expect("realtime bars request failed!");
 

examples/readme_realtime_data_2.rs

@@ -10,10 +10,10 @@ fn main() {
     let contract_aapl = Contract::stock("AAPL");
     let contract_nvda = Contract::stock("NVDA");
 
-    let mut subscription_aapl = client
+    let subscription_aapl = client
         .realtime_bars(&contract_aapl, BarSize::Sec5, WhatToShow::Trades, false)
         .expect("realtime bars request failed!");
-    let mut subscription_nvda = client
+    let subscription_nvda = client
         .realtime_bars(&contract_nvda, BarSize::Sec5, WhatToShow::Trades, false)
         .expect("realtime bars request failed!");
 

examples/stream_bars.rs

@@ -34,7 +34,7 @@ fn main() -> anyhow::Result<()> {
 
     let bars = client.realtime_bars(&contract, BarSize::Sec5, WhatToShow::Trades, false)?;
 
-    for (i, bar) in bars.enumerate().take(60) {
+    for (i, bar) in bars.iter().enumerate().take(60) {
         println!("bar: {i:?} {bar:?}");
     }
 

examples/tick_by_tick.rs

@@ -102,7 +102,7 @@ fn stream_mid_point(client: &mut Client, _symbol: &str) -> anyhow::Result<()> {
     let contract = contract_es();
     let ticks = client.tick_by_tick_midpoint(&contract, 0, false)?;
 
-    for (i, tick) in ticks.enumerate().take(60) {
+    for (i, tick) in ticks.iter().enumerate().take(60) {
         println!("tick: {i:?} {tick:?}");
     }
 

src/accounts.rs

@@ -108,7 +108,7 @@ impl Subscribable<PositionUpdate> for PositionUpdate {
     }
 
     fn cancel_message(_server_version: i32, _request_id: Option<i32>) -> Result<RequestMessage, Error> {
-        Ok(encoders::encode_cancel_positions()?)
+        encoders::encode_cancel_positions()
     }
 }
 
@@ -129,15 +129,15 @@ impl From<PositionMulti> for PositionUpdateMulti {
 #[derive(Debug, Clone, Default)]
 pub struct PositionMulti {
     /// The account holding the position.
-    account: String,
+    pub account: String,
     /// The model code holding the position.
-    model_code: String,
+    pub model_code: String,
     /// The position's Contract
-    contract: Contract,
+    pub contract: Contract,
     /// The number of positions held.
-    position: f64,
+    pub position: f64,
     /// The average cost of the position.
-    average_cost: f64,
+    pub average_cost: f64,
 }
 
 impl Subscribable<PositionUpdateMulti> for PositionUpdateMulti {
@@ -153,7 +153,7 @@ impl Subscribable<PositionUpdateMulti> for PositionUpdateMulti {
 
     fn cancel_message(_server_version: i32, request_id: Option<i32>) -> Result<RequestMessage, Error> {
         let request_id = request_id.expect("Request ID required to encode cancel positions multi");
-        Ok(encoders::encode_cancel_positions_multi(request_id)?)
+        encoders::encode_cancel_positions_multi(request_id)
     }
 }
 
@@ -207,11 +207,10 @@ pub(crate) fn positions_multi<'a>(
 pub(crate) fn family_codes(client: &Client) -> Result<Vec<FamilyCode>, Error> {
     client.check_server_version(server_versions::REQ_FAMILY_CODES, "It does not support family codes requests.")?;
 
-    let message = encoders::encode_request_family_codes()?;
-
-    let mut messages = client.send_shared_request(OutgoingMessages::RequestFamilyCodes, message)?;
+    let request = encoders::encode_request_family_codes()?;
+    let subscription = client.send_shared_request(OutgoingMessages::RequestFamilyCodes, request)?;
 
-    if let Some(mut message) = messages.next() {
+    if let Some(mut message) = subscription.next() {
         decoders::decode_family_codes(&mut message)
     } else {
         Ok(Vec::default())