Use futures::future::abortable instead

6 files changed, 24 insertions, 29 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 2fcdf232d9..e9415bc4b6 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1527,7 +1527,6 @@ dependencies = [
  "regex 1.3.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde 1.0.101 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 1.0.41 (registry+https://github.com/rust-lang/crates.io-index)",
- "talpid-core 0.1.0",
  "talpid-types 0.1.0",
  "tempfile 3.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio 0.2.21 (registry+https://github.com/rust-lang/crates.io-index)",
diff --git a/mullvad-daemon/src/lib.rs b/mullvad-daemon/src/lib.rs
index 3e5ba1dda9..997770f8de 100644
--- a/mullvad-daemon/src/lib.rs
+++ b/mullvad-daemon/src/lib.rs
@@ -18,6 +18,7 @@ mod settings;
 pub mod version;
 mod version_check;
 
+use futures::future::{abortable, AbortHandle};
 use futures01::{
     future::{self, Executor},
     stream::Wait,
@@ -58,7 +59,6 @@ use std::{
 #[cfg(target_os = "linux")]
 use talpid_core::split_tunnel;
 use talpid_core::{
-    future_cancel::{CancelHandle, Cancellable},
     mpsc::Sender,
     tunnel_state_machine::{self, TunnelCommand, TunnelParametersGenerator},
 };
@@ -462,7 +462,7 @@ pub struct Daemon<L: EventListener> {
     exclude_pids: split_tunnel::PidManager,
     rx: Wait<UnboundedReceiver<InternalDaemonEvent>>,
     tx: DaemonEventSender,
-    reconnection_job: Option<CancelHandle>,
+    reconnection_job: Option<AbortHandle>,
     event_listener: L,
     settings: SettingsPersister,
     account_history: account_history::AccountHistory,
@@ -948,19 +948,19 @@ where
 
     fn schedule_reconnect(&mut self, delay: Duration) {
         let tunnel_command_tx = self.tx.to_specialized_sender();
-        let (future, cancel_handle) = Cancellable::new(Box::pin(async move {
+        let (future, abort_handle) = abortable(Box::pin(async move {
             tokio02::time::delay_for(delay).await;
             log::debug!("Attempting to reconnect");
             let _ = tunnel_command_tx.send(DaemonCommand::Reconnect);
         }));
 
         self.spawn_future(future);
-        self.reconnection_job = Some(cancel_handle);
+        self.reconnection_job = Some(abort_handle);
     }
 
     fn unschedule_reconnect(&mut self) {
         if let Some(job) = self.reconnection_job.take() {
-            job.cancel();
+            job.abort();
         }
     }
 
diff --git a/mullvad-daemon/src/wireguard.rs b/mullvad-daemon/src/wireguard.rs
index 9206f97be2..2ebd5fae50 100644
--- a/mullvad-daemon/src/wireguard.rs
+++ b/mullvad-daemon/src/wireguard.rs
@@ -9,8 +9,8 @@ use std::{
     time::{Duration, Instant},
 };
 
+use futures::future::{abortable, AbortHandle};
 use talpid_core::{
-    future_cancel::{CancelHandle, Cancellable},
     future_retry::{retry_future_with_backoff, ExponentialBackoff, Jittered},
     mpsc::Sender,
 };
@@ -44,9 +44,9 @@ pub type Result<T> = std::result::Result<T, Error>;
 pub struct KeyManager {
     daemon_tx: DaemonEventSender,
     http_handle: MullvadRestHandle,
-    current_job: Option<CancelHandle>,
+    current_job: Option<AbortHandle>,
 
-    abort_scheduler_tx: Option<CancelHandle>,
+    abort_scheduler_tx: Option<AbortHandle>,
     auto_rotation_interval: Duration,
 }
 
@@ -99,7 +99,7 @@ impl KeyManager {
     /// Stop current key generation
     pub fn reset(&mut self) {
         if let Some(job) = self.current_job.take() {
-            job.cancel()
+            job.abort()
         }
     }
 
@@ -204,7 +204,7 @@ impl KeyManager {
             retry_future_with_backoff(future_generator, should_retry, retry_strategy);
 
 
-        let (cancellable_upload, cancel_handle) = Cancellable::new(Box::pin(upload_future));
+        let (cancellable_upload, abort_handle) = abortable(Box::pin(upload_future));
         let daemon_tx = self.daemon_tx.clone();
         let future = async move {
             match cancellable_upload.await {
@@ -223,7 +223,7 @@ impl KeyManager {
 
 
         self.http_handle.service().spawn(Box::pin(future));
-        self.current_job = Some(cancel_handle);
+        self.current_job = Some(abort_handle);
     }
 
 
@@ -392,16 +392,16 @@ impl KeyManager {
             self.auto_rotation_interval.as_secs(),
             account_token,
         );
-        let (cancellable, cancel_handle) = Cancellable::new(Box::pin(fut));
+        let (request, abort_handle) = abortable(Box::pin(fut));
 
-        self.http_handle.service().spawn(cancellable);
-        self.abort_scheduler_tx = Some(cancel_handle);
+        self.http_handle.service().spawn(request);
+        self.abort_scheduler_tx = Some(abort_handle);
     }
 
     fn stop_automatic_rotation(&mut self) {
-        if let Some(cancel_handle) = self.abort_scheduler_tx.take() {
+        if let Some(abort_handle) = self.abort_scheduler_tx.take() {
             log::info!("Stopping automatic key rotation");
-            cancel_handle.cancel();
+            abort_handle.abort();
         }
     }
 }
diff --git a/mullvad-rpc/Cargo.toml b/mullvad-rpc/Cargo.toml
index ad82fd736d..3c5a29a617 100644
--- a/mullvad-rpc/Cargo.toml
+++ b/mullvad-rpc/Cargo.toml
@@ -28,7 +28,6 @@ webpki = { version = "0.21", features =  [] }
 
 mullvad-types = { path = "../mullvad-types" }
 talpid-types = { path = "../talpid-types" }
-talpid-core = { path = "../talpid-core" }
 
 [dev-dependencies]
 filetime = "0.2"
diff --git a/mullvad-rpc/src/rest.rs b/mullvad-rpc/src/rest.rs
index 4a8220749b..205f350ea5 100644
--- a/mullvad-rpc/src/rest.rs
+++ b/mullvad-rpc/src/rest.rs
@@ -1,5 +1,6 @@
 use futures::{
     channel::{mpsc, oneshot},
+    future::{abortable, AbortHandle, Aborted},
     sink::SinkExt,
     stream::StreamExt,
     TryFutureExt,
@@ -11,7 +12,6 @@ use hyper::{
     Method, Uri,
 };
 use std::{collections::BTreeMap, future::Future, mem, net::IpAddr, str::FromStr, time::Duration};
-use talpid_core::future_cancel::{CancelErr, CancelHandle, Cancellable};
 use tokio::runtime::Handle;
 
 pub use hyper::StatusCode;
@@ -27,7 +27,7 @@ const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
 #[derive(err_derive::Error, Debug)]
 pub enum Error {
     #[error(display = "Request cancelled")]
-    Cancelled(CancelErr),
+    Aborted(Aborted),
 
     #[error(display = "Hyper error")]
     HyperError(#[error(source)] hyper::Error),
@@ -68,7 +68,7 @@ pub(crate) struct RequestService<C> {
     connector: C,
     handle: Handle,
     next_id: u64,
-    in_flight_requests: BTreeMap<u64, CancelHandle>,
+    in_flight_requests: BTreeMap<u64, AbortHandle>,
 }
 
 impl<C: Connect + Clone + Send + Sync + 'static> RequestService<C> {
@@ -107,7 +107,7 @@ impl<C: Connect + Clone + Send + Sync + 'static> RequestService<C> {
                 let mut tx = self.command_tx.clone();
                 let timeout = request.timeout();
 
-                let (request_future, cancel_handle) = Cancellable::new(
+                let (request_future, abort_handle) = abortable(
                     self.client
                         .request(request.into_request())
                         .map_err(Error::from),
@@ -115,7 +115,7 @@ impl<C: Connect + Clone + Send + Sync + 'static> RequestService<C> {
 
                 let future = async move {
                     let response =
-                        tokio::time::timeout(timeout, request_future.map_err(Error::Cancelled))
+                        tokio::time::timeout(timeout, request_future.map_err(Error::Aborted))
                             .await
                             .map_err(Error::TimeoutError);
 
@@ -131,7 +131,7 @@ impl<C: Connect + Clone + Send + Sync + 'static> RequestService<C> {
 
 
                 self.handle.spawn(future);
-                self.in_flight_requests.insert(id, cancel_handle);
+                self.in_flight_requests.insert(id, abort_handle);
             }
 
             RequestCommand::RequestFinished(id) => {
@@ -146,8 +146,8 @@ impl<C: Connect + Clone + Send + Sync + 'static> RequestService<C> {
 
     fn reset(&mut self) {
         let old_requests = mem::replace(&mut self.in_flight_requests, BTreeMap::new());
-        for (_, cancel_handle) in old_requests.into_iter() {
-            cancel_handle.cancel();
+        for (_, abort_handle) in old_requests.into_iter() {
+            abort_handle.abort();
         }
         let _ = mem::replace(&mut self.client, Self::new_client(self.connector.clone()));
         self.next_id = 0;
diff --git a/talpid-core/src/lib.rs b/talpid-core/src/lib.rs
index f41031ad56..fbf06c8815 100644
--- a/talpid-core/src/lib.rs
+++ b/talpid-core/src/lib.rs
@@ -48,9 +48,6 @@ pub mod tunnel_state_machine;
 /// Future utilities
 pub mod future_retry;
 
-/// Cancel a future remotely
-pub mod future_cancel;
-
 #[cfg(not(target_os = "android"))]
 /// Internal code for managing bundled proxy software.
 mod proxy;