Trade Engine Peer Message thru Taker

nobu-maeda · Nov 26, 2023 · f6a81f6 · f6a81f6
1 parent cb89827
commit f6a81f6
Show file tree

Hide file tree

Showing 2 changed files with 58 additions and 16 deletions.
diff --git a/src/order_sm/maker/maker.rs b/src/order_sm/maker/maker.rs
@@ -180,13 +180,13 @@ pub(super) enum MakerRequest {
     CancelOrder {
         rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
     },
-    TradeComplete {
-        rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
-    },
     PeerMessage {
         message: Box<dyn SerdeGenericTrait>,
         rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
     },
+    TradeComplete {
+        rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
+    },
     RegisterOfferNotifTx {
         tx: mpsc::Sender<Result<OfferEnvelope, N3xbError>>,
         rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
@@ -286,13 +286,13 @@ impl MakerActor {
                 self.cancel_order(rsp_tx).await;
                 terminate = true;
             }
+            MakerRequest::PeerMessage { message, rsp_tx } => {
+                self.send_peer_message(message, rsp_tx).await;
+            }
             MakerRequest::TradeComplete { rsp_tx } => {
                 self.trade_complete(rsp_tx).await;
                 terminate = true;
             }
-            MakerRequest::PeerMessage { message, rsp_tx } => {
-                self.send_peer_message(message, rsp_tx).await;
-            }
             MakerRequest::RegisterOfferNotifTx { tx, rsp_tx } => {
                 self.register_notif_tx(tx, rsp_tx).await;
             }
@@ -540,7 +540,7 @@ impl MakerActor {
             .communicator_accessor
             .send_trade_engine_specific_message(
                 pubkey,
-                Some(accepted_offer_event_id),
+                None,
                 maker_order_note_id,
                 self.order.trade_uuid.clone(),
                 message,

diff --git a/src/order_sm/taker/taker.rs b/src/order_sm/taker/taker.rs
@@ -42,8 +42,17 @@ impl TakerAccess<New> {
 }
 
 impl TakerAccess<Trading> {
-    pub async fn send_peer_message(&self, _content: Box<dyn SerdeGenericTrait>) {
-        todo!();
+    pub async fn send_peer_message(
+        &self,
+        _content: Box<dyn SerdeGenericTrait>,
+    ) -> Result<(), N3xbError> {
+        let (rsp_tx, rsp_rx) = oneshot::channel::<Result<(), N3xbError>>();
+        let request = TakerRequest::PeerMessage {
+            message: _content,
+            rsp_tx,
+        };
+        self.tx.send(request).await.unwrap();
+        rsp_rx.await.unwrap()
     }
 
     pub async fn trade_complete(self) -> Result<(), N3xbError> {
@@ -113,16 +122,20 @@ pub(super) enum TakerRequest {
     SendTakerOffer {
         rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
     },
+    PeerMessage {
+        message: Box<dyn SerdeGenericTrait>,
+        rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
+    },
+    TradeComplete {
+        rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
+    },
     RegisterTradeNotifTx {
         tx: mpsc::Sender<Result<TradeResponseEnvelope, N3xbError>>,
         rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
     },
     UnregisterTradeNotifTx {
         rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
     },
-    TradeComplete {
-        rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
-    },
 }
 
 struct TakerActor {
@@ -197,16 +210,19 @@ impl TakerActor {
 
         match request {
             TakerRequest::SendTakerOffer { rsp_tx } => self.send_taker_offer(rsp_tx).await,
+            TakerRequest::PeerMessage { message, rsp_tx } => {
+                self.send_peer_message(message, rsp_tx).await;
+            }
+            TakerRequest::TradeComplete { rsp_tx } => {
+                self.trade_complete(rsp_tx).await;
+                terminate = true;
+            }
             TakerRequest::RegisterTradeNotifTx { tx, rsp_tx } => {
                 self.register_trade_notif_tx(tx, rsp_tx).await
             }
             TakerRequest::UnregisterTradeNotifTx { rsp_tx } => {
                 self.unregister_trade_notif_tx(rsp_tx).await
             }
-            TakerRequest::TradeComplete { rsp_tx } => {
-                self.trade_complete(rsp_tx).await;
-                terminate = true;
-            }
         }
         terminate
     }
@@ -237,6 +253,32 @@ impl TakerActor {
         }
     }
 
+    async fn send_peer_message(
+        &mut self,
+        message: Box<dyn SerdeGenericTrait>,
+        rsp_tx: oneshot::Sender<Result<(), N3xbError>>,
+    ) {
+        let result = self
+            .communicator_accessor
+            .send_trade_engine_specific_message(
+                self.order_envelope.pubkey,
+                None,
+                self.order_envelope.event_id.clone(),
+                self.order_envelope.order.trade_uuid,
+                message,
+            )
+            .await;
+
+        match result {
+            Ok(_) => {
+                rsp_tx.send(Ok(())).unwrap(); // oneshot should not fail
+            }
+            Err(err) => {
+                rsp_tx.send(Err(err)).unwrap(); // oneshot should not fail
+            }
+        }
+    }
+
     async fn register_trade_notif_tx(
         &mut self,
         tx: mpsc::Sender<Result<TradeResponseEnvelope, N3xbError>>,