Refactors connection management (#127)

examples/account_summary.rs

@@ -2,13 +2,16 @@ use ibapi::accounts::{AccountSummaries, AccountSummaryTags};
 use ibapi::Client;
 
 fn main() {
+    env_logger::init();
+
     let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
 
     let group = "All";
 
     let subscription = client
         .account_summary(group, AccountSummaryTags::ALL)
         .expect("error requesting account summary");
+
     for update in &subscription {
         match update {
             AccountSummaries::Summary(summary) => println!("{summary:?}"),

examples/breakout.rs

@@ -6,6 +6,8 @@ use ibapi::orders::{order_builder, Action, OrderNotification};
 use ibapi::Client;
 
 fn main() {
+    env_logger::init();
+
     let client = Client::connect("127.0.0.1:4002", 100).unwrap();
 
     let symbol = "TSLA";

examples/contract_details.rs

@@ -2,6 +2,8 @@ use ibapi::contracts::Contract;
 use ibapi::Client;
 
 fn main() -> anyhow::Result<()> {
+    env_logger::init();
+
     let client = Client::connect("127.0.0.1:4002", 100)?;
 
     println!("server_version: {}", client.server_version());

examples/executions.rs

@@ -2,6 +2,8 @@ use ibapi::orders::ExecutionFilter;
 use ibapi::Client;
 
 fn main() -> anyhow::Result<()> {
+    env_logger::init();
+
     let mut filter = ExecutionFilter::default();
 
     filter.client_id = Some(32);

examples/family_codes.rs

@@ -1,6 +1,8 @@
 use ibapi::Client;
 
 fn main() {
+    env_logger::init();
+
     let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
 
     let family_codes = client.family_codes().expect("request failed");

examples/managed_accounts.rs

@@ -1,6 +1,8 @@
 use ibapi::Client;
 
 fn main() {
+    env_logger::init();
+
     let client = Client::connect("127.0.0.1:4002", 101).expect("connection failed");
 
     let accounts = client.managed_accounts().expect("error requesting managed accounts");

examples/market_rule.rs

@@ -1,6 +1,8 @@
 use ibapi::Client;
 
 fn main() {
+    env_logger::init();
+
     let client = Client::connect("127.0.0.1:4002", 100).unwrap();
 
     let market_rule_id = 12;

examples/matching_symbols.rs

@@ -1,6 +1,8 @@
 use ibapi::Client;
 
 fn main() {
+    env_logger::init();
+
     let client = Client::connect("127.0.0.1:4002", 100).unwrap();
 
     let pattern = "TSLA";

examples/next_order_id.rs

@@ -1,6 +1,8 @@
 use ibapi::Client;
 
 fn main() {
+    env_logger::init();
+
     let client = Client::connect("127.0.0.1:4002", 100).unwrap();
 
     let order_id = client.next_valid_order_id().unwrap();

examples/positions.rs

@@ -1,6 +1,8 @@
 use ibapi::{accounts::PositionUpdate, Client};
 
 fn main() {
+    env_logger::init();
+
     let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");
 
     let positions = client.positions().expect("request failed");

examples/positions_multi.rs

@@ -3,6 +3,8 @@ use std::env;
 use ibapi::Client;
 
 pub fn main() {
+    env_logger::init();
+
     let account = env::var("IBKR_ACCOUNT").expect("Please set IBKR_ACCOUNT environment variable to an account ID");
 
     let client = Client::connect("127.0.0.1:4002", 100).expect("connection failed");

examples/readme_connection.rs

@@ -2,6 +2,8 @@
 use ibapi::Client;
 
 fn main() {
+    env_logger::init();
+
     let connection_url = "127.0.0.1:4002";
 
     let _client = Client::connect(connection_url, 100).expect("connection to TWS failed!");

examples/readme_historical_data.rs

@@ -5,6 +5,8 @@ use ibapi::market_data::historical::{BarSize, ToDuration, WhatToShow};
 use ibapi::Client;
 
 fn main() {
+    env_logger::init();
+
     let connection_url = "127.0.0.1:4002";
     let client = Client::connect(connection_url, 100).expect("connection to TWS failed!");
 

examples/readme_place_order.rs

@@ -3,6 +3,8 @@ use ibapi::orders::{order_builder, Action, OrderNotification};
 use ibapi::Client;
 
 pub fn main() {
+    env_logger::init();
+
     let connection_url = "127.0.0.1:4002";
     let client = Client::connect(connection_url, 100).expect("connection to TWS failed!");
 

examples/readme_realtime_data_1.rs

@@ -3,6 +3,8 @@ use ibapi::market_data::realtime::{BarSize, WhatToShow};
 use ibapi::Client;
 
 fn main() {
+    env_logger::init();
+
     let connection_url = "127.0.0.1:4002";
     let client = Client::connect(connection_url, 100).expect("connection to TWS failed!");
 

src/accounts.rs

@@ -9,11 +9,10 @@
 //! - Real-time PnL updates for individual positions
 //!
 
-use std::marker::PhantomData;
-
 use crate::client::{SharesChannel, Subscribable, Subscription};
 use crate::contracts::Contract;
 use crate::messages::{IncomingMessages, OutgoingMessages, RequestMessage, ResponseMessage};
+use crate::transport::Response;
 use crate::{server_versions, Client, Error};
 
 mod decoders;
@@ -329,14 +328,9 @@ pub(crate) fn positions(client: &Client) -> Result<Subscription<PositionUpdate>,
     client.check_server_version(server_versions::ACCOUNT_SUMMARY, "It does not support position requests.")?;
 
     let request = encoders::encode_request_positions()?;
-    let responses = client.send_shared_request(OutgoingMessages::RequestPositions, request)?;
-
-    Ok(Subscription {
-        client,
-        request_id: None,
-        subscription: responses,
-        phantom: PhantomData,
-    })
+    let subscription = client.send_shared_request(OutgoingMessages::RequestPositions, request)?;
+
+    Ok(Subscription::new(client, subscription))
 }
 
 impl SharesChannel for Subscription<'_, PositionUpdate> {}
@@ -349,16 +343,10 @@ pub(crate) fn positions_multi<'a>(
     client.check_server_version(server_versions::MODELS_SUPPORT, "It does not support positions multi requests.")?;
 
     let request_id = client.next_request_id();
-
     let request = encoders::encode_request_positions_multi(request_id, account, model_code)?;
-    let responses = client.send_request(request_id, request)?;
-
-    Ok(Subscription {
-        client,
-        request_id: Some(request_id),
-        subscription: responses,
-        phantom: PhantomData,
-    })
+    let subscription = client.send_request(request_id, request)?;
+
+    Ok(Subscription::new(client, subscription))
 }
 
 // Determine whether an account exists under an account family and find the account family code.
@@ -368,7 +356,8 @@ pub(crate) fn family_codes(client: &Client) -> Result<Vec<FamilyCode>, Error> {
     let request = encoders::encode_request_family_codes()?;
     let subscription = client.send_shared_request(OutgoingMessages::RequestFamilyCodes, request)?;
 
-    if let Some(mut message) = subscription.next() {
+    // TODO: enumerate
+    if let Some(Response::Message(mut message)) = subscription.next() {
         decoders::decode_family_codes(&mut message)
     } else {
         Ok(Vec::default())
@@ -385,16 +374,10 @@ pub(crate) fn pnl<'a>(client: &'a Client, account: &str, model_code: Option<&str
     client.check_server_version(server_versions::PNL, "It does not support PnL requests.")?;
 
     let request_id = client.next_request_id();
-
     let request = encoders::encode_request_pnl(request_id, account, model_code)?;
-    let responses = client.send_request(request_id, request)?;
-
-    Ok(Subscription {
-        client,
-        request_id: Some(request_id),
-        subscription: responses,
-        phantom: PhantomData,
-    })
+    let subscription = client.send_request(request_id, request)?;
+
+    Ok(Subscription::new(client, subscription))
 }
 
 // Requests real time updates for daily PnL of individual positions.
@@ -413,46 +396,36 @@ pub(crate) fn pnl_single<'a>(
     client.check_server_version(server_versions::REALIZED_PNL, "It does not support PnL requests.")?;
 
     let request_id = client.next_request_id();
-
     let request = encoders::encode_request_pnl_single(request_id, account, contract_id, model_code)?;
-    let responses = client.send_request(request_id, request)?;
-
-    Ok(Subscription {
-        client,
-        request_id: Some(request_id),
-        subscription: responses,
-        phantom: PhantomData,
-    })
+    let subscription = client.send_request(request_id, request)?;
+
+    Ok(Subscription::new(client, subscription))
 }
 
 pub fn account_summary<'a>(client: &'a Client, group: &str, tags: &[&str]) -> Result<Subscription<'a, AccountSummaries>, Error> {
     client.check_server_version(server_versions::ACCOUNT_SUMMARY, "It does not support account summary requests.")?;
 
     let request_id = client.next_request_id();
-
     let request = encoders::encode_request_account_summary(request_id, group, tags)?;
     let subscription = client.send_request(request_id, request)?;
 
-    Ok(Subscription {
-        client,
-        request_id: Some(request_id),
-        subscription,
-        phantom: PhantomData,
-    })
+    Ok(Subscription::new(client, subscription))
 }
 
 pub fn managed_accounts(client: &Client) -> Result<Vec<String>, Error> {
     let request = encoders::encode_request_managed_accounts()?;
     let subscription = client.send_shared_request(OutgoingMessages::RequestManagedAccounts, request)?;
 
     match subscription.next() {
-        Some(mut message) => {
+        Some(Response::Message(mut message)) => {
             message.skip(); // message type
             message.skip(); // message version
 
             let accounts = message.next_string()?;
             Ok(accounts.split(",").map(String::from).collect())
         }
+        Some(Response::Cancelled) => Err(Error::Cancelled),
+        Some(Response::Disconnected) => Err(Error::ConnectionFailed),
         None => Ok(Vec::default()),
     }
 }