Merge #318

318: Introduce a `fan_out::Actor` instead of iterating in the `oracle::Actor` r=thomaseizinger a=thomaseizinger

A fan-out actor takes an incoming message and forwards it to a set
of actors. This allows our `oracle::Actor` to be unaware, how many
other actors are interested in the attestation, thus simplifying it.


Co-authored-by: Thomas Eizinger <thomas@eizinger.io>

daemon/src/fan_out.rs

@@ -0,0 +1,29 @@
+use xtra::prelude::MessageChannel;
+
+/// A fan-out actor takes every incoming message and forwards it to a set of other actors.
+pub struct Actor<M: xtra::Message<Result = ()>> {
+    receivers: Vec<Box<dyn MessageChannel<M>>>,
+}
+
+impl<M: xtra::Message<Result = ()>> Actor<M> {
+    pub fn new(receivers: &[&dyn MessageChannel<M>]) -> Self {
+        Self {
+            receivers: receivers.iter().map(|c| c.clone_channel()).collect(),
+        }
+    }
+}
+
+impl<M> xtra::Actor for Actor<M> where M: xtra::Message<Result = ()> {}
+
+#[async_trait::async_trait]
+impl<M> xtra::Handler<M> for Actor<M>
+where
+    M: xtra::Message<Result = ()> + Clone + Sync + 'static,
+{
+    async fn handle(&mut self, message: M, _: &mut xtra::Context<Self>) {
+        for receiver in &self.receivers {
+            // Not sure why here is no `do_send_async` ...
+            let _ = receiver.do_send(message.clone());
+        }
+    }
+}

daemon/src/lib.rs

@@ -2,6 +2,7 @@ pub mod actors;
 pub mod auth;
 pub mod bitmex_price_feed;
 pub mod db;
+pub mod fan_out;
 pub mod housekeeping;
 pub mod keypair;
 pub mod logger;

daemon/src/maker.rs

@@ -9,8 +9,8 @@ use daemon::model::WalletInfo;
 use daemon::seed::Seed;
 use daemon::wallet::Wallet;
 use daemon::{
-    bitmex_price_feed, housekeeping, logger, maker_cfd, maker_inc_connections, monitor, oracle,
-    wallet_sync,
+    bitmex_price_feed, fan_out, housekeeping, logger, maker_cfd, maker_inc_connections, monitor,
+    oracle, wallet_sync,
 };
 use rocket::fairing::AdHoc;
 use rocket_db_pools::Database;
@@ -259,13 +259,11 @@ async fn main() -> Result<()> {
                         .notify_interval(Duration::from_secs(60), || oracle::Sync)
                         .unwrap(),
                 );
-                tokio::spawn(oracle_actor_context.run(oracle::Actor::new(
-                    cfds,
-                    [
-                        Box::new(cfd_maker_actor_inbox.clone()),
-                        Box::new(monitor_actor_address),
-                    ],
-                )));
+                let actor = fan_out::Actor::new(&[&cfd_maker_actor_inbox, &monitor_actor_address])
+                    .create(None)
+                    .spawn_global();
+
+                tokio::spawn(oracle_actor_context.run(oracle::Actor::new(cfds, actor)));
 
                 oracle_actor_address
                     .do_send_async(oracle::Sync)

daemon/src/oracle.rs

@@ -15,7 +15,7 @@ pub struct Actor {
     announcements: HashMap<BitMexPriceEventId, (OffsetDateTime, Vec<schnorrsig::PublicKey>)>,
     pending_announcements: HashSet<BitMexPriceEventId>,
     pending_attestations: HashSet<BitMexPriceEventId>,
-    attestation_channels: [Box<dyn StrongMessageChannel<Attestation>>; 2],
+    attestation_channel: Box<dyn StrongMessageChannel<Attestation>>,
 }
 
 pub struct Sync;
@@ -67,7 +67,7 @@ struct NewAttestationFetched {
 impl Actor {
     pub fn new(
         cfds: Vec<Cfd>,
-        attestation_channels: [Box<dyn StrongMessageChannel<Attestation>>; 2],
+        attestation_channel: impl StrongMessageChannel<Attestation> + 'static,
     ) -> Self {
         let mut pending_attestations = HashSet::new();
 
@@ -101,7 +101,7 @@ impl Actor {
             announcements: HashMap::new(),
             pending_announcements: HashSet::new(),
             pending_attestations,
-            attestation_channels,
+            attestation_channel: Box::new(attestation_channel),
         }
     }
 }
@@ -192,10 +192,7 @@ impl Actor {
     ) -> Result<()> {
         tracing::info!("Fetched new attestation for {}", id);
 
-        for channel in self.attestation_channels.iter() {
-            channel.do_send(attestation.clone())?;
-        }
-
+        let _ = self.attestation_channel.send(attestation).await;
         self.pending_attestations.remove(&id);
 
         Ok(())

daemon/src/taker.rs

@@ -8,7 +8,7 @@ use daemon::model::WalletInfo;
 use daemon::seed::Seed;
 use daemon::wallet::Wallet;
 use daemon::{
-    bitmex_price_feed, housekeeping, logger, monitor, oracle, send_to_socket, taker_cfd,
+    bitmex_price_feed, fan_out, housekeeping, logger, monitor, oracle, send_to_socket, taker_cfd,
     wallet_sync, wire,
 };
 use futures::StreamExt;
@@ -253,13 +253,11 @@ async fn main() -> Result<()> {
                         .notify_interval(Duration::from_secs(60), || oracle::Sync)
                         .unwrap(),
                 );
-                tokio::spawn(oracle_actor_context.run(oracle::Actor::new(
-                    cfds,
-                    [
-                        Box::new(cfd_actor_inbox.clone()),
-                        Box::new(monitor_actor_address),
-                    ],
-                )));
+                let actor = fan_out::Actor::new(&[&cfd_actor_inbox, &monitor_actor_address])
+                    .create(None)
+                    .spawn_global();
+
+                tokio::spawn(oracle_actor_context.run(oracle::Actor::new(cfds, actor)));
 
                 oracle_actor_address
                     .do_send_async(oracle::Sync)