Emit relay_list update events in the daemon subscription

5 files changed, 81 insertions, 27 deletions

diff --git a/mullvad-cli/src/cmds/status.rs b/mullvad-cli/src/cmds/status.rs
index fc6df014fc..862514a9f4 100644
--- a/mullvad-cli/src/cmds/status.rs
+++ b/mullvad-cli/src/cmds/status.rs
@@ -15,7 +15,13 @@ impl Command for Status {
         clap::SubCommand::with_name(self.name())
             .about("View the state of the VPN tunnel")
             .subcommand(
-                clap::SubCommand::with_name("listen").about("Listen for VPN tunnel state changes"),
+                clap::SubCommand::with_name("listen")
+                    .about("Listen for VPN tunnel state changes")
+                    .arg(
+                        clap::Arg::with_name("verbose")
+                            .short("v")
+                            .help("Enables verbose output"),
+                    ),
             )
     }
 
@@ -25,20 +31,31 @@ impl Command for Status {
 
         print_state(&state);
         print_location(&mut rpc)?;
-        if matches.subcommand_matches("listen").is_some() {
+        if let Some(listen_matches) = matches.subcommand_matches("listen") {
+            let verbose = listen_matches.is_present("verbose");
             let subscription = rpc
                 .daemon_event_subscribe()
                 .wait()
                 .map_err(|_err| Error::from(ErrorKind::CantSubscribe))?;
             for event in subscription.wait() {
-                if let DaemonEvent::StateTransition(new_state) =
-                    event.chain_err(|| "Subscription failed")?
-                {
-                    print_state(&new_state);
-                    use self::TunnelStateTransition::*;
-                    match new_state {
-                        Connected(_) | Disconnected => print_location(&mut rpc)?,
-                        _ => {}
+                match event.chain_err(|| "Subscription failed")? {
+                    DaemonEvent::StateTransition(new_state) => {
+                        print_state(&new_state);
+                        use self::TunnelStateTransition::*;
+                        match new_state {
+                            Connected(_) | Disconnected => print_location(&mut rpc)?,
+                            _ => {}
+                        }
+                    }
+                    DaemonEvent::Settings(settings) => {
+                        if verbose {
+                            println!("New settings: {:#?}", settings);
+                        }
+                    }
+                    DaemonEvent::RelayList(relay_list) => {
+                        if verbose {
+                            println!("New relay list: {:#?}", relay_list);
+                        }
                     }
                 }
             }
diff --git a/mullvad-daemon/src/lib.rs b/mullvad-daemon/src/lib.rs
index 455c362df3..55bfce8f62 100644
--- a/mullvad-daemon/src/lib.rs
+++ b/mullvad-daemon/src/lib.rs
@@ -198,15 +198,28 @@ impl Daemon {
         let rpc_handle = rpc_handle.chain_err(|| "Unable to create RPC client")?;
         let https_handle = https_handle.chain_err(|| "Unable to create am.i.mullvad client")?;
 
-        let relay_selector =
-            relays::RelaySelector::new(rpc_handle.clone(), &resource_dir, &cache_dir);
+        let (internal_event_tx, internal_event_rx) = mpsc::channel();
+
+        let management_interface_result =
+            Self::start_management_interface(internal_event_tx.clone())?;
+
+        let management_interface_broadcaster = management_interface_result.0.clone();
+        let on_relay_list_update = move |relay_list: &RelayList| {
+            management_interface_broadcaster.notify_relay_list(relay_list.clone());
+        };
+        let relay_selector = relays::RelaySelector::new(
+            rpc_handle.clone(),
+            on_relay_list_update,
+            &resource_dir,
+            &cache_dir,
+        );
         let settings = Settings::load().chain_err(|| "Unable to read settings")?;
         let account_history = account_history::AccountHistory::new(&cache_dir)
             .chain_err(|| "Unable to read wireguard key cache")?;
 
-
-        let (tx, rx) = mpsc::channel();
-        let tunnel_parameters_generator = MullvadTunnelParametersGenerator { tx: tx.clone() };
+        let tunnel_parameters_generator = MullvadTunnelParametersGenerator {
+            tx: internal_event_tx.clone(),
+        };
         let tunnel_command_tx = tunnel_state_machine::spawn(
             settings.get_allow_lan(),
             settings.get_block_when_disconnected(),
@@ -214,22 +227,19 @@ impl Daemon {
             log_dir,
             resource_dir,
             cache_dir.clone(),
-            IntoSender::from(tx.clone()),
+            IntoSender::from(internal_event_tx.clone()),
         )?;
 
-        let target_state = TargetState::Unsecured;
-        let management_interface_result = Self::start_management_interface(tx.clone())?;
-
         // Attempt to download a fresh relay list
         relay_selector.update();
 
         Ok(Daemon {
             tunnel_command_tx: Sink::wait(tunnel_command_tx),
             tunnel_state: TunnelStateTransition::Disconnected,
-            target_state,
+            target_state: TargetState::Unsecured,
             state: DaemonExecutionState::Running,
-            rx,
-            tx,
+            rx: internal_event_rx,
+            tx: internal_event_tx,
             reconnection_loop_tx: None,
             management_interface_broadcaster: management_interface_result.0,
             #[cfg(unix)]
diff --git a/mullvad-daemon/src/management_interface.rs b/mullvad-daemon/src/management_interface.rs
index 17da57f37a..a6bc12f8b8 100644
--- a/mullvad-daemon/src/management_interface.rs
+++ b/mullvad-daemon/src/management_interface.rs
@@ -271,6 +271,7 @@ impl ManagementInterfaceServer {
 }
 
 /// A handle that allows broadcasting messages to all subscribers of the management interface.
+#[derive(Clone)]
 pub struct EventBroadcaster {
     subscriptions: Arc<RwLock<HashMap<SubscriptionId, pubsub::Sink<DaemonEvent>>>>,
 }
@@ -288,6 +289,12 @@ impl EventBroadcaster {
         self.notify(DaemonEvent::Settings(settings));
     }
 
+    /// Sends settings to all `settings` subscribers of the management interface.
+    pub fn notify_relay_list(&self, relay_list: RelayList) {
+        log::debug!("Broadcasting new relay list");
+        self.notify(DaemonEvent::RelayList(relay_list));
+    }
+
     fn notify(&self, value: DaemonEvent) {
         let subscriptions = self.subscriptions.read().unwrap();
         for sink in subscriptions.values() {
diff --git a/mullvad-daemon/src/relays.rs b/mullvad-daemon/src/relays.rs
index a6ba4cb0af..7725477885 100644
--- a/mullvad-daemon/src/relays.rs
+++ b/mullvad-daemon/src/relays.rs
@@ -147,7 +147,12 @@ pub struct RelaySelector {
 impl RelaySelector {
     /// Returns a new `RelaySelector` backed by relays cached on disk. Use the `update` method
     /// to refresh the relay list from the internet.
-    pub fn new(rpc_handle: HttpHandle, resource_dir: &Path, cache_dir: &Path) -> Self {
+    pub fn new(
+        rpc_handle: HttpHandle,
+        on_update: impl Fn(&RelayList) + Send + 'static,
+        resource_dir: &Path,
+        cache_dir: &Path,
+    ) -> Self {
         let cache_path = cache_dir.join(RELAYS_FILENAME);
         let resource_path = resource_dir.join(RELAYS_FILENAME);
         let unsynchronized_parsed_relays = Self::read_relays_from_disk(&cache_path, &resource_path)
@@ -163,7 +168,12 @@ impl RelaySelector {
                 .format(DATE_TIME_FORMAT_STR)
         );
         let parsed_relays = Arc::new(Mutex::new(unsynchronized_parsed_relays));
-        let updater = RelayListUpdater::spawn(rpc_handle, cache_path, parsed_relays.clone());
+        let updater = RelayListUpdater::spawn(
+            rpc_handle,
+            cache_path,
+            parsed_relays.clone(),
+            Box::new(on_update),
+        );
         RelaySelector {
             parsed_relays,
             rng: rand::thread_rng(),
@@ -498,6 +508,7 @@ struct RelayListUpdater {
     rpc_client: RelayListProxy<HttpHandle>,
     cache_path: PathBuf,
     parsed_relays: Arc<Mutex<ParsedRelays>>,
+    on_update: Box<dyn Fn(&RelayList)>,
     close_handle: mpsc::Receiver<()>,
 }
 
@@ -506,10 +517,13 @@ impl RelayListUpdater {
         rpc_handle: HttpHandle,
         cache_path: PathBuf,
         parsed_relays: Arc<Mutex<ParsedRelays>>,
+        on_update: Box<dyn Fn(&RelayList) + Send + 'static>,
     ) -> RelayListUpdaterHandle {
         let (tx, rx) = mpsc::channel();
 
-        thread::spawn(move || Self::new(rpc_handle, cache_path, parsed_relays, rx).run());
+        thread::spawn(move || {
+            Self::new(rpc_handle, cache_path, parsed_relays, on_update, rx).run()
+        });
 
         tx
     }
@@ -518,6 +532,7 @@ impl RelayListUpdater {
         rpc_handle: HttpHandle,
         cache_path: PathBuf,
         parsed_relays: Arc<Mutex<ParsedRelays>>,
+        on_update: Box<dyn Fn(&RelayList)>,
         close_handle: mpsc::Receiver<()>,
     ) -> Self {
         let rpc_client = RelayListProxy::new(rpc_handle);
@@ -526,6 +541,7 @@ impl RelayListUpdater {
             rpc_client,
             cache_path,
             parsed_relays,
+            on_update,
             close_handle,
         }
     }
@@ -582,8 +598,9 @@ impl RelayListUpdater {
             new_parsed_relays.relays().len()
         );
 
-        *self.lock_parsed_relays() = new_parsed_relays;
-
+        let mut parsed_relays = self.lock_parsed_relays();
+        *parsed_relays = new_parsed_relays;
+        (self.on_update)(parsed_relays.locations());
         Ok(())
     }
 
diff --git a/mullvad-types/src/lib.rs b/mullvad-types/src/lib.rs
index 2ac2295982..ad83e0276d 100644
--- a/mullvad-types/src/lib.rs
+++ b/mullvad-types/src/lib.rs
@@ -32,4 +32,7 @@ pub enum DaemonEvent {
 
     /// The daemon settings changed.
     Settings(settings::Settings),
+
+    /// The daemon got an updated relay list.
+    RelayList(relay_list::RelayList),
 }