refactored request order ids

Author: Wil Boayue

src/client.rs

@@ -86,10 +86,7 @@ impl Client {
         client.start_api()?;
         client.receive_account_info()?;
 
-        client
-            .message_bus
-            .lock()?
-            .process_messages(client.server_version)?;
+        client.message_bus.lock()?.process_messages(client.server_version)?;
 
         Ok(client)
     }
@@ -971,8 +968,8 @@ impl Client {
     }
 
     /// Sends request for the next valid order id.
-    pub(crate) fn request_next_order_id(&self, message: RequestMessage) -> Result<BusSubscription, Error> {
-        self.message_bus.lock()?.request_next_order_id(&message)
+    pub(crate) fn send_shared_message(&self, message_id: OutgoingMessages, message: RequestMessage) -> Result<BusSubscription, Error> {
+        self.message_bus.lock()?.send_shared_message(message_id, &message)
     }
 
     /// Sends request for open orders.

src/orders.rs

@@ -1371,7 +1371,7 @@ pub(crate) fn global_cancel(client: &Client) -> Result<(), Error> {
 pub(crate) fn next_valid_order_id(client: &Client) -> Result<i32, Error> {
     let message = encoders::encode_next_valid_order_id()?;
 
-    let mut messages = client.request_next_order_id(message)?;
+    let mut messages = client.send_shared_message(OutgoingMessages::RequestIds, message)?;
 
     if let Some(message) = messages.next() {
         let order_id_index = 2;

src/stubs.rs

@@ -2,7 +2,7 @@ use std::sync::{Arc, RwLock};
 
 use crossbeam::channel;
 
-use crate::messages::{RequestMessage, ResponseMessage};
+use crate::messages::{OutgoingMessages, RequestMessage, ResponseMessage};
 use crate::transport::{BusSubscription, MessageBus, SubscriptionBuilder};
 use crate::Error;
 
@@ -24,10 +24,7 @@ impl MessageBus for MessageBusStub {
     }
 
     fn write_message(&mut self, message: &RequestMessage) -> Result<(), Error> {
-        self.request_messages
-            .write()
-            .unwrap()
-            .push(message.clone());
+        self.request_messages.write().unwrap().push(message.clone());
         Ok(())
     }
 
@@ -43,7 +40,7 @@ impl MessageBus for MessageBusStub {
         mock_request(self, request_id, message)
     }
 
-    fn request_next_order_id(&mut self, message: &RequestMessage) -> Result<BusSubscription, Error> {
+    fn send_shared_message(&mut self, _message_id: OutgoingMessages, message: &RequestMessage) -> Result<BusSubscription, Error> {
         mock_global_request(self, message)
     }
 
@@ -73,10 +70,7 @@ impl MessageBus for MessageBusStub {
 }
 
 fn mock_request(stub: &mut MessageBusStub, _request_id: i32, message: &RequestMessage) -> Result<BusSubscription, Error> {
-    stub.request_messages
-        .write()
-        .unwrap()
-        .push(message.clone());
+    stub.request_messages.write().unwrap().push(message.clone());
 
     let (sender, receiver) = channel::unbounded();
     let (s1, _r1) = channel::unbounded();
@@ -91,10 +85,7 @@ fn mock_request(stub: &mut MessageBusStub, _request_id: i32, message: &RequestMe
 }
 
 fn mock_global_request(stub: &mut MessageBusStub, message: &RequestMessage) -> Result<BusSubscription, Error> {
-    stub.request_messages
-        .write()
-        .unwrap()
-        .push(message.clone());
+    stub.request_messages.write().unwrap().push(message.clone());
 
     let (sender, receiver) = channel::unbounded();
 

src/transport.rs

@@ -31,11 +31,8 @@ pub(crate) trait MessageBus: Send + Sync {
     fn send_durable_message(&mut self, request_id: i32, packet: &RequestMessage) -> Result<BusSubscription, Error>;
     fn send_order_message(&mut self, request_id: i32, packet: &RequestMessage) -> Result<BusSubscription, Error>;
 
-    fn send_shared_message(&mut self, message_id: OutgoingMessages, packet: &RequestMessage) -> Result<BusSubscription, Error> {
-        Ok(BusSubscription::new(todo!(), todo!(), todo!(), todo!(), todo!()))
-    }
+    fn send_shared_message(&mut self, message_id: OutgoingMessages, packet: &RequestMessage) -> Result<BusSubscription, Error>;
 
-    fn request_next_order_id(&mut self, message: &RequestMessage) -> Result<BusSubscription, Error>;
     fn request_open_orders(&mut self, message: &RequestMessage) -> Result<BusSubscription, Error>;
     fn request_market_rule(&mut self, message: &RequestMessage) -> Result<BusSubscription, Error>;
     fn request_positions(&mut self, message: &RequestMessage) -> Result<BusSubscription, Error>;
@@ -70,6 +67,20 @@ impl SharedChannels {
 
         // Register request/response pairs.
         instance.register(OutgoingMessages::RequestIds, &[IncomingMessages::NextValidId]);
+        instance.register(OutgoingMessages::RequestFamilyCodes, &[IncomingMessages::FamilyCodes]);
+        instance.register(OutgoingMessages::RequestMarketRule, &[IncomingMessages::MarketRule]);
+        instance.register(
+            OutgoingMessages::RequestPositions,
+            &[IncomingMessages::Position, IncomingMessages::PositionEnd],
+        );
+        instance.register(
+            OutgoingMessages::RequestPositionsMulti,
+            &[IncomingMessages::PositionMulti, IncomingMessages::PositionMultiEnd],
+        );
+        instance.register(
+            OutgoingMessages::RequestOpenOrders,
+            &[IncomingMessages::OpenOrder, IncomingMessages::OpenOrderEnd],
+        );
 
         instance
     }
@@ -88,15 +99,20 @@ impl SharedChannels {
     }
 
     pub fn get_receiver(&self, message_id: OutgoingMessages) -> Arc<Receiver<ResponseMessage>> {
-        let receiver = self.receivers.get(&message_id).expect("unsupport type");
+        let receiver = self
+            .receivers
+            .get(&message_id)
+            .expect(&format!("unsupported request message {:?}", message_id));
         Arc::clone(receiver)
     }
 
-    pub fn get_sender(&self, message_id: OutgoingMessages) -> Arc<Receiver<ResponseMessage>> {
-        let receiver = self.receivers.get(&message_id).expect("unsupport type");
-        Arc::clone(receiver)
+    pub fn get_sender(&self, message_id: IncomingMessages) -> Arc<Sender<ResponseMessage>> {
+        let sender = self
+            .senders
+            .get(&message_id)
+            .expect(&format!("unsupported response message {:?}", message_id));
+        Arc::clone(sender)
     }
-
 }
 
 #[derive(Debug)]
@@ -254,16 +270,6 @@ impl MessageBus for TcpMessageBus {
         Ok(subscription)
     }
 
-    fn request_next_order_id(&mut self, message: &RequestMessage) -> Result<BusSubscription, Error> {
-        self.write_message(message)?;
-
-        let subscription = SubscriptionBuilder::new()
-            .shared_receiver(Arc::clone(&self.globals.order_ids_out))
-            .build();
-
-        Ok(subscription)
-    }
-
     fn request_open_orders(&mut self, message: &RequestMessage) -> Result<BusSubscription, Error> {
         self.write_message(message)?;