Implements account updates (#126)

examples/managed_accounts.rs

@@ -5,4 +5,4 @@ fn main() {
 
     let accounts = client.managed_accounts().expect("error requesting managed accounts");
     println!("managed accounts: {accounts:?}")
-}
+}

src/accounts.rs

@@ -272,6 +272,57 @@ pub struct FamilyCode {
     pub family_code: String,
 }
 
+/// Account's information, portfolio and last update time
+#[allow(clippy::large_enum_variant)]
+pub enum AccountUpdates {
+    /// Receives the subscribed account's information.
+    Value(AccountValue),
+    /// Receives the subscribed account's portfolio.
+    Portfolio(AccountPortfolio),
+    /// Receives the last time on which the account was updated.
+    Time(AccountTime),
+    /// Notifies when all the account’s information has finished.
+    End,
+}
+
+/// A value of subscribed account's information.
+pub struct AccountValue {
+    /// The value being updated.
+    pub key: String,
+    /// Current value
+    pub value: String,
+    /// The currency inn which the value is expressed.
+    pub currency: String,
+    /// The account identifier.
+    pub account: String,
+}
+
+/// Subscribed account's portfolio.
+pub struct AccountPortfolio {
+    /// The Contract for which a position is held.
+    pub contract: Contract,
+    /// The number of positions held.
+    pub position: f64,
+    /// The instrument's unitary price
+    pub market_price: f64,
+    /// Total market value of the instrument.
+    pub market_value: f64,
+    /// Average cost of the overall position.
+    pub average_cost: f64,
+    /// Daily unrealized profit and loss on the position.
+    pub unrealized_pnl: f64,
+    /// Daily realized profit and loss on the position.   
+    pub realized_pnl: f64,
+    /// Account identifier for the update.
+    pub account: String,
+}
+
+/// Last time at which the account was updated.
+pub struct AccountTime {
+    /// The last update system time.
+    pub timestamp: String,
+}
+
 // Subscribes to position updates for all accessible accounts.
 // All positions sent initially, and then only updates as positions change.
 pub(crate) fn positions(client: &Client) -> Result<Subscription<PositionUpdate>, Error> {

src/accounts/tests.rs

@@ -1,7 +1,7 @@
 use std::sync::{Arc, Mutex, RwLock};
 
-use crate::{accounts::AccountSummaryTags, server_versions, stubs::MessageBusStub, Client};
 use crate::testdata::responses;
+use crate::{accounts::AccountSummaryTags, server_versions, stubs::MessageBusStub, Client};
 
 #[test]
 fn test_pnl() {
@@ -117,9 +117,7 @@ fn test_account_summary() {
 fn test_managed_accounts() {
     let message_bus = Arc::new(Mutex::new(MessageBusStub {
         request_messages: RwLock::new(vec![]),
-        response_messages: vec![
-            responses::MANAGED_ACCOUNT.into(),
-        ],
+        response_messages: vec![responses::MANAGED_ACCOUNT.into()],
     }));
 
     let client = Client::stubbed(message_bus, server_versions::SIZE_RULES);

src/testdata/responses.rs

@@ -1,2 +1,2 @@
 pub const POSITION: &str = "61\03\0DU1234567\076792991\0TSLA\0STK\0\00.0\0\0\0NASDAQ\0USD\0TSLA\0NMS\0500\0196.77\0";
-pub const MANAGED_ACCOUNT: &str = "15|1|DU1234567,DU7654321|";
+pub const MANAGED_ACCOUNT: &str = "15|1|DU1234567,DU7654321|";

src/transport.rs

@@ -104,10 +104,9 @@ impl SharedChannels {
 
     // Get receiver for specified message type. Panics if receiver not found.
     pub fn get_receiver(&self, message_type: OutgoingMessages) -> Arc<Receiver<ResponseMessage>> {
-        let receiver = self
-            .receivers
-            .get(&message_type)
-            .unwrap_or_else(|| panic!("unsupported request message {message_type:?}. check mapping in SharedChannels::new() located in transport.rs"));
+        let receiver = self.receivers.get(&message_type).unwrap_or_else(|| {
+            panic!("unsupported request message {message_type:?}. check mapping in SharedChannels::new() located in transport.rs")
+        });
 
         Arc::clone(receiver)
     }
@@ -321,7 +320,7 @@ fn dispatch_message(
             } else {
                 process_response(requests, orders, shared_channels, message);
             }
-        },
+        }
         IncomingMessages::OrderStatus
         | IncomingMessages::OpenOrder
         | IncomingMessages::OpenOrderEnd

src/transport/recorder.rs

@@ -1,3 +1,11 @@
+//! The MessageRecorder is used to log interactions between the client and
+//! the TWS server.
+//! The record is enabled by setting the environment variable IBAPI_RECORDING_DIR
+//! IBAPI_RECORDING_DIR is set to the path to store logs
+//! e.g.  set to /tmp/logs
+//! /tmp/logs/0001-request.msg
+//! /tmp/logs/0002-response.msg
+
 use std::env;
 use std::fs;
 use std::sync::atomic::{AtomicUsize, Ordering};

src/transport/recorder/tests.rs

@@ -1,30 +1,38 @@
 use super::*;
 use std::env;
+use std::sync::Mutex;
+
+struct EnvMutex {}
+
+static ENV_MUTEX: Mutex<EnvMutex> = Mutex::new(EnvMutex {});
 
 #[test]
 fn env_var_enables_recorder() {
+    let _guard = ENV_MUTEX.lock().unwrap();
+
     let key = String::from("IBAPI_RECORDING_DIR");
     let dir = String::from("/tmp/records");
 
     env::set_var(&key, &dir);
 
     let recorder = MessageRecorder::new();
 
-    // TODO - refactor
-    // assert_eq!(true, recorder.enabled);
-    // assert!(&recorder.recording_dir.starts_with(&dir), "{} != {}", &recorder.recording_dir, &dir)
+    assert_eq!(true, recorder.enabled);
+    assert!(&recorder.recording_dir.starts_with(&dir), "{} != {}", &recorder.recording_dir, &dir)
 }
 
 #[test]
 fn recorder_is_disabled() {
+    let _guard = ENV_MUTEX.lock().unwrap();
+
     let key = String::from("IBAPI_RECORDING_DIR");
 
     env::set_var(&key, &"");
 
-    let _recorder = MessageRecorder::new();
+    let recorder = MessageRecorder::new();
 
-    // assert_eq!(false, recorder.enabled);
-    // assert_eq!("", &recorder.recording_dir);
+    assert_eq!(false, recorder.enabled);
+    assert_eq!("", &recorder.recording_dir);
 }
 
 #[test]