Merge #574

574: Use the same actor systems in production and testing r=thomaseizinger a=klochowicz

Removes the need to manually align test harness whenever actor systems get
expanded.

Soon actor systems will be storing remote handles as well, and they will not be
public.

Co-authored-by: Mariusz Klochowicz <mariusz@klochowicz.com>

daemon/tests/happy_path.rs

@@ -11,14 +11,12 @@ async fn taker_receives_order_from_maker_on_publication() {
     let _guard = init_tracing();
     let (mut maker, mut taker) = start_both().await;
 
-    assert!(is_next_none(&mut taker.order_feed).await);
+    assert!(is_next_none(taker.order_feed()).await);
 
     maker.publish_order(dummy_new_order()).await;
 
-    let (published, received) = tokio::join!(
+    let (published, received) =
-        next_some(&mut maker.order_feed),
+        tokio::join!(next_some(maker.order_feed()), next_some(taker.order_feed()));
-        next_some(&mut taker.order_feed)
-    );
 
     assert_is_same_order(&published, &received);
 }
@@ -29,15 +27,15 @@ async fn taker_takes_order_and_maker_rejects() {
     let (mut maker, mut taker) = start_both().await;
 
     // TODO: Why is this needed? For the cfd stream it is not needed
-    is_next_none(&mut taker.order_feed).await;
+    is_next_none(taker.order_feed()).await;
 
     maker.publish_order(dummy_new_order()).await;
 
-    let (_, received) = next_order(&mut maker.order_feed, &mut taker.order_feed).await;
+    let (_, received) = next_order(maker.order_feed(), taker.order_feed()).await;
 
     taker.take_order(received.clone(), Usd::new(dec!(10))).await;
 
-    let (taker_cfd, maker_cfd) = next_cfd(&mut taker.cfd_feed, &mut maker.cfd_feed).await;
+    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await;
     assert_is_same_order(&taker_cfd.order, &received);
     assert_is_same_order(&maker_cfd.order, &received);
     assert!(matches!(
@@ -51,7 +49,7 @@ async fn taker_takes_order_and_maker_rejects() {
 
     maker.reject_take_request(received.clone()).await;
 
-    let (taker_cfd, maker_cfd) = next_cfd(&mut taker.cfd_feed, &mut maker.cfd_feed).await;
+    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await;
     // TODO: More elaborate Cfd assertions
     assert_is_same_order(&taker_cfd.order, &received);
     assert_is_same_order(&maker_cfd.order, &received);
@@ -64,14 +62,14 @@ async fn taker_takes_order_and_maker_accepts_and_contract_setup() {
     let _guard = init_tracing();
     let (mut maker, mut taker) = start_both().await;
 
-    is_next_none(&mut taker.order_feed).await;
+    is_next_none(taker.order_feed()).await;
 
     maker.publish_order(dummy_new_order()).await;
 
-    let (_, received) = next_order(&mut maker.order_feed, &mut taker.order_feed).await;
+    let (_, received) = next_order(maker.order_feed(), taker.order_feed()).await;
 
     taker.take_order(received.clone(), Usd::new(dec!(5))).await;
-    let (_, _) = next_cfd(&mut taker.cfd_feed, &mut maker.cfd_feed).await;
+    let (_, _) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await;
 
     maker.mocks.mock_oracle_annoucement().await;
     taker.mocks.mock_oracle_annoucement().await;
@@ -81,7 +79,7 @@ async fn taker_takes_order_and_maker_accepts_and_contract_setup() {
 
     maker.accept_take_request(received.clone()).await;
 
-    let (taker_cfd, maker_cfd) = next_cfd(&mut taker.cfd_feed, &mut maker.cfd_feed).await;
+    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await;
     // TODO: More elaborate Cfd assertions
     assert_eq!(taker_cfd.order.id, received.id);
     assert_eq!(maker_cfd.order.id, received.id);
@@ -91,7 +89,7 @@ async fn taker_takes_order_and_maker_accepts_and_contract_setup() {
     maker.mocks.mock_wallet_sign_and_broadcast().await;
     taker.mocks.mock_wallet_sign_and_broadcast().await;
 
-    let (taker_cfd, maker_cfd) = next_cfd(&mut taker.cfd_feed, &mut maker.cfd_feed).await;
+    let (taker_cfd, maker_cfd) = next_cfd(taker.cfd_feed(), maker.cfd_feed()).await;
     // TODO: More elaborate Cfd assertions
     assert_eq!(taker_cfd.order.id, received.id);
     assert_eq!(maker_cfd.order.id, received.id);

daemon/tests/harness/mod.rs

@@ -2,12 +2,12 @@ use crate::harness::mocks::monitor::MonitorActor;
 use crate::harness::mocks::oracle::OracleActor;
 use crate::harness::mocks::wallet::WalletActor;
 use crate::schnorrsig;
-use daemon::connection::{Connect, ConnectionStatus};
+use daemon::connection::Connect;
 use daemon::maker_cfd::CfdAction;
 use daemon::model::cfd::{Cfd, Order, Origin};
 use daemon::model::{Price, Usd};
 use daemon::seed::Seed;
-use daemon::{db, maker_cfd, maker_inc_connections, taker_cfd};
+use daemon::{db, maker_cfd, maker_inc_connections, taker_cfd, MakerActorSystem};
 use rust_decimal_macros::dec;
 use sqlx::SqlitePool;
 use std::net::SocketAddr;
@@ -40,21 +40,23 @@ pub async fn start_both() -> (Maker, Taker) {
 const N_PAYOUTS_FOR_TEST: usize = 5;
 
 /// Maker Test Setup
-#[derive(Clone)]
 pub struct Maker {
-    pub cfd_actor_addr: xtra::Address<
+    pub system:
-        maker_cfd::Actor<OracleActor, MonitorActor, maker_inc_connections::Actor, WalletActor>,
+        MakerActorSystem<OracleActor, MonitorActor, maker_inc_connections::Actor, WalletActor>,
-    >,
-    pub order_feed: watch::Receiver<Option<Order>>,
-    pub cfd_feed: watch::Receiver<Vec<Cfd>>,
-    #[allow(dead_code)] // we need to keep the xtra::Address for refcounting
-    pub inc_conn_actor_addr: xtra::Address<maker_inc_connections::Actor>,
-    pub listen_addr: SocketAddr,
     pub mocks: mocks::Mocks,
+    pub listen_addr: SocketAddr,
     pub identity_pk: x25519_dalek::PublicKey,
 }
 
 impl Maker {
+    pub fn cfd_feed(&mut self) -> &mut watch::Receiver<Vec<Cfd>> {
+        &mut self.system.cfd_feed_receiver
+    }
+
+    pub fn order_feed(&mut self) -> &mut watch::Receiver<Option<Order>> {
+        &mut self.system.order_feed_receiver
+    }
+
     pub async fn start(oracle_pk: schnorrsig::PublicKey) -> Self {
         let db = in_memory_db().await;
 
@@ -101,18 +103,16 @@ impl Maker {
         tokio::spawn(maker.inc_conn_addr.clone().attach_stream(listener_stream));
 
         Self {
-            cfd_actor_addr: maker.cfd_actor_addr,
+            system: maker,
-            order_feed: maker.order_feed_receiver,
-            cfd_feed: maker.cfd_feed_receiver,
-            inc_conn_actor_addr: maker.inc_conn_addr,
-            listen_addr: address,
             identity_pk,
+            listen_addr: address,
             mocks,
         }
     }
 
     pub async fn publish_order(&mut self, new_order_params: maker_cfd::NewOrder) {
-        self.cfd_actor_addr
+        self.system
+            .cfd_actor_addr
             .send(new_order_params)
             .await
             .unwrap()
@@ -120,7 +120,8 @@ impl Maker {
     }
 
     pub async fn reject_take_request(&self, order: Order) {
-        self.cfd_actor_addr
+        self.system
+            .cfd_actor_addr
             .send(CfdAction::RejectOrder { order_id: order.id })
             .await
             .unwrap()
@@ -128,7 +129,8 @@ impl Maker {
     }
 
     pub async fn accept_take_request(&self, order: Order) {
-        self.cfd_actor_addr
+        self.system
+            .cfd_actor_addr
             .send(CfdAction::AcceptOrder { order_id: order.id })
             .await
             .unwrap()
@@ -137,16 +139,20 @@ impl Maker {
 }
 
 /// Taker Test Setup
-#[derive(Clone)]
 pub struct Taker {
-    pub order_feed: watch::Receiver<Option<Order>>,
+    pub system: daemon::TakerActorSystem<OracleActor, MonitorActor, WalletActor>,
-    pub cfd_feed: watch::Receiver<Vec<Cfd>>,
-    pub maker_status_feed: watch::Receiver<ConnectionStatus>,
-    pub cfd_actor_addr: xtra::Address<taker_cfd::Actor<OracleActor, MonitorActor, WalletActor>>,
     pub mocks: mocks::Mocks,
 }
 
 impl Taker {
+    pub fn cfd_feed(&mut self) -> &mut watch::Receiver<Vec<Cfd>> {
+        &mut self.system.cfd_feed_receiver
+    }
+
+    pub fn order_feed(&mut self) -> &mut watch::Receiver<Option<Order>> {
+        &mut self.system.order_feed_receiver
+    }
+
     pub async fn start(
         oracle_pk: schnorrsig::PublicKey,
         maker_address: SocketAddr,
@@ -187,16 +193,14 @@ impl Taker {
             .unwrap();
 
         Self {
-            order_feed: taker.order_feed_receiver,
+            system: taker,
-            cfd_feed: taker.cfd_feed_receiver,
-            maker_status_feed: taker.maker_online_status_feed_receiver,
-            cfd_actor_addr: taker.cfd_actor_addr,
             mocks,
         }
     }
 
     pub async fn take_order(&self, order: Order, quantity: Usd) {
-        self.cfd_actor_addr
+        self.system
+            .cfd_actor_addr
             .send(taker_cfd::TakeOffer {
                 order_id: order.id,
                 quantity,