Merge branch 'master-new-daemon'

8 files changed, 225 insertions, 230 deletions

diff --git a/talpid_core/Cargo.toml b/talpid_core/Cargo.toml
index 18c998cf93..573cc0b328 100644
--- a/talpid_core/Cargo.toml
+++ b/talpid_core/Cargo.toml
@@ -5,11 +5,15 @@ authors = ["Linus Färnstrand <linus@mullvad.net>", "Erik Larkö <erik@mullvad.n
 description = "Core backend functionality of the Mullvad VPN client"
 
 [dependencies]
-duct = "0.8"
+duct = "0.9.1"
 error-chain = "0.10"
 log = "0.3"
-jsonrpc-core = { git = "https://github.com/faern/jsonrpc", branch = "bind-zero" }
-jsonrpc-macros = { git = "https://github.com/faern/jsonrpc", branch = "bind-zero" }
+lazy_static = "0.2"
+jsonrpc-core = { git = "https://github.com/faern/jsonrpc", branch = "ws-close-handle" }
+jsonrpc-macros = { git = "https://github.com/faern/jsonrpc", branch = "ws-close-handle" }
+
+[target.'cfg(unix)'.dependencies]
+libc = "0.2.20"
 
 [dependencies.talpid_ipc]
 path = "../talpid_ipc"
diff --git a/talpid_core/src/lib.rs b/talpid_core/src/lib.rs
index 8a053bffa8..5496d038ce 100644
--- a/talpid_core/src/lib.rs
+++ b/talpid_core/src/lib.rs
@@ -9,6 +9,8 @@ extern crate assert_matches;
 extern crate duct;
 
 #[macro_use]
+extern crate lazy_static;
+#[macro_use]
 extern crate log;
 
 #[macro_use]
diff --git a/talpid_core/src/process/mod.rs b/talpid_core/src/process/mod.rs
index a7d76ea9f4..88bdc12ea4 100644
--- a/talpid_core/src/process/mod.rs
+++ b/talpid_core/src/process/mod.rs
@@ -1,5 +1,6 @@
-/// A module for monitoring child processes and get notified of events on them.
-pub mod monitor;
-
 /// A module for all OpenVPN related process management.
 pub mod openvpn;
+
+/// Unix specific process management features.
+#[cfg(unix)]
+pub mod unix;
diff --git a/talpid_core/src/process/monitor.rs b/talpid_core/src/process/monitor.rs
deleted file mode 100644
index 1568d80615..0000000000
--- a/talpid_core/src/process/monitor.rs
+++ /dev/null
@@ -1,113 +0,0 @@
-use duct;
-
-use std::io;
-use std::process;
-use std::sync::Arc;
-use std::thread;
-
-/// A child process monitor. Takes care of starting and monitoring a child process and calls the
-/// listener on child exit. If the child is still running when a `ChildMonitor` instance goes out
-/// of scope it will kill the child and wait for it to exit properly.
-pub struct ChildMonitor {
-    child: Arc<duct::Handle>,
-    thread: Option<thread::JoinHandle<()>>,
-}
-
-impl ChildMonitor {
-    /// Starts the child process and begins to monitor it. `on_exit` will be called as soon as the
-    /// child process exits.
-    pub fn start<L>(expression: &duct::Expression, mut on_exit: L) -> io::Result<Self>
-        where L: FnMut(io::Result<&process::Output>) + Send + 'static
-    {
-        let child = Arc::new(expression.start()?);
-        let child_clone = child.clone();
-        let thread = Some(thread::spawn(move || on_exit(child_clone.wait())));
-        Ok(ChildMonitor { child, thread })
-    }
-
-    /// Wait for the child to exit. Blocking the current thread. The `on_exit` callback is
-    /// guaranteed to fire before this method returns.
-    pub fn wait(&mut self) -> io::Result<&process::Output> {
-        if let Some(thread) = self.thread.take() {
-            if let Err(e) = thread.join() {
-                error!("Panic in the on_exit callback in ChildMonitor: {:?}", e);
-            }
-        }
-        self.child.wait()
-    }
-
-    /// Send a kill signal to the child. No need to call `wait` after to free the PID. The monitor
-    /// will wait for the process for you.
-    pub fn kill(&self) -> io::Result<()> {
-        self.child.kill()
-    }
-}
-
-impl Drop for ChildMonitor {
-    fn drop(&mut self) {
-        let _ = self.kill();
-        let _ = self.child.wait();
-    }
-}
-
-
-#[cfg(test)]
-mod child_monitor_tests {
-    use super::*;
-    use duct::{Expression, cmd};
-
-    use std::sync::mpsc;
-    use std::time::Duration;
-
-    fn echo_cmd(s: &str) -> Expression {
-        cmd("echo", &[s]).stdout_capture().unchecked()
-    }
-
-    fn invalid_cmd() -> Expression {
-        cmd("this command does not exist", &[""]).unchecked()
-    }
-
-    fn sleep_cmd(secs: u32) -> Expression {
-        if cfg!(windows) {
-            cmd("ping", &["127.0.0.1", "-n", &(secs + 1).to_string()]).unchecked()
-        } else {
-            cmd("sleep", &[secs.to_string()]).unchecked()
-        }
-    }
-
-    fn spawn(cmd: &Expression) -> (ChildMonitor, mpsc::Receiver<io::Result<process::Output>>) {
-        let (tx, rx) = mpsc::channel();
-        let child =
-            ChildMonitor::start(cmd, move |res| tx.send(res.map(|out| out.clone())).unwrap())
-                .expect("Unable to start process");
-        (child, rx)
-    }
-
-    #[test]
-    fn echo() {
-        let (mut child, rx) = spawn(&echo_cmd("foobar"));
-        let wait_output = child.wait().unwrap();
-        let callback_output = rx.try_recv().unwrap().unwrap();
-
-        assert!(callback_output.status.success());
-        assert_eq!("foobar\n".as_bytes(), &callback_output.stdout[..]);
-        assert_eq!(wait_output.status, callback_output.status);
-        assert_eq!(wait_output.stdout, callback_output.stdout);
-    }
-
-    #[test]
-    fn invalid_command() {
-        assert!(ChildMonitor::start(&invalid_cmd(), |_| {}).is_err());
-    }
-
-    #[test]
-    fn callback_after_kill() {
-        let (child, rx) = spawn(&sleep_cmd(100000));
-        // Make sure on_exit is not triggered within the first second. It should not be called
-        // until we kill the process.
-        assert!(rx.recv_timeout(Duration::from_secs(1)).is_err());
-
-        child.kill().unwrap();
-        assert!(!rx.recv_timeout(Duration::from_secs(10)).unwrap().unwrap().status.success());
-    }
-}
diff --git a/talpid_core/src/process/openvpn.rs b/talpid_core/src/process/openvpn.rs
index df1e4c7682..97c88a99c0 100644
--- a/talpid_core/src/process/openvpn.rs
+++ b/talpid_core/src/process/openvpn.rs
@@ -54,7 +54,7 @@ impl OpenVpnCommand {
     /// Build a runnable expression from the current state of the command.
     pub fn build(&self) -> duct::Expression {
         debug!("Building expression: {}", &self);
-        duct::cmd(&self.openvpn_bin, self.get_arguments())
+        duct::cmd(&self.openvpn_bin, self.get_arguments()).unchecked()
     }
 
     /// Returns all arguments that the subprocess would be spawned with.
diff --git a/talpid_core/src/process/unix.rs b/talpid_core/src/process/unix.rs
new file mode 100644
index 0000000000..64599a6930
--- /dev/null
+++ b/talpid_core/src/process/unix.rs
@@ -0,0 +1,33 @@
+extern crate libc;
+
+use duct;
+use duct::unix::HandleExt;
+
+use std::io;
+use std::sync::{Arc, mpsc};
+use std::thread;
+use std::time::Duration;
+
+/// Kills a process by first sending it the `SIGTERM` signal and then wait up to `timeout`. If the
+/// process has not died after the timeout has expired it is killed.
+pub fn nice_kill(handle: Arc<duct::Handle>, timeout: Duration) -> io::Result<()> {
+    trace!("Sending SIGTERM to child process");
+    handle.send_signal(libc::SIGTERM)?;
+
+    if wait_timeout(handle.clone(), timeout) {
+        debug!("Child process exited from SIGTERM");
+        Ok(())
+    } else {
+        debug!("Child process did not exit from SIGTERM, sending SIGKILL");
+        handle.kill()
+    }
+}
+
+/// Wait for a process to die for a maximum of `timeout`. Returns true if the process died within
+/// the timeout. Warning, if the process does not exit in the given time, this function will leave
+/// a thread running until it does exit.
+fn wait_timeout(handle: Arc<duct::Handle>, timeout: Duration) -> bool {
+    let (stop, stopped) = mpsc::channel();
+    thread::spawn(move || { let _ = stop.send(handle.wait().is_ok()); });
+    stopped.recv_timeout(timeout).unwrap_or(false)
+}
diff --git a/talpid_core/src/tunnel/mod.rs b/talpid_core/src/tunnel/mod.rs
index c453c52056..aead09b13b 100644
--- a/talpid_core/src/tunnel/mod.rs
+++ b/talpid_core/src/tunnel/mod.rs
@@ -1,11 +1,12 @@
 use net;
 use openvpn_ffi::OpenVpnPluginEvent;
 use process::openvpn::OpenVpnCommand;
+use std::io;
 
 /// A module for all OpenVPN related tunnel management.
 pub mod openvpn;
 
-use self::openvpn::{OpenVpnEvent, OpenVpnMonitor};
+use self::openvpn::{OpenVpnCloseHandle, OpenVpnMonitor};
 
 mod errors {
     error_chain!{
@@ -14,10 +15,6 @@ mod errors {
             TunnelMonitoringError {
                 description("Error while setting up or processing events from the VPN tunnel")
             }
-            /// An error indicating that there was an error when trying to start up a VPN tunnel.
-            TunnelStartError {
-                description("Error while trying to start the tunnel")
-            }
         }
     }
 }
@@ -25,26 +22,18 @@ pub use self::errors::*;
 
 
 /// Possible events from the VPN tunnel and the child process managing it.
+#[derive(Debug)]
 pub enum TunnelEvent {
     /// Sent when the tunnel comes up and is ready for traffic.
     Up,
     /// Sent when the tunnel goes down.
     Down,
-    /// Sent when the process managing the tunnel exits.
-    Shutdown,
 }
 
 impl TunnelEvent {
-    /// Converts an `OpenVpnEvent` to a `TunnelEvent`.
+    /// Converts an `OpenVpnPluginEvent` to a `TunnelEvent`.
     /// Returns `None` if there is no corresponding `TunnelEvent`.
-    pub fn from_openvpn_event(event: &OpenVpnEvent) -> Option<TunnelEvent> {
-        match *event {
-            OpenVpnEvent::PluginEvent(ref event, _) => Self::from_openvpn_plugin_event(event),
-            OpenVpnEvent::Shutdown(_) => Some(TunnelEvent::Shutdown),
-        }
-    }
-
-    fn from_openvpn_plugin_event(event: &OpenVpnPluginEvent) -> Option<TunnelEvent> {
+    fn from_openvpn_event(event: &OpenVpnPluginEvent) -> Option<TunnelEvent> {
         match *event {
             OpenVpnPluginEvent::Up => Some(TunnelEvent::Up),
             OpenVpnPluginEvent::RoutePredown => Some(TunnelEvent::Down),
@@ -60,28 +49,49 @@ pub struct TunnelMonitor {
 }
 
 impl TunnelMonitor {
-    /// Creates a new `TunnelMonitor` with the given event callback.
-    pub fn new<L>(on_event: L) -> Result<Self>
+    /// Creates a new `TunnelMonitor` that connects to the given remote and notifies `on_event`
+    /// on tunnel state changes.
+    pub fn new<L>(remote: net::RemoteAddr, on_event: L) -> Result<Self>
         where L: Fn(TunnelEvent) + Send + Sync + 'static
     {
-        let on_openvpn_event = move |openvpn_event| {
-            // FIXME: This comment must be here to make rustfmt 0.8.3 not screw up.
-            match TunnelEvent::from_openvpn_event(&openvpn_event) {
-                Some(tunnel_event) => on_event(tunnel_event),
-                None => debug!("Ignoring OpenVpnEvent {:?}", openvpn_event),
-            }
+        let on_openvpn_event = move |event, _env| match TunnelEvent::from_openvpn_event(&event) {
+            Some(tunnel_event) => on_event(tunnel_event),
+            None => debug!("Ignoring OpenVpnEvent {:?}", event),
         };
-        let monitor = openvpn::OpenVpnMonitor::new(on_openvpn_event, get_plugin_path())
+        let cmd = Self::create_openvpn_cmd(remote);
+        let monitor = openvpn::OpenVpnMonitor::new(cmd, on_openvpn_event, get_plugin_path())
             .chain_err(|| ErrorKind::TunnelMonitoringError)?;
         Ok(TunnelMonitor { monitor })
     }
 
-    /// Tries to start a VPN tunnel towards the given address. Will fail if there is a tunnel
-    /// running already.
-    pub fn start(&self, remote: net::RemoteAddr) -> Result<()> {
+    fn create_openvpn_cmd(remote: net::RemoteAddr) -> OpenVpnCommand {
         let mut cmd = OpenVpnCommand::new("openvpn");
-        cmd.config(get_config_path()).remotes(remote).unwrap();
-        self.monitor.start(cmd).chain_err(|| ErrorKind::TunnelStartError)
+        cmd.config(get_config_path())
+            .remotes(remote)
+            .unwrap();
+        cmd
+    }
+
+    /// Creates a handle to this monitor, allowing the tunnel to be closed while some other thread
+    /// is blocked in `wait`.
+    pub fn close_handle(&self) -> CloseHandle {
+        CloseHandle(self.monitor.close_handle())
+    }
+
+    /// Consumes the monitor and block until the tunnel exits or there is an error.
+    pub fn wait(self) -> Result<()> {
+        self.monitor.wait().chain_err(|| ErrorKind::TunnelMonitoringError)
+    }
+}
+
+
+/// A handle to a `TunnelMonitor`
+pub struct CloseHandle(OpenVpnCloseHandle);
+
+impl CloseHandle {
+    /// Closes the underlying tunnel, making the `TunnelMonitor::wait` method return.
+    pub fn close(self) -> io::Result<()> {
+        self.0.close()
     }
 }
 
diff --git a/talpid_core/src/tunnel/openvpn.rs b/talpid_core/src/tunnel/openvpn.rs
index 9d366ebb4a..08c2c14a32 100644
--- a/talpid_core/src/tunnel/openvpn.rs
+++ b/talpid_core/src/tunnel/openvpn.rs
@@ -1,30 +1,29 @@
+use duct;
 use jsonrpc_core::{Error, IoHandler};
-use openvpn_ffi;
-use process::monitor::ChildMonitor;
+use openvpn_ffi::{OpenVpnEnv, OpenVpnPluginEvent};
 use process::openvpn::OpenVpnCommand;
-use std::io;
 
-use std::path::{Path, PathBuf};
-use std::process;
+use std::io;
+use std::path::Path;
 use std::result::Result as StdResult;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, mpsc};
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::thread;
+use std::time::Duration;
 
 use talpid_ipc;
 
 mod errors {
     error_chain!{
         errors {
-            /// The `OpenVpnMonitor` is in an invalid state for the requested operation.
-            InvalidState {
-                description("Invalid state. OpenVPN is already running")
+            /// Unable to start, wait for or kill the OpenVPN process.
+            ChildProcessError(msg: &'static str) {
+                description("Unable to start, wait for or kill the OpenVPN process")
+                display("OpenVPN process error: {}", msg)
             }
-            /// Unable to start or kill the OpenVPN process.
-            ChildProcessError {
-                description("Unable to start or kill the OpenVPN process")
-            }
-            /// Unable to start or manage the IPC server listening for events from OpenVPN
-            IpcServerError {
-                description("Unable to start or manage the IPC server")
+            /// Unable to start or manage the IPC server listening for events from OpenVPN.
+            EventDispatcherError {
+                description("Unable to start or manage the event dispatcher IPC server")
             }
         }
     }
@@ -32,89 +31,145 @@ mod errors {
 pub use self::errors::*;
 
 
-/// Possible events from OpenVPN
-#[derive(Debug)]
-pub enum OpenVpnEvent {
-    /// An event from the plugin loaded into OpenVPN.
-    PluginEvent(openvpn_ffi::OpenVpnPluginEvent, openvpn_ffi::OpenVpnEnv),
-    /// The OpenVPN process exited. Containing the result of waiting for the process.
-    Shutdown(io::Result<process::ExitStatus>),
+lazy_static!{
+    static ref OPENVPN_DIE_TIMEOUT: Duration = Duration::from_secs(2);
 }
 
-/// Struct for monitoring OpenVPN processes.
+
+/// Struct for monitoring an OpenVPN process.
 pub struct OpenVpnMonitor {
-    on_event: Arc<Fn(OpenVpnEvent) + Send + Sync + 'static>,
-    plugin_path: PathBuf,
-    child: Arc<Mutex<Option<ChildMonitor>>>,
-    event_dispatcher: OpenVpnEventDispatcher,
+    child: Arc<duct::Handle>,
+    event_dispatcher: Option<OpenVpnEventDispatcher>,
+    closed: Arc<AtomicBool>,
 }
 
 impl OpenVpnMonitor {
     /// Creates a new `OpenVpnMonitor` with the given listener and using the plugin at the given
     /// path.
-    pub fn new<L, P>(on_event: L, plugin_path: P) -> Result<Self>
-        where L: Fn(OpenVpnEvent) + Send + Sync + 'static,
+    pub fn new<L, P>(mut cmd: OpenVpnCommand, on_event: L, plugin_path: P) -> Result<Self>
+        where L: Fn(OpenVpnPluginEvent, OpenVpnEnv) + Send + Sync + 'static,
               P: AsRef<Path>
     {
-        let on_event = Arc::new(on_event);
-        let event_dispatcher = Self::start_event_dispatcher(on_event.clone())?;
+        let event_dispatcher = OpenVpnEventDispatcher::start(on_event)
+            .chain_err(|| ErrorKind::EventDispatcherError)?;
+
+        cmd.plugin(plugin_path, vec![event_dispatcher.address().to_owned()]);
+        let child = cmd.build()
+            .start()
+            .chain_err(|| ErrorKind::ChildProcessError("Failed to start"))?;
+
         Ok(
             OpenVpnMonitor {
-                on_event,
-                plugin_path: plugin_path.as_ref().to_owned(),
-                child: Arc::new(Mutex::new(None)),
-                event_dispatcher,
+                child: Arc::new(child),
+                event_dispatcher: Some(event_dispatcher),
+                closed: Arc::new(AtomicBool::new(false)),
             },
         )
     }
 
-    fn start_event_dispatcher(on_event: Arc<Fn(OpenVpnEvent) + Send + Sync + 'static>)
-                              -> Result<OpenVpnEventDispatcher> {
-        let on_plugin_event = move |event, env| (*on_event)(OpenVpnEvent::PluginEvent(event, env));
-        OpenVpnEventDispatcher::start(on_plugin_event).chain_err(|| ErrorKind::IpcServerError)
+    /// Creates a handle to this monitor, allowing the tunnel to be closed while some other thread
+    /// is blocked in `wait`.
+    pub fn close_handle(&self) -> OpenVpnCloseHandle {
+        OpenVpnCloseHandle {
+            child: self.child.clone(),
+            closed: self.closed.clone(),
+        }
     }
 
-    /// Tries to start a new OpenVPN process if one is not already running.
-    /// If this `OpenVpnMonitor is already monitoring a running process it will return an
-    /// `InvalidState` error.
-    pub fn start(&self, cmd: OpenVpnCommand) -> Result<()> {
-        let mut child_lock = self.child.lock().unwrap();
-        if child_lock.is_some() {
-            bail!(ErrorKind::InvalidState);
+    /// Consumes the monitor and blocks until OpenVPN exits or there is an error in either waiting
+    /// for the process or in the event dispatcher.
+    pub fn wait(mut self) -> Result<()> {
+        match self.wait_result() {
+            WaitResult::Child(Ok(exit_status)) => {
+                if exit_status.success() || self.closed.load(Ordering::SeqCst) {
+                    debug!(
+                        "OpenVPN exited, as expected, with exit status: {}",
+                        exit_status
+                    );
+                    Ok(())
+                } else {
+                    error!("OpenVPN died unexpectedly with status: {}", exit_status);
+                    Err(ErrorKind::ChildProcessError("Died unexpectedly").into())
+                }
+            }
+            WaitResult::Child(Err(e)) => {
+                error!("OpenVPN process wait error: {}", e);
+                Err(e).chain_err(|| ErrorKind::ChildProcessError("Error when waiting"))
+            }
+            WaitResult::EventDispatcher(result) => {
+                error!("OpenVpnEventDispatcher exited unexpectedly: {:?}", result);
+                match result {
+                    Ok(()) => Err(ErrorKind::EventDispatcherError.into()),
+                    Err(e) => Err(e).chain_err(|| ErrorKind::EventDispatcherError),
+                }
+            }
         }
-        *child_lock = Some(self.start_child_monitor(cmd)?);
-        Ok(())
     }
 
-    fn start_child_monitor(&self, mut cmd: OpenVpnCommand) -> Result<ChildMonitor> {
-        self.set_plugin(&mut cmd);
+    /// Waits for both the child process and the event dispatcher in parallel. After both have
+    /// returned this returns the earliest result.
+    fn wait_result(&mut self) -> WaitResult {
+        let child_wait_handle = self.child.clone();
+        let child_close_handle = self.close_handle();
+        let event_dispatcher = self.event_dispatcher.take().unwrap();
+        let dispatcher_handle = event_dispatcher.close_handle();
 
-        let child = self.child.clone();
-        let on_event = self.on_event.clone();
+        let (child_tx, rx) = mpsc::channel();
+        let dispatcher_tx = child_tx.clone();
 
-        let on_exit = move |exit_status: io::Result<&process::Output>| {
-            *child.lock().unwrap() = None;
-            (*on_event)(OpenVpnEvent::Shutdown(exit_status.map(|output| output.status)),)
-        };
-        ChildMonitor::start(&cmd.build(), on_exit).chain_err(|| ErrorKind::ChildProcessError)
-    }
+        thread::spawn(
+            move || {
+                let result = child_wait_handle.wait().map(|output| output.status);
+                child_tx.send(WaitResult::Child(result)).unwrap();
+                dispatcher_handle.close();
+            },
+        );
+        thread::spawn(
+            move || {
+                let result = event_dispatcher.wait();
+                dispatcher_tx.send(WaitResult::EventDispatcher(result)).unwrap();
+                let _ = child_close_handle.close();
+            },
+        );
 
-    fn set_plugin(&self, cmd: &mut OpenVpnCommand) {
-        let event_dispatcher_address = self.event_dispatcher.address().to_string();
-        cmd.plugin(&self.plugin_path, vec![event_dispatcher_address]);
+        let result = rx.recv().unwrap();
+        let _ = rx.recv().unwrap();
+        result
     }
+}
+
+/// A handle to an `OpenVpnMonitor` for closing it.
+pub struct OpenVpnCloseHandle {
+    child: Arc<duct::Handle>,
+    closed: Arc<AtomicBool>,
+}
 
-    /// Tries to kill the OpenVPN process if it is running. If it is already dead, this does
-    /// nothing.
-    pub fn kill(&self) -> Result<()> {
-        if let Some(ref child) = *self.child.lock().unwrap() {
-            child.kill().chain_err(|| ErrorKind::ChildProcessError)?;
+impl OpenVpnCloseHandle {
+    /// Kills the underlying OpenVPN process, making the `OpenVpnMonitor::wait` method return.
+    pub fn close(self) -> io::Result<()> {
+        if !self.closed.swap(true, Ordering::SeqCst) {
+            self.kill_openvpn()
+        } else {
+            Ok(())
         }
-        Ok(())
     }
-}
 
+    #[cfg(unix)]
+    fn kill_openvpn(self) -> io::Result<()> {
+        ::process::unix::nice_kill(self.child, *OPENVPN_DIE_TIMEOUT)
+    }
 
+    #[cfg(not(unix))]
+    fn kill_openvpn(self) -> io::Result<()> {
+        self.child.kill()
+    }
+}
+
+/// Internal enum to differentiate between if the child process or the event dispatcher died first.
+enum WaitResult {
+    Child(io::Result<::std::process::ExitStatus>),
+    EventDispatcher(talpid_ipc::Result<()>),
+}
 
 
 /// IPC server for listening to events coming from plugin loaded into OpenVPN.
@@ -125,8 +180,7 @@ pub struct OpenVpnEventDispatcher {
 impl OpenVpnEventDispatcher {
     /// Construct and start the IPC server with the given event listener callback.
     pub fn start<L>(on_event: L) -> talpid_ipc::Result<Self>
-        where L: Fn(openvpn_ffi::OpenVpnPluginEvent, openvpn_ffi::OpenVpnEnv),
-              L: Send + Sync + 'static
+        where L: Fn(OpenVpnPluginEvent, OpenVpnEnv) + Send + Sync + 'static
     {
         let rpc = OpenVpnEventApiImpl { on_event };
         let mut io = IoHandler::new();
@@ -140,7 +194,14 @@ impl OpenVpnEventDispatcher {
         self.server.address()
     }
 
-    /// Consumes the server and waits for it to finish.
+    /// Creates a handle to this event dispatcher, allowing the listening server to be closed while
+    /// some other thread is blocked in `wait`.
+    pub fn close_handle(&self) -> talpid_ipc::CloseHandle {
+        self.server.close_handle()
+    }
+
+    /// Consumes the server and waits for it to finish. Returns an error if the server exited
+    /// due to an error.
     pub fn wait(self) -> talpid_ipc::Result<()> {
         self.server.wait()
     }
@@ -153,8 +214,8 @@ mod api {
         pub trait OpenVpnEventApi {
             #[rpc(name = "openvpn_event")]
             fn openvpn_event(&self,
-                             openvpn_ffi::OpenVpnPluginEvent,
-                             openvpn_ffi::OpenVpnEnv)
+                             OpenVpnPluginEvent,
+                             OpenVpnEnv)
                              -> StdResult<(), Error>;
         }
     }
@@ -162,18 +223,15 @@ mod api {
 use self::api::*;
 
 struct OpenVpnEventApiImpl<L>
-    where L: Fn(openvpn_ffi::OpenVpnPluginEvent, openvpn_ffi::OpenVpnEnv) + Send + Sync + 'static
+    where L: Fn(OpenVpnPluginEvent, OpenVpnEnv) + Send + Sync + 'static
 {
     on_event: L,
 }
 
 impl<L> OpenVpnEventApi for OpenVpnEventApiImpl<L>
-    where L: Fn(openvpn_ffi::OpenVpnPluginEvent, openvpn_ffi::OpenVpnEnv) + Send + Sync + 'static
+    where L: Fn(OpenVpnPluginEvent, OpenVpnEnv) + Send + Sync + 'static
 {
-    fn openvpn_event(&self,
-                     event: openvpn_ffi::OpenVpnPluginEvent,
-                     env: openvpn_ffi::OpenVpnEnv)
-                     -> StdResult<(), Error> {
+    fn openvpn_event(&self, event: OpenVpnPluginEvent, env: OpenVpnEnv) -> StdResult<(), Error> {
         debug!("OpenVPN event {:?}", event);
         (self.on_event)(event, env);
         Ok(())