Merge branch 'fix-openvpn-plugin-error-handling'

1 files changed, 10 insertions, 16 deletions

diff --git a/talpid-openvpn-plugin/src/processing.rs b/talpid-openvpn-plugin/src/processing.rs
index bc896cc3b3..9601fa368e 100644
--- a/talpid-openvpn-plugin/src/processing.rs
+++ b/talpid-openvpn-plugin/src/processing.rs
@@ -6,7 +6,7 @@ use jsonrpc_client_core::{
     expand_params, jsonrpc_client, Future, Result as ClientResult, Transport,
 };
 use jsonrpc_client_ipc::IpcTransport;
-use std::{collections::HashMap, thread};
+use std::{collections::HashMap, sync::mpsc, thread};
 use tokio::{reactor::Handle, runtime::Runtime};
 
 error_chain! {
@@ -26,13 +26,14 @@ error_chain! {
 /// Struct processing OpenVPN events and notifies listeners over IPC
 pub struct EventProcessor {
     ipc_client: EventProxy,
-    client_stop: ::std::sync::mpsc::Receiver<ClientResult<()>>,
+    client_result_rx: mpsc::Receiver<ClientResult<()>>,
 }
 
 impl EventProcessor {
     pub fn new(arguments: Arguments) -> Result<EventProcessor> {
         log::trace!("Creating EventProcessor");
-        let (start_tx, start_rx) = futures::sync::oneshot::channel();
+        let (start_tx, start_rx) = mpsc::channel();
+        let (client_result_tx, client_result_rx) = mpsc::channel();
         thread::spawn(move || {
             let mut rt = Runtime::new().expect("failed to spawn runtime");
 
@@ -41,22 +42,16 @@ impl EventProcessor {
                     .expect("Unable to create IPC transport")
                     .into_client();
 
-            let (tx, client_stop) = ::std::sync::mpsc::channel();
-            let client_future = client.then(move |result| tx.send(result)).map_err(|_| ());
-            start_tx
-                .send((client_stop, client_handle))
-                .expect("failed to send client handles");
-
-            rt.block_on(client_future)
-                .expect("RPC client should not fail");
+            let _ = start_tx.send(client_handle);
+            let _ = client_result_tx.send(rt.block_on(client));
         });
 
-        let (client_stop, client_handle) = start_rx.wait().chain_err(|| ErrorKind::Shutdown)?;
+        let client_handle = start_rx.recv().chain_err(|| ErrorKind::Shutdown)?;
         let ipc_client = EventProxy::new(client_handle);
 
         Ok(EventProcessor {
             ipc_client,
-            client_stop,
+            client_result_rx,
         })
     }
 
@@ -76,10 +71,9 @@ impl EventProcessor {
 
     fn check_client_status(&mut self) -> Result<()> {
         use std::sync::mpsc::TryRecvError::*;
-        match self.client_stop.try_recv() {
+        match self.client_result_rx.try_recv() {
             Err(Empty) => Ok(()),
-            Err(Disconnected) => Err(ErrorKind::Shutdown.into()),
-            Ok(Ok(_)) => Err(ErrorKind::Shutdown.into()),
+            Err(Disconnected) | Ok(Ok(())) => Err(ErrorKind::Shutdown.into()),
             Ok(Err(e)) => Err(Error::with_chain(e, ErrorKind::IpcSendingError)),
         }
     }