WIP Move talpid-core::mpsc to talpid-channelstalpid-channels

15 files changed, 51 insertions, 22 deletions

diff --git a/Cargo.lock b/Cargo.lock
index f9464141bc..a0ca1ebd4c 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3080,6 +3080,7 @@ dependencies = [
  "serde_json",
  "simple-signal",
  "socket2 0.5.8",
+ "talpid-channels",
  "talpid-core",
  "talpid-dbus",
  "talpid-dns",
@@ -5465,6 +5466,14 @@ dependencies = [
 ]
 
 [[package]]
+name = "talpid-channels"
+version = "0.0.0"
+dependencies = [
+ "futures",
+ "thiserror 2.0.9",
+]
+
+[[package]]
 name = "talpid-core"
 version = "0.0.0"
 dependencies = [
@@ -5483,6 +5492,7 @@ dependencies = [
  "rand 0.9.2",
  "serde",
  "socket2 0.5.8",
+ "talpid-channels",
  "talpid-dbus",
  "talpid-dns",
  "talpid-firewall",
diff --git a/Cargo.toml b/Cargo.toml
index b17068aa4e..ffeb83134c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -34,6 +34,7 @@ members = [
   "mullvad-update",
   "mullvad-update/mullvad-release",
   "mullvad-version",
+  "talpid-channels",
   "talpid-core",
   "talpid-dbus",
   "talpid-dns",
diff --git a/mullvad-daemon/Cargo.toml b/mullvad-daemon/Cargo.toml
index 2918afa7a9..a634904ecf 100644
--- a/mullvad-daemon/Cargo.toml
+++ b/mullvad-daemon/Cargo.toml
@@ -45,6 +45,7 @@ mullvad-version = { path = "../mullvad-version" }
 mullvad-update = { path = "../mullvad-update", features = ["client"] }
 mullvad-leak-checker = { path = "../mullvad-leak-checker", default-features = false }
 talpid-core = { path = "../talpid-core" }
+talpid-channels = { path = "../talpid-channels" }
 talpid-dns = { path = "../talpid-dns" }
 talpid-firewall = { path = "../talpid-firewall" }
 talpid-future = { path = "../talpid-future" }
diff --git a/mullvad-daemon/src/device/mod.rs b/mullvad-daemon/src/device/mod.rs
index cc6ce4c50a..9995c431d0 100644
--- a/mullvad-daemon/src/device/mod.rs
+++ b/mullvad-daemon/src/device/mod.rs
@@ -24,7 +24,7 @@ use std::{
     },
     time::{Duration, SystemTime},
 };
-use talpid_core::mpsc::Sender;
+use talpid_channels::mpsc::Sender;
 use talpid_types::{ErrorExt, tunnel::TunnelStateTransition};
 use tokio::{
     fs,
diff --git a/mullvad-daemon/src/geoip.rs b/mullvad-daemon/src/geoip.rs
index 09d52ba824..6f0e3ef0ad 100644
--- a/mullvad-daemon/src/geoip.rs
+++ b/mullvad-daemon/src/geoip.rs
@@ -4,7 +4,7 @@ use futures::join;
 use mullvad_api::rest::{Error, RequestServiceHandle};
 use mullvad_types::location::{AmIMullvad, GeoIpLocation, LocationEventData};
 use std::sync::LazyLock;
-use talpid_core::mpsc::Sender;
+use talpid_channels::mpsc::Sender;
 use talpid_future::retry::{ExponentialBackoff, Jittered, retry_future};
 use talpid_types::ErrorExt;
 
diff --git a/mullvad-daemon/src/lib.rs b/mullvad-daemon/src/lib.rs
index 9ba64cd007..3413c80f56 100644
--- a/mullvad-daemon/src/lib.rs
+++ b/mullvad-daemon/src/lib.rs
@@ -89,14 +89,12 @@ use std::{
     sync::{Arc, Weak},
     time::Duration,
 };
+use talpid_channels::mpsc::Sender;
 #[cfg(target_os = "android")]
 use talpid_core::connectivity_listener::ConnectivityListener;
 #[cfg(not(target_os = "android"))]
 use talpid_core::tunnel_state_machine::LockdownMode;
-use talpid_core::{
-    mpsc::Sender,
-    tunnel_state_machine::{self, TunnelCommand, TunnelStateMachineHandle},
-};
+use talpid_core::tunnel_state_machine::{self, TunnelCommand, TunnelStateMachineHandle};
 use talpid_routing::RouteManagerHandle;
 #[cfg(not(target_os = "android"))]
 use talpid_split_tunnel as split_tunnel;
@@ -619,12 +617,12 @@ impl<E> Sender<E> for DaemonEventSender<E>
 where
     InternalDaemonEvent: From<E>,
 {
-    fn send(&self, event: E) -> Result<(), talpid_core::mpsc::Error> {
+    fn send(&self, event: E) -> Result<(), talpid_channels::mpsc::Error> {
         match self.sender.upgrade() {
             Some(sender) => sender
                 .unbounded_send(InternalDaemonEvent::from(event))
-                .map_err(|_| talpid_core::mpsc::Error::ChannelClosed),
-            _ => Err(talpid_core::mpsc::Error::ChannelClosed),
+                .map_err(|_| talpid_channels::mpsc::Error::ChannelClosed),
+            _ => Err(talpid_channels::mpsc::Error::ChannelClosed),
         }
     }
 }
diff --git a/mullvad-daemon/src/migrations/device.rs b/mullvad-daemon/src/migrations/device.rs
index efbf926e83..700dfeeef3 100644
--- a/mullvad-daemon/src/migrations/device.rs
+++ b/mullvad-daemon/src/migrations/device.rs
@@ -13,7 +13,7 @@ use crate::{
 };
 use mullvad_types::{account::AccountNumber, wireguard::WireguardData};
 use std::time::Duration;
-use talpid_core::mpsc::Sender;
+use talpid_channels::mpsc::Sender;
 use talpid_types::ErrorExt;
 use tokio::time::timeout;
 
diff --git a/mullvad-daemon/src/version/check.rs b/mullvad-daemon/src/version/check.rs
index b21ad8a022..6f0a67224e 100644
--- a/mullvad-daemon/src/version/check.rs
+++ b/mullvad-daemon/src/version/check.rs
@@ -17,7 +17,7 @@ use std::{
     sync::LazyLock,
     time::{Duration, SystemTime},
 };
-use talpid_core::mpsc::Sender;
+use talpid_channels::mpsc::Sender;
 use talpid_future::retry::{ConstantInterval, retry_future};
 use talpid_types::ErrorExt;
 
diff --git a/mullvad-daemon/src/version/router.rs b/mullvad-daemon/src/version/router.rs
index 590e3336e8..eed924d6d5 100644
--- a/mullvad-daemon/src/version/router.rs
+++ b/mullvad-daemon/src/version/router.rs
@@ -9,7 +9,7 @@ use mullvad_types::version::{AppVersionInfo, SuggestedUpgrade};
 use mullvad_update::app::{AppDownloader, AppDownloaderParameters, HttpAppDownloader};
 use mullvad_update::version::VersionInfo;
 use mullvad_update::version::rollout::Rollout;
-use talpid_core::mpsc::Sender;
+use talpid_channels::mpsc::Sender;
 #[cfg(in_app_upgrade)]
 use talpid_types::ErrorExt;
 
diff --git a/talpid-channels/Cargo.toml b/talpid-channels/Cargo.toml
new file mode 100644
index 0000000000..ba4379b25f
--- /dev/null
+++ b/talpid-channels/Cargo.toml
@@ -0,0 +1,15 @@
+[package]
+name = "talpid-channels"
+description = "Extra channel abstractions and extensions"
+authors.workspace = true
+repository.workspace = true
+license.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+thiserror = { workspace = true }
+futures = { workspace = true }
diff --git a/talpid-channels/src/lib.rs b/talpid-channels/src/lib.rs
new file mode 100644
index 0000000000..6746321920
--- /dev/null
+++ b/talpid-channels/src/lib.rs
@@ -0,0 +1,2 @@
+//! Abstractions for working generically with different channel implementations.
+pub mod mpsc;
diff --git a/talpid-core/src/mpsc.rs b/talpid-channels/src/mpsc.rs
index d95f99fc7b..10f0b38dd7 100644
--- a/talpid-core/src/mpsc.rs
+++ b/talpid-channels/src/mpsc.rs
@@ -1,3 +1,11 @@
+//! Abstractions and extra features on `std::mpsc`
+
+/// Abstraction over any type that can be used similarly to an [std::mpsc::Sender].
+pub trait Sender<T> {
+    /// Sends an item over the underlying channel, failing only if the channel is closed.
+    fn send(&self, item: T) -> Result<(), Error>;
+}
+
 /// Error type for `Sender` trait.
 #[derive(thiserror::Error, Debug)]
 pub enum Error {
@@ -6,12 +14,7 @@ pub enum Error {
     ChannelClosed,
 }
 
-/// Abstraction over any type that can be used similarly to an `std::mpsc::Sender`.
-pub trait Sender<T> {
-    /// Sends an item over the underlying channel, failing only if the channel is closed.
-    fn send(&self, item: T) -> Result<(), Error>;
-}
-
+/// Implement [Sender] on [futures::channel::mpsc::UnboundedSender].
 impl<E> Sender<E> for futures::channel::mpsc::UnboundedSender<E> {
     fn send(&self, content: E) -> Result<(), Error> {
         self.unbounded_send(content)
diff --git a/talpid-core/Cargo.toml b/talpid-core/Cargo.toml
index f3c715b5be..1d64c57810 100644
--- a/talpid-core/Cargo.toml
+++ b/talpid-core/Cargo.toml
@@ -20,6 +20,7 @@ thiserror = { workspace = true }
 futures = { workspace = true }
 log = { workspace = true }
 rand = { workspace = true }
+talpid-channels = { path = "../talpid-channels" }
 talpid-dns = { path = "../talpid-dns" }
 talpid-firewall = { path = "../talpid-firewall" }
 talpid-routing = { path = "../talpid-routing" }
diff --git a/talpid-core/src/lib.rs b/talpid-core/src/lib.rs
index ffa61f7883..8e856c9091 100644
--- a/talpid-core/src/lib.rs
+++ b/talpid-core/src/lib.rs
@@ -12,9 +12,6 @@ mod offline;
 /// Helper function to preserve previous log files.
 pub mod logging;
 
-/// Abstractions and extra features on `std::mpsc`
-pub mod mpsc;
-
 /// State machine to handle tunnel configuration.
 pub mod tunnel_state_machine;
 
diff --git a/talpid-core/src/tunnel_state_machine/mod.rs b/talpid-core/src/tunnel_state_machine/mod.rs
index bce855b8d7..9e8345052c 100644
--- a/talpid-core/src/tunnel_state_machine/mod.rs
+++ b/talpid-core/src/tunnel_state_machine/mod.rs
@@ -23,6 +23,7 @@ use futures::{
     channel::{mpsc, oneshot},
     stream,
 };
+use talpid_channels::mpsc::Sender;
 use talpid_dns::{DnsConfig, DnsMonitor};
 use talpid_firewall::{Firewall, FirewallArguments, InitialFirewallState};
 use talpid_routing::RouteManagerHandle;
@@ -49,9 +50,9 @@ use self::{
 };
 #[cfg(target_os = "android")]
 use crate::connectivity_listener::ConnectivityListener;
+use crate::offline;
 #[cfg(any(windows, target_os = "android", target_os = "macos"))]
 use crate::split_tunnel;
-use crate::{mpsc::Sender, offline};
 
 const TUNNEL_STATE_MACHINE_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);