Make Client: Send + Sync (#116)

Author: mondeja

src/client.rs

@@ -1,7 +1,7 @@
-use std::cell::RefCell;
 use std::fmt::Debug;
 use std::io::Write;
 use std::sync::atomic::{AtomicI32, Ordering};
+use std::sync::{Arc, Mutex};
 
 use byteorder::{BigEndian, WriteBytesExt};
 use log::{debug, error, info};
@@ -40,7 +40,7 @@ pub struct Client {
 
     managed_accounts: String,
     client_id: i32, // ID of client.
-    pub(crate) message_bus: RefCell<Box<dyn MessageBus>>,
+    pub(crate) message_bus: Arc<Mutex<dyn MessageBus>>,
     next_request_id: AtomicI32, // Next available request_id.
     order_id: AtomicI32,        // Next available order_id. Starts with value returned on connection.
 }
@@ -67,11 +67,11 @@ impl Client {
     /// println!("next_order_id: {}", client.next_order_id());
     /// ```
     pub fn connect(address: &str, client_id: i32) -> Result<Client, Error> {
-        let message_bus = RefCell::new(Box::new(TcpMessageBus::connect(address)?));
+        let message_bus = Arc::new(Mutex::new(TcpMessageBus::connect(address)?));
         Client::do_connect(client_id, message_bus)
     }
 
-    fn do_connect(client_id: i32, message_bus: RefCell<Box<dyn MessageBus>>) -> Result<Client, Error> {
+    fn do_connect(client_id: i32, message_bus: Arc<Mutex<dyn MessageBus>>) -> Result<Client, Error> {
         let mut client = Client {
             server_version: 0,
             connection_time: None,
@@ -87,7 +87,11 @@ impl Client {
         client.start_api()?;
         client.receive_account_info()?;
 
-        client.message_bus.borrow_mut().process_messages(client.server_version)?;
+        client
+            .message_bus
+            .lock()
+            .expect("MessageBus is poisoned")
+            .process_messages(client.server_version)?;
 
         Ok(client)
     }
@@ -98,9 +102,9 @@ impl Client {
         let version = format!("v{MIN_SERVER_VERSION}..{MAX_SERVER_VERSION}");
 
         let packet = prefix.to_owned() + &encode_packet(&version);
-        self.message_bus.borrow_mut().write(&packet)?;
+        self.message_bus.lock().expect("MessageBus is poisoned").write(&packet)?;
 
-        let ack = self.message_bus.borrow_mut().read_message();
+        let ack = self.message_bus.lock().expect("MessageBus is poisoned").read_message();
 
         match ack {
             Ok(mut response_message) => {
@@ -133,7 +137,7 @@ impl Client {
             prelude.push_field(&"");
         }
 
-        self.message_bus.borrow_mut().write_message(prelude)?;
+        self.message_bus.lock().expect("MessageBus is poisoned").write_message(prelude)?;
 
         Ok(())
     }
@@ -146,7 +150,7 @@ impl Client {
         let mut attempts = 0;
         const MAX_ATTEMPTS: i32 = 100;
         loop {
-            let mut message = self.message_bus.borrow_mut().read_message()?;
+            let mut message = self.message_bus.lock().expect("MessageBus is poisoned").read_message()?;
 
             match message.message_type() {
                 IncomingMessages::NextValidId => {
@@ -886,7 +890,7 @@ impl Client {
     // == Internal Use ==
 
     #[cfg(test)]
-    pub(crate) fn stubbed(message_bus: RefCell<Box<dyn MessageBus>>, server_version: i32) -> Client {
+    pub(crate) fn stubbed(message_bus: Arc<Mutex<dyn MessageBus>>, server_version: i32) -> Client {
         Client {
             server_version: server_version,
             connection_time: None,
@@ -900,47 +904,56 @@ impl Client {
     }
 
     pub(crate) fn send_message(&self, packet: RequestMessage) -> Result<(), Error> {
-        self.message_bus.borrow_mut().write_message(&packet)
+        self.message_bus.lock().expect("MessageBus is poisoned").write_message(&packet)
     }
 
     pub(crate) fn send_request(&self, request_id: i32, message: RequestMessage) -> Result<ResponseIterator, Error> {
         debug!("send_message({:?}, {:?})", request_id, message);
-        self.message_bus.borrow_mut().send_generic_message(request_id, &message)
+        self.message_bus
+            .lock()
+            .expect("MessageBus is poisoned")
+            .send_generic_message(request_id, &message)
     }
 
     pub(crate) fn send_durable_request(&self, request_id: i32, message: RequestMessage) -> Result<ResponseIterator, Error> {
         debug!("send_durable_request({:?}, {:?})", request_id, message);
-        self.message_bus.borrow_mut().send_durable_message(request_id, &message)
+        self.message_bus
+            .lock()
+            .expect("MessageBus is poisoned")
+            .send_durable_message(request_id, &message)
     }
 
     pub(crate) fn send_order(&self, order_id: i32, message: RequestMessage) -> Result<ResponseIterator, Error> {
         debug!("send_order({:?}, {:?})", order_id, message);
-        self.message_bus.borrow_mut().send_order_message(order_id, &message)
+        self.message_bus
+            .lock()
+            .expect("MessageBus is poisoned")
+            .send_order_message(order_id, &message)
     }
 
     /// Sends request for the next valid order id.
     pub(crate) fn request_next_order_id(&self, message: RequestMessage) -> Result<GlobalResponseIterator, Error> {
-        self.message_bus.borrow_mut().request_next_order_id(&message)
+        self.message_bus.lock().expect("MessageBus is poisoned").request_next_order_id(&message)
     }
 
     /// Sends request for open orders.
     pub(crate) fn request_order_data(&self, message: RequestMessage) -> Result<GlobalResponseIterator, Error> {
-        self.message_bus.borrow_mut().request_open_orders(&message)
+        self.message_bus.lock().expect("MessageBus is poisoned").request_open_orders(&message)
     }
 
     /// Sends request for market rule.
     pub(crate) fn request_market_rule(&self, message: RequestMessage) -> Result<GlobalResponseIterator, Error> {
-        self.message_bus.borrow_mut().request_market_rule(&message)
+        self.message_bus.lock().expect("MessageBus is poisoned").request_market_rule(&message)
     }
 
     /// Sends request for positions.
     pub(crate) fn request_positions(&self, message: RequestMessage) -> Result<GlobalResponseIterator, Error> {
-        self.message_bus.borrow_mut().request_positions(&message)
+        self.message_bus.lock().expect("MessageBus is poisoned").request_positions(&message)
     }
 
     /// Sends request for family codes.
     pub(crate) fn request_family_codes(&self, message: RequestMessage) -> Result<GlobalResponseIterator, Error> {
-        self.message_bus.borrow_mut().request_family_codes(&message)
+        self.message_bus.lock().expect("MessageBus is poisoned").request_family_codes(&message)
     }
 
     pub(crate) fn check_server_version(&self, version: i32, message: &str) -> Result<(), Error> {

src/client/transport.rs

@@ -5,6 +5,7 @@ use std::net::TcpStream;
 use std::sync::{Arc, RwLock};
 use std::thread::{self, JoinHandle};
 use std::time::Duration;
+use std::sync::Mutex;
 
 use byteorder::{BigEndian, ReadBytesExt, WriteBytesExt};
 use crossbeam::channel::{self, Receiver, Sender};
@@ -17,7 +18,7 @@ use recorder::MessageRecorder;
 
 mod recorder;
 
-pub(crate) trait MessageBus {
+pub(crate) trait MessageBus: Send + Sync {
     fn read_message(&mut self) -> Result<ResponseMessage, Error>;
 
     fn write_message(&mut self, packet: &RequestMessage) -> Result<(), Error>;
@@ -43,7 +44,7 @@ pub(crate) trait MessageBus {
 #[derive(Debug)]
 pub struct TcpMessageBus {
     reader: Arc<TcpStream>,
-    writer: Box<TcpStream>,
+    writer: Arc<Mutex<TcpStream>>,
     handles: Vec<JoinHandle<i32>>,
     requests: Arc<SenderHash<i32, ResponseMessage>>,
     orders: Arc<SenderHash<i32, ResponseMessage>>,
@@ -101,7 +102,7 @@ impl TcpMessageBus {
         let stream = TcpStream::connect(connection_string)?;
 
         let reader = Arc::new(stream.try_clone()?);
-        let writer = Box::new(stream);
+        let writer = Arc::new(Mutex::new(stream));
         let requests = Arc::new(SenderHash::new());
         let orders = Arc::new(SenderHash::new());
 
@@ -213,7 +214,7 @@ impl MessageBus for TcpMessageBus {
         packet.write_u32::<BigEndian>(data.len() as u32)?;
         packet.write_all(data)?;
 
-        self.writer.write_all(&packet)?;
+        self.writer.lock().expect("MessageBus writer is poisoned").write_all(&packet)?;
 
         self.recorder.record_request(message);
 
@@ -222,7 +223,7 @@ impl MessageBus for TcpMessageBus {
 
     fn write(&mut self, data: &str) -> Result<(), Error> {
         debug!("{data:?} ->");
-        self.writer.write_all(data.as_bytes())?;
+        self.writer.lock().expect("MessageBus writer is poisoned").write_all(data.as_bytes())?;
         Ok(())
     }
 

src/contracts/tests.rs

@@ -1,13 +1,14 @@
-use std::cell::RefCell;
+use std::sync::RwLock;
+use std::sync::{Arc, Mutex};
 
 use super::*;
 
 use crate::stubs::MessageBusStub;
 
 #[test]
 fn request_stock_contract_details() {
-    let message_bus = RefCell::new(Box::new(MessageBusStub{
-        request_messages: RefCell::new(vec![]),
+    let message_bus = Arc::new(Mutex::new(MessageBusStub{
+        request_messages: RwLock::new(vec![]),
         response_messages: vec![
             "10|9001|TSLA|STK||0||SMART|USD|TSLA|NMS|NMS|76792991|0.01||ACTIVETIM,AD,ADJUST,ALERT,ALGO,ALLOC,AON,AVGCOST,BASKET,BENCHPX,CASHQTY,COND,CONDORDER,DARKONLY,DARKPOLL,DAY,DEACT,DEACTDIS,DEACTEOD,DIS,DUR,GAT,GTC,GTD,GTT,HID,IBKRATS,ICE,IMB,IOC,LIT,LMT,LOC,MIDPX,MIT,MKT,MOC,MTL,NGCOMB,NODARK,NONALGO,OCA,OPG,OPGREROUT,PEGBENCH,PEGMID,POSTATS,POSTONLY,PREOPGRTH,PRICECHK,REL,REL2MID,RELPCTOFS,RPI,RTH,SCALE,SCALEODD,SCALERST,SIZECHK,SNAPMID,SNAPMKT,SNAPREL,STP,STPLMT,SWEEP,TRAIL,TRAILLIT,TRAILLMT,TRAILMIT,WHATIF|SMART,AMEX,NYSE,CBOE,PHLX,ISE,CHX,ARCA,ISLAND,DRCTEDGE,BEX,BATS,EDGEA,CSFBALGO,JEFFALGO,BYX,IEX,EDGX,FOXRIVER,PEARL,NYSENAT,LTSE,MEMX,PSX|1|0|TESLA INC|NASDAQ||Consumer, Cyclical|Auto Manufacturers|Auto-Cars/Light Trucks|US/Eastern|20221229:0400-20221229:2000;20221230:0400-20221230:2000;20221231:CLOSED;20230101:CLOSED;20230102:CLOSED;20230103:0400-20230103:2000|20221229:0930-20221229:1600;20221230:0930-20221230:1600;20221231:CLOSED;20230101:CLOSED;20230102:CLOSED;20230103:0930-20230103:1600|||1|ISIN|US88160R1014|1|||26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26||COMMON|1|1|100||".to_string(),
             "10|9001|TSLA|STK||0||AMEX|USD|TSLA|NMS|NMS|76792991|0.01||ACTIVETIM,AD,ADJUST,ALERT,ALLOC,AVGCOST,BASKET,BENCHPX,CASHQTY,COND,CONDORDER,DAY,DEACT,DEACTDIS,DEACTEOD,GAT,GTC,GTD,GTT,HID,IOC,LIT,LMT,MIT,MKT,MTL,NGCOMB,NONALGO,OCA,PEGBENCH,SCALE,SCALERST,SNAPMID,SNAPMKT,SNAPREL,STP,STPLMT,TRAIL,TRAILLIT,TRAILLMT,TRAILMIT,WHATIF|SMART,AMEX,NYSE,CBOE,PHLX,ISE,CHX,ARCA,ISLAND,DRCTEDGE,BEX,BATS,EDGEA,CSFBALGO,JEFFALGO,BYX,IEX,EDGX,FOXRIVER,PEARL,NYSENAT,LTSE,MEMX,PSX|1|0|TESLA INC|NASDAQ||Consumer, Cyclical|Auto Manufacturers|Auto-Cars/Light Trucks|US/Eastern|20221229:0700-20221229:2000;20221230:0700-20221230:2000;20221231:CLOSED;20230101:CLOSED;20230102:CLOSED;20230103:0700-20230103:2000|20221229:0700-20221229:2000;20221230:0700-20221230:2000;20221231:CLOSED;20230101:CLOSED;20230102:CLOSED;20230103:0700-20230103:2000|||1|ISIN|US88160R1014|1|||26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26,26||COMMON|1|1|100||".to_string(),
@@ -21,7 +22,7 @@ fn request_stock_contract_details() {
 
     let results = client.contract_details(&contract);
 
-    let request_messages = client.message_bus.borrow().request_messages();
+    let request_messages = client.message_bus.lock().expect("MessageBus is poisoned").request_messages();
 
     assert_eq!(request_messages[0].encode_simple(), "9|8|9000|0|TSLA|STK||0|||SMART||USD|||0|||");
 

src/market_data/historical/tests.rs

@@ -1,4 +1,5 @@
-use std::cell::RefCell;
+use std::sync::RwLock;
+use std::sync::{Arc, Mutex};
 
 use time::macros::datetime;
 
@@ -10,8 +11,8 @@ use super::*;
 
 #[test]
 fn test_head_timestamp() {
-    let message_bus = RefCell::new(Box::new(MessageBusStub {
-        request_messages: RefCell::new(vec![]),
+    let message_bus = Arc::new(Mutex::new(MessageBusStub {
+        request_messages: RwLock::new(vec![]),
         response_messages: vec!["9|9000|1678323335|".to_owned()],
     }));
 
@@ -27,7 +28,7 @@ fn test_head_timestamp() {
 
     assert_eq!(head_timestamp, OffsetDateTime::from_unix_timestamp(1678323335).unwrap(), "bar.date");
 
-    let request_messages = client.message_bus.borrow().request_messages();
+    let request_messages = client.message_bus.lock().expect("MessageBus is poisoned").request_messages();
 
     let head_timestamp_request = &request_messages[0];
     assert_eq!(
@@ -70,8 +71,8 @@ fn test_histogram_data() {
 
 #[test]
 fn test_historical_data() {
-    let message_bus = RefCell::new(Box::new(MessageBusStub {
-        request_messages: RefCell::new(vec![]),
+    let message_bus = Arc::new(Mutex::new(MessageBusStub {
+        request_messages: RwLock::new(vec![]),
         response_messages: vec![
             "17\09000\020230413  16:31:22\020230415  16:31:22\02\020230413\0182.9400\0186.5000\0180.9400\0185.9000\0948837.22\0184.869\0324891\020230414\0183.8800\0186.2800\0182.0100\0185.0000\0810998.27\0183.9865\0277547\0".to_owned()
         ],
@@ -107,7 +108,7 @@ fn test_historical_data() {
 
     // Assert Request
 
-    let request_messages = client.message_bus.borrow().request_messages();
+    let request_messages = client.message_bus.lock().expect("MessageBus is poisoned").request_messages();
 
     let head_timestamp_request = &request_messages[0];
     assert_eq!(

src/market_data/realtime/tests.rs

@@ -1,4 +1,5 @@
-use std::cell::RefCell;
+use std::sync::RwLock;
+use std::sync::{Arc, Mutex};
 
 use time::OffsetDateTime;
 
@@ -11,8 +12,8 @@ use super::*;
 
 #[test]
 fn realtime_bars() {
-    let message_bus = RefCell::new(Box::new(MessageBusStub {
-        request_messages: RefCell::new(vec![]),
+    let message_bus = Arc::new(Mutex::new(MessageBusStub {
+        request_messages: RwLock::new(vec![]),
         response_messages: vec!["50|3|9001|1678323335|4028.75|4029.00|4028.25|4028.50|2|4026.75|1|".to_owned()],
     }));
 
@@ -46,7 +47,7 @@ fn realtime_bars() {
     // Should trigger cancel realtime bars
     drop(bars);
 
-    let request_messages = client.message_bus.borrow().request_messages();
+    let request_messages = client.message_bus.lock().expect("MessageBus is poisoned").request_messages();
 
     // Verify Requests
     let realtime_bars_request = &request_messages[0];