Merge branch 'upgrade-tokio-1'

6 files changed, 25 insertions, 17 deletions

diff --git a/talpid-core/src/future_retry.rs b/talpid-core/src/future_retry.rs
index 2a7a08a092..6e7c05fc35 100644
--- a/talpid-core/src/future_retry.rs
+++ b/talpid-core/src/future_retry.rs
@@ -32,10 +32,10 @@ pub async fn retry_future_with_backoff<
 async fn sleep(mut delay: Duration) {
     while delay > MAX_SINGLE_DELAY {
         delay -= MAX_SINGLE_DELAY;
-        tokio::time::delay_for(MAX_SINGLE_DELAY).await;
+        tokio::time::sleep(MAX_SINGLE_DELAY).await;
     }
 
-    tokio::time::delay_for(delay).await;
+    tokio::time::sleep(delay).await;
 }
 
 /// Provides an exponential back-off timer to delay the next retry of a failed operation.
diff --git a/talpid-core/src/routing/linux.rs b/talpid-core/src/routing/linux.rs
index a63430fed3..72ebe5b27c 100644
--- a/talpid-core/src/routing/linux.rs
+++ b/talpid-core/src/routing/linux.rs
@@ -629,7 +629,8 @@ impl RouteManagerImpl {
                 let mut add_message = self
                     .handle
                     .route()
-                    .add_v4()
+                    .add()
+                    .v4()
                     .destination_prefix(v4_prefix.ip(), v4_prefix.prefix());
 
                 if v4_prefix.prefix() > 0 && v4_prefix.prefix() < 32 {
@@ -653,7 +654,8 @@ impl RouteManagerImpl {
                 let mut add_message = self
                     .handle
                     .route()
-                    .add_v6()
+                    .add()
+                    .v6()
                     .destination_prefix(v6_prefix.ip(), v6_prefix.prefix());
 
                 if v6_prefix.prefix() > 0 && v6_prefix.prefix() < 128 {
@@ -826,7 +828,7 @@ mod test {
     /// Tests if dropping inside a tokio runtime panics
     #[test]
     fn test_drop_in_executor() {
-        let mut runtime = tokio::runtime::Runtime::new().expect("Failed to initialize runtime");
+        let runtime = tokio::runtime::Runtime::new().expect("Failed to initialize runtime");
         runtime.block_on(async {
             let manager = RouteManagerImpl::new(HashSet::new())
                 .await
@@ -838,7 +840,7 @@ mod test {
     /// Tests if dropping outside a runtime panics
     #[test]
     fn test_drop() {
-        let mut runtime = tokio::runtime::Runtime::new().expect("Failed to initialize runtime");
+        let runtime = tokio::runtime::Runtime::new().expect("Failed to initialize runtime");
         let manager = runtime.block_on(async {
             RouteManagerImpl::new(HashSet::new())
                 .await
diff --git a/talpid-core/src/routing/macos.rs b/talpid-core/src/routing/macos.rs
index e15899c551..6808ad3e7b 100644
--- a/talpid-core/src/routing/macos.rs
+++ b/talpid-core/src/routing/macos.rs
@@ -13,6 +13,7 @@ use std::{
     process::{ExitStatus, Stdio},
 };
 use tokio::{io::AsyncBufReadExt, process::Command};
+use tokio_stream::wrappers::LinesStream;
 
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -332,7 +333,7 @@ async fn listen_for_default_route_changes() -> Result<impl Stream<Item = std::io
     let mut add_or_delete_message = false;
     let mut contains_default = false;
 
-    let monitor = lines.try_filter_map(move |line| {
+    let monitor = LinesStream::new(lines).try_filter_map(move |line| {
         if add_or_delete_message {
             if line.contains("default") {
                 contains_default = true;
diff --git a/talpid-core/src/tunnel/openvpn/mod.rs b/talpid-core/src/tunnel/openvpn/mod.rs
index f25f2624da..8ae967975a 100644
--- a/talpid-core/src/tunnel/openvpn/mod.rs
+++ b/talpid-core/src/tunnel/openvpn/mod.rs
@@ -528,9 +528,8 @@ impl<C: OpenVpnBuilder + Send + 'static> OpenVpnMonitor<C> {
             format!("/tmp/talpid-openvpn-{}", uuid)
         };
 
-        let mut runtime = tokio::runtime::Builder::new()
-            .threaded_scheduler()
-            .core_threads(1)
+        let runtime = tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(1)
             .enable_all()
             .build()
             .map_err(Error::RuntimeError)?;
@@ -955,7 +954,7 @@ mod event_server {
     };
     #[cfg(any(target_os = "linux", windows))]
     use talpid_types::ErrorExt;
-    use tokio::io::{AsyncRead, AsyncWrite};
+    use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
     use tonic::{
         self,
         transport::{server::Connected, Server},
@@ -1201,13 +1200,19 @@ mod event_server {
 
     #[derive(Debug)]
     pub struct StreamBox<T: AsyncRead + AsyncWrite>(pub T);
-    impl<T: AsyncRead + AsyncWrite> Connected for StreamBox<T> {}
+    impl<T: AsyncRead + AsyncWrite> Connected for StreamBox<T> {
+        type ConnectInfo = Option<()>;
+
+        fn connect_info(&self) -> Self::ConnectInfo {
+            None
+        }
+    }
     impl<T: AsyncRead + AsyncWrite + Unpin> AsyncRead for StreamBox<T> {
         fn poll_read(
             mut self: Pin<&mut Self>,
             cx: &mut Context<'_>,
-            buf: &mut [u8],
-        ) -> Poll<std::io::Result<usize>> {
+            buf: &mut ReadBuf<'_>,
+        ) -> Poll<std::io::Result<()>> {
             Pin::new(&mut self.0).poll_read(cx, buf)
         }
     }
diff --git a/talpid-core/src/tunnel/wireguard/mod.rs b/talpid-core/src/tunnel/wireguard/mod.rs
index 034cabd316..53740a93b4 100644
--- a/talpid-core/src/tunnel/wireguard/mod.rs
+++ b/talpid-core/src/tunnel/wireguard/mod.rs
@@ -123,11 +123,11 @@ impl TcpProxy {
             .block_on(Udp2Tcp::new(
                 listen_addr,
                 endpoint,
-                Some(&TcpOptions {
+                TcpOptions {
                     #[cfg(target_os = "linux")]
                     fwmark: Some(crate::linux::TUNNEL_FW_MARK),
                     ..TcpOptions::default()
-                }),
+                },
             ))
             .map_err(Error::Udp2TcpError)?;
 
diff --git a/talpid-core/src/tunnel/wireguard/wireguard_kernel/mod.rs b/talpid-core/src/tunnel/wireguard/wireguard_kernel/mod.rs
index 803613c332..04f34011ff 100644
--- a/talpid-core/src/tunnel/wireguard/wireguard_kernel/mod.rs
+++ b/talpid-core/src/tunnel/wireguard/wireguard_kernel/mod.rs
@@ -15,7 +15,7 @@ use netlink_proto::{
     ConnectionHandle, Error as NetlinkError,
 };
 use std::{ffi::CString, net::IpAddr};
-use tokio::stream::StreamExt;
+use tokio_stream::StreamExt;
 
 mod parsers;