install_programs: broadcast to all backends via new BackendBroadcaster

Author: fcangialosi

src/ipc/mod.rs

@@ -34,6 +34,7 @@ pub trait Ipc: 'static + Send {
 pub trait Backend<T: Ipc> {
     //fn new(sock: T, continue_listening: Arc<atomic::AtomicBool>) -> Self;
     fn sender(&self) -> BackendSender<T>;
+    fn broadcaster(&self) -> BackendBroadcaster<T>;
     fn clone_atomic_bool(&self) -> Arc<atomic::AtomicBool>;
     fn next(&mut self) -> Option<Msg>;
 }
@@ -149,7 +150,7 @@ where
         }
 
         MultiBackend {
-            last_recvd: Some(0), // TODO need a better solution for this, should be none
+            last_recvd: None,
             continue_listening,
             sel,
             backends,
@@ -174,10 +175,16 @@ impl<T: Ipc> Backend<T> for MultiBackend<T> {
     fn sender(&self) -> BackendSender<T> {
         match self.last_recvd {
             Some(i) => self.backends[i as usize].clone(),
-            None    => panic!("Called sender but no messages have been received yet!")
+            None => {
+                panic!("No messages have been received yet, so there is no corresponding sender")
+            }
         }
     }
 
+    fn broadcaster(&self) -> BackendBroadcaster<T> {
+        BackendBroadcaster(self.backends.clone())
+    }
+
     /// Return a copy of the flag variable that indicates that the
     /// `Backend` should continue listening (i.e., not exit).
     fn clone_atomic_bool(&self) -> Arc<atomic::AtomicBool> {
@@ -209,6 +216,10 @@ impl<T: Ipc> Backend<T> for SingleBackend<T> {
         BackendSender(Arc::downgrade(&self.sock))
     }
 
+    fn broadcaster(&self) -> BackendBroadcaster<T> {
+        BackendBroadcaster(vec![self.sender()])
+    }
+
     /// Return a copy of the flag variable that indicates that the
     /// `Backend` should continue listening (i.e., not exit).
     fn clone_atomic_bool(&self) -> Arc<atomic::AtomicBool> {

src/lib.rs

@@ -102,7 +102,7 @@ mod errors;
 pub use crate::errors::*;
 
 use crate::ipc::Ipc;
-use crate::ipc::{BackendBuilder, BackendSender};
+use crate::ipc::{BackendBroadcaster, BackendBuilder, BackendSender};
 use crate::lang::{Bin, Reg, Scope};
 use crate::serialize::Msg;
 
@@ -239,7 +239,7 @@ where
         instrs: bin,
     };
     let buf = serialize::serialize(&msg)?;
-    sender.send_msg(&buf[..])?;
+    sender.broadcast_msg(&buf[..])?;
     Ok(())
 }
 
@@ -474,12 +474,12 @@ where
 fn install_programs<I: Ipc + Sync>(
     programs: &HashMap<&'static str, String>,
     mut scope_map: &mut Rc<HashMap<String, Scope>>,
-    b: BackendSender<I>,
+    b: &BackendBroadcaster<I>,
 ) -> Result<()> {
     for (program_name, program) in programs.iter() {
         match lang::compile(program.as_bytes(), &[]) {
             Ok((bin, sc)) => {
-                match send_and_install(0, &b, bin, &sc) {
+                match send_and_install(0, b, bin, &sc) {
                     Ok(_) => {}
                     Err(e) => {
                         return Err(Error(format!(
@@ -539,7 +539,8 @@ where
     let mut scope_map = Rc::new(HashMap::<String, Scope>::default());
 
     let programs = alg.datapath_programs();
-    match install_programs(&programs, &mut scope_map, b.sender()) {
+    let broadcaster = b.broadcaster();
+    match install_programs(&programs, &mut scope_map, &broadcaster) {
         Ok(()) => {} // great, the datapath is ready, keep going!
         Err(_) => {
             if let Some(log) = cfg.logger.as_ref() {
@@ -561,7 +562,7 @@ where
                 info!(log, "got message from datapath, installing programs...");
             }
             // got a msg from the datapath, it must be up, so let's try to install programs again
-            install_programs(&programs, &mut scope_map, b.sender()).unwrap();
+            install_programs(&programs, &mut scope_map, &broadcaster).unwrap();
         }
     }
     if let Some(log) = cfg.logger.as_ref() {