Relax T: Clone bound to only apply to static values

Each provider is enabled via a feature flag.
The currently implemented providers are:
- std::mpsc
- flume
- crossbeam_channel

.github/workflows/ci.yml

@@ -26,7 +26,7 @@ jobs:
           toolchain: stable
       - uses: Swatinem/rust-cache@v2
       - name: Minimal build
-        run: cargo build --no-default-features
+        run: cargo build 
       - name: Clippy
         run: cargo clippy --all-features --examples -- -D warnings
       - name: Build all

CHANGELOG.md

@@ -6,6 +6,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+## [0.0.2] - 2025-09-04
+
+### 🚀 Features
+
+- Add multiple channel providers
 ## [0.0.1] - 2025-09-02
 
 ### 🚀 Features

Cargo.toml

@@ -2,7 +2,7 @@
 name = "suck"
 description = "Suck data up through a channel"
 authors = ["Callum Leslie <git@cleslie.uk>"]
-version = "0.0.1"
+version = "0.0.2"
 edition = "2024"
 documentation = "https://docs.rs/suck"
 homepage = "https://github.com/callumio/suck"
@@ -15,5 +15,25 @@ exclude = ["flake.nix", "flake.lock", ".envrc", "cliff.toml", "release-plz.toml"
 
 [dependencies]
 thiserror = "2.0"
+flume = { version = "0.11", optional = true }
+crossbeam-channel = { version = "0.5", optional = true }
+arc-swap = "1.7.1"
+
+[features]
+default = ["all"]
+
+sync = []
+async = []
+
+sync-std = ["sync"]
+sync-flume = ["sync", "dep:flume"]
+sync-crossbeam = ["sync", "dep:crossbeam-channel"]
+
+all-sync = ["sync-std", "sync-flume", "sync-crossbeam"]
+
+all = ["all-sync"]
 
 [lib]
+
+[package.metadata.docs.rs]
+all-features = true

README.md

@@ -32,11 +32,11 @@ suck = "*"
 ## Quick Start
 
 ```rust
-use suck::SuckPair;
+use suck::sync::StdSuck;
 
 fn main() -> Result<(), Box<dyn std::error::Error>> {
-    // Create a pair
-    let (sucker, sourcer) = SuckPair::<i32>::pair();
+    // Create a pair (using default std backend)
+    let (sucker, sourcer) = StdSuck::<i32>::pair();
 
     // Start producer in a thread
     let producer = std::thread::spawn(move || {

flake.nix

@@ -20,11 +20,17 @@
     flake-utils.lib.eachDefaultSystem (system: let
       overlays = [(import rust-overlay)];
       pkgs = import nixpkgs {inherit system overlays;};
-      rustToolchain = pkgs.pkgsBuildHost.rust-bin.stable.latest.default;
+      rustToolchain = pkgs.pkgsBuildHost.rust-bin.fromRustupToolchainFile ./rust-toolchain.toml;
+      rustToolchainNightly = pkgs.pkgsBuildHost.rust-bin.nightly.latest.default;
       tools = with pkgs; [cargo-nextest];
-      nativeBuildInputs = with pkgs; [rustToolchain pkg-config] ++ tools;
+      nativeBuildInputs = with pkgs; [rustToolchain rustToolchainNightly pkg-config] ++ tools;
     in
       with pkgs; {
-        devShells.default = mkShell {inherit nativeBuildInputs;};
+        devShells.default = mkShell {
+          inherit nativeBuildInputs;
+          shellHook = ''
+            export CARGO_NIGHTLY="${rustToolchainNightly}/bin/cargo"
+          '';
+        };
       });
 }

src/channel.rs

@@ -1,90 +1,109 @@
-use std::sync::mpsc;
+use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Mutex};
 
 use crate::error::Error;
+use crate::sync::traits::{ChannelReceiver, ChannelSender};
 use crate::types::{ChannelState, Request, Response, ValueSource};
 
 /// The consumer side of the channel that requests values
-pub struct Sucker<T> {
-    request_tx: mpsc::Sender<Request>,
-    response_rx: mpsc::Receiver<Response<T>>,
-    closed: Mutex<bool>,
-}
-
-/// The producer side of the channel that provides values
-pub struct Sourcer<T> {
-    request_rx: mpsc::Receiver<Request>,
-    response_tx: mpsc::Sender<Response<T>>,
-    state: Arc<Mutex<ChannelState<T>>>,
-}
-
-/// Helper type for creating Sucker and Sourcer instances
-pub struct SuckPair<T> {
+pub struct Sucker<T, ST, SR>
+where
+    ST: ChannelSender<Request>,
+    SR: ChannelReceiver<Response<T>>,
+{
+    request_tx: ST,
+    response_rx: SR,
+    closed: AtomicBool,
     _phantom: std::marker::PhantomData<T>,
 }
 
-impl<T> SuckPair<T> {
-    /// Create a new suck pair
-    pub fn pair() -> (Sucker<T>, Sourcer<T>)
+impl<T, ST, SR> Sucker<T, ST, SR>
 where
-        T: Clone + Send + 'static,
+    ST: ChannelSender<Request>,
+    SR: ChannelReceiver<Response<T>>,
 {
-        let (request_tx, request_rx) = mpsc::channel();
-        let (response_tx, response_rx) = mpsc::channel();
-
-        let state = Arc::new(Mutex::new(ChannelState {
-            source: ValueSource::None,
-            closed: false,
-        }));
-
-        let sucker = Sucker {
+    /// Create a new Sucker instance
+    pub(crate) fn new(request_tx: ST, response_rx: SR) -> Self {
+        Self {
             request_tx,
             response_rx,
-            closed: Mutex::new(false),
-        };
+            closed: AtomicBool::new(false),
+            _phantom: std::marker::PhantomData,
+        }
+    }
+}
 
-        let sourcer = Sourcer {
+/// The producer side of the channel that provides values
+pub struct Sourcer<T, SR, ST>
+where
+    SR: ChannelReceiver<Request>,
+    ST: ChannelSender<Response<T>>,
+{
+    request_rx: SR,
+    response_tx: ST,
+    state: ChannelState<T>,
+    _phantom: std::marker::PhantomData<T>,
+}
+
+impl<T, SR, ST> Sourcer<T, SR, ST>
+where
+    SR: ChannelReceiver<Request>,
+    ST: ChannelSender<Response<T>>,
+{
+    /// Create a new Sourcer instance
+    pub(crate) fn new(request_rx: SR, response_tx: ST, state: ChannelState<T>) -> Self {
+        Self {
             request_rx,
             response_tx,
-            state: Arc::clone(&state),
-        };
-
-        (sucker, sourcer)
+            state,
+            _phantom: std::marker::PhantomData,
+        }
     }
 }
 
-impl<T> Sourcer<T>
+impl<T, SR, ST> Sourcer<T, SR, ST>
 where
-    T: Clone + Send + 'static,
+    T: Send + 'static,
+    SR: ChannelReceiver<Request>,
+    ST: ChannelSender<Response<T>>,
 {
     /// Set a fixed value
-    pub fn set_static(&self, value: T) -> Result<(), Error> {
-        let mut state = self.state.lock().map_err(|_| Error::InternalError)?;
-        if state.closed {
-            return Err(Error::ChannelClosed);
-        }
-        state.source = ValueSource::Static(value);
+    pub fn set_static(&self, val: T) -> Result<(), Error>
+    where
+        T: Clone,
+    {
+        self.state.swap(Arc::new(ValueSource::Static {
+            val,
+            clone: T::clone,
+        }));
         Ok(())
     }
 
-    /// Set a closure
+    /// Set a closure that implements [Fn]
     pub fn set<F>(&self, closure: F) -> Result<(), Error>
     where
         F: Fn() -> T + Send + Sync + 'static,
     {
-        let mut state = self.state.lock().map_err(|_| Error::InternalError)?;
-        if state.closed {
-            return Err(Error::ChannelClosed);
+        self.state
+            .swap(Arc::new(ValueSource::Dynamic(Box::new(closure))));
+        Ok(())
     }
-        state.source = ValueSource::Dynamic(Box::new(closure));
+
+    /// Set a closure that implements [FnMut]
+    pub fn set_mut<F>(&self, closure: F) -> Result<(), Error>
+    where
+        F: FnMut() -> T + Send + Sync + 'static,
+    {
+        self.state
+            .swap(Arc::new(ValueSource::DynamicMut(Mutex::new(Box::new(
+                closure,
+            )))));
         Ok(())
     }
 
     /// Close the channel
     pub fn close(&self) -> Result<(), Error> {
-        let mut state = self.state.lock().map_err(|_| Error::InternalError)?;
-        state.closed = true;
-        state.source = ValueSource::None;
+        self.state.swap(Arc::new(ValueSource::Cleared));
         Ok(())
     }
 
@@ -101,9 +120,7 @@ where
                 }
                 Ok(Request::Close) => {
                     // Close channel
-                    let mut state = self.state.lock().map_err(|_| Error::InternalError)?;
-                    state.closed = true;
-                    state.source = ValueSource::None;
+                    self.close()?;
                     break;
                 }
                 Err(_) => {
@@ -116,37 +133,53 @@ where
     }
 
     fn handle_get_value(&self) -> Result<Response<T>, Error> {
-        let state = self.state.lock().map_err(|_| Error::InternalError)?;
-        if state.closed {
-            return Ok(Response::Closed);
-        }
+        let state = self.state.load();
 
-        match &state.source {
-            ValueSource::Static(value) => Ok(Response::Value(value.clone())),
-            ValueSource::Dynamic(closure) => {
-                let value = self.execute_closure_safely(closure);
+        match &**state {
+            ValueSource::Static { val, clone } => {
+                let value = self.execute_closure_safely(&mut || clone(val));
                 match value {
                     Ok(v) => Ok(Response::Value(v)),
                     Err(_) => Ok(Response::NoSource), // Closure execution failed
                 }
             }
-            ValueSource::None => Ok(Response::NoSource),
+            ValueSource::Dynamic(closure) => {
+                let value = self.execute_closure_safely(&mut || closure());
+                match value {
+                    Ok(v) => Ok(Response::Value(v)),
+                    Err(_) => Ok(Response::NoSource), // Closure execution failed
+                }
+            }
+            ValueSource::DynamicMut(closure) => {
+                let mut closure = closure.lock().unwrap();
+                let value = self.execute_closure_safely(&mut *closure);
+                match value {
+                    Ok(v) => Ok(Response::Value(v)),
+                    Err(_) => Ok(Response::NoSource), // Closure execution failed
+                }
+            }
+            ValueSource::None => Ok(Response::NoSource), // No source was ever set
+            ValueSource::Cleared => Ok(Response::Closed), // Channel was closed (source was set then cleared)
         }
     }
 
     fn execute_closure_safely(
         &self,
-        closure: &dyn Fn() -> T,
+        closure: &mut dyn FnMut() -> T,
     ) -> Result<T, Box<dyn std::any::Any + Send>> {
         std::panic::catch_unwind(std::panic::AssertUnwindSafe(closure))
     }
 }
 
-impl<T> Sucker<T> {
+impl<T, ST, SR> Sucker<T, ST, SR>
+where
+    ST: ChannelSender<Request>,
+    SR: ChannelReceiver<Response<T>>,
+{
     /// Get the current value from the producer
     pub fn get(&self) -> Result<T, Error> {
         // Check if locally marked as closed
-        if *self.closed.lock().unwrap() {
+        if self.closed.load(Ordering::Acquire) {
             return Err(Error::ChannelClosed);
         }
 
@@ -171,7 +204,7 @@ impl<T> Sucker<T> {
     /// Close the channel from the consumer side
     pub fn close(&self) -> Result<(), Error> {
         // Mark locally as closed
-        *self.closed.lock().unwrap() = true;
+        self.closed.store(true, Ordering::Release);
 
         // Send close request
         self.request_tx

src/lib.rs

@@ -1,145 +1,12 @@
 #![doc = include_str!("../README.md")]
+#![cfg_attr(docsrs, feature(doc_auto_cfg))]
 
 pub mod channel;
 pub mod error;
+
+#[cfg(feature = "sync")]
+pub mod sync;
 pub mod types;
 
-// Re-exports
-pub use channel::{Sourcer, SuckPair, Sucker};
+pub use channel::{Sourcer, Sucker};
 pub use error::Error;
-pub use types::ValueSource;
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::thread;
-
-    #[test]
-    fn test_pre_computed_value() {
-        let (sucker, sourcer) = SuckPair::<i32>::pair();
-
-        // Start producer
-        let producer_handle = thread::spawn(move || {
-            sourcer.set_static(42).unwrap();
-            sourcer.run().unwrap();
-        });
-
-        // Ensure consumer gets the value
-        let value = sucker.get().unwrap();
-        assert_eq!(value, 42);
-
-        // Close consumer
-        sucker.close().unwrap();
-
-        producer_handle.join().unwrap();
-    }
-
-    #[test]
-    fn test_closure_value() {
-        let (sucker, sourcer) = SuckPair::<i32>::pair();
-
-        // Start producer
-        let producer_handle = std::thread::spawn(move || {
-            let counter = std::sync::Arc::new(std::sync::Mutex::new(0));
-            let counter_clone = std::sync::Arc::clone(&counter);
-            sourcer
-                .set(move || {
-                    let mut count = counter_clone.lock().unwrap();
-                    *count += 1;
-                    *count
-                })
-                .unwrap();
-            sourcer.run().unwrap();
-        });
-
-        // Ensure consumer gets the value
-        let value1 = sucker.get().unwrap();
-        assert_eq!(value1, 1);
-
-        // Ensure consumer gets the next value
-        let value2 = sucker.get().unwrap();
-        assert_eq!(value2, 2);
-
-        // Close consumer
-        sucker.close().unwrap();
-
-        producer_handle.join().unwrap();
-    }
-
-    #[test]
-    fn test_no_source_error() {
-        let (sucker, sourcer) = SuckPair::<i32>::pair();
-
-        // Start producer
-        let producer_handle = thread::spawn(move || {
-            sourcer.run().unwrap();
-        });
-
-        // Consumer should get NoSource error
-        let result = sucker.get();
-        assert!(matches!(result, Err(Error::NoSource)));
-
-        // Close consumer
-        sucker.close().unwrap();
-
-        producer_handle.join().unwrap();
-    }
-
-    #[test]
-    fn test_channel_closed_error() {
-        let (sucker, sourcer) = SuckPair::<i32>::pair();
-
-        // Start producer
-        let producer_handle = thread::spawn(move || {
-            sourcer.set_static(42).unwrap();
-            sourcer.run().unwrap();
-        });
-
-        // Close consumer
-        sucker.close().unwrap();
-
-        let result = sucker.get();
-        assert!(matches!(result, Err(Error::ChannelClosed)));
-
-        producer_handle.join().unwrap();
-    }
-
-    #[test]
-    fn test_producer_disconnection_error() {
-        let (sucker, sourcer) = SuckPair::<i32>::pair();
-
-        // Start producer
-        let producer_handle = thread::spawn(move || {
-            sourcer.set_static(42).unwrap();
-            // Simulate crash
-            panic!("Producer crashed!");
-        });
-
-        let result = sucker.get();
-        assert!(matches!(result, Err(Error::ProducerDisconnected)));
-
-        let _ = producer_handle.join();
-    }
-
-    #[test]
-    fn test_is_closed() {
-        let (sucker, sourcer) = SuckPair::<i32>::pair();
-
-        assert!(!sucker.is_closed());
-
-        // Start producer
-        let producer_handle = thread::spawn(move || {
-            sourcer.set_static(42).unwrap();
-            sourcer.run().unwrap();
-        });
-
-        // Get one value
-        let _ = sucker.get().unwrap();
-        assert!(!sucker.is_closed());
-
-        // Close and check
-        sucker.close().unwrap();
-
-        producer_handle.join().unwrap();
-    }
-}

src/sync/crossbeam.rs

@@ -0,0 +1,211 @@
+use std::sync::Arc;
+
+#[cfg(feature = "sync-crossbeam")]
+use crate::sync::traits::{ChannelError, ChannelReceiver, ChannelSender, ChannelType};
+use crate::types;
+use arc_swap::ArcSwap;
+use crossbeam_channel;
+
+type CrossbeamSucker<T> =
+    crate::Sucker<T, CrossbeamSender<types::Request>, CrossbeamReceiver<types::Response<T>>>;
+type CrossbeamSourcer<T> =
+    crate::Sourcer<T, CrossbeamReceiver<types::Request>, CrossbeamSender<types::Response<T>>>;
+
+/// Internal sender type for crossbeam backend
+pub struct CrossbeamSender<T>(crossbeam_channel::Sender<T>);
+/// Internal receiver type for crossbeam backend
+pub struct CrossbeamReceiver<T>(crossbeam_channel::Receiver<T>);
+
+impl<T> ChannelSender<T> for CrossbeamSender<T> {
+    fn send(&self, msg: T) -> Result<(), ChannelError> {
+        self.0
+            .send(msg)
+            .map_err(|_| ChannelError::ProducerDisconnected)
+    }
+}
+
+impl<T> ChannelReceiver<T> for CrossbeamReceiver<T> {
+    fn recv(&self) -> Result<T, ChannelError> {
+        self.0
+            .recv()
+            .map_err(|_| ChannelError::ProducerDisconnected)
+    }
+}
+
+/// Internal channel type for crossbeam backend
+pub struct CrossbeamChannel;
+
+impl ChannelType for CrossbeamChannel {
+    type Sender<T> = CrossbeamSender<T>;
+    type Receiver<T> = CrossbeamReceiver<T>;
+
+    fn create_request_channel() -> (Self::Sender<types::Request>, Self::Receiver<types::Request>) {
+        let (tx, rx) = crossbeam_channel::unbounded();
+        (CrossbeamSender(tx), CrossbeamReceiver(rx))
+    }
+
+    fn create_response_channel<T>() -> (
+        Self::Sender<types::Response<T>>,
+        Self::Receiver<types::Response<T>>,
+    ) {
+        let (tx, rx) = crossbeam_channel::unbounded();
+        (CrossbeamSender(tx), CrossbeamReceiver(rx))
+    }
+}
+
+pub struct CrossbeamSuck<T> {
+    _phantom: std::marker::PhantomData<T>,
+}
+
+impl<T> CrossbeamSuck<T> {
+    pub fn pair() -> (CrossbeamSucker<T>, CrossbeamSourcer<T>)
+    where
+        T: Clone + Send + 'static,
+    {
+        let (request_tx, request_rx) = CrossbeamChannel::create_request_channel();
+        let (response_tx, response_rx) = CrossbeamChannel::create_response_channel::<T>();
+
+        let state = ArcSwap::new(Arc::new(crate::types::ValueSource::None));
+
+        let sucker = crate::Sucker::new(request_tx, response_rx);
+        let sourcer = crate::Sourcer::new(request_rx, response_tx, state);
+
+        (sucker, sourcer)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::Error;
+    use std::thread;
+
+    #[test]
+    fn test_pre_computed_value() {
+        let (sucker, sourcer) = CrossbeamSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Ensure consumer gets the value
+        let value = sucker.get().unwrap();
+        assert_eq!(value, 42);
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_closure_value() {
+        let (sucker, sourcer) = CrossbeamSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = std::thread::spawn(move || {
+            let counter = std::sync::Arc::new(std::sync::Mutex::new(0));
+            let counter_clone = std::sync::Arc::clone(&counter);
+            sourcer
+                .set(move || {
+                    let mut count = counter_clone.lock().unwrap();
+                    *count += 1;
+                    *count
+                })
+                .unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Ensure consumer gets the value
+        let value1 = sucker.get().unwrap();
+        assert_eq!(value1, 1);
+
+        // Ensure consumer gets the next value
+        let value2 = sucker.get().unwrap();
+        assert_eq!(value2, 2);
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_no_source_error() {
+        let (sucker, sourcer) = CrossbeamSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.run().unwrap();
+        });
+
+        // Consumer should get NoSource error
+        let result = sucker.get();
+        assert!(matches!(result, Err(Error::NoSource)));
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_channel_closed_error() {
+        let (sucker, sourcer) = CrossbeamSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        let result = sucker.get();
+        assert!(matches!(result, Err(Error::ChannelClosed)));
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_producer_disconnection_error() {
+        let (sucker, sourcer) = CrossbeamSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            // Simulate crash
+            panic!("Producer crashed!");
+        });
+
+        let result = sucker.get();
+        assert!(matches!(result, Err(Error::ProducerDisconnected)));
+
+        let _ = producer_handle.join();
+    }
+
+    #[test]
+    fn test_is_closed() {
+        let (sucker, sourcer) = CrossbeamSuck::<i32>::pair();
+
+        assert!(!sucker.is_closed());
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Get one value
+        let _ = sucker.get().unwrap();
+        assert!(!sucker.is_closed());
+
+        // Close and check
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+}

src/sync/flume.rs

@@ -0,0 +1,212 @@
+use std::sync::Arc;
+
+#[cfg(feature = "sync-flume")]
+use crate::sync::traits::{ChannelError, ChannelReceiver, ChannelSender, ChannelType};
+use crate::types;
+use arc_swap::ArcSwap;
+use flume;
+
+type FlumeSucker<T> =
+    crate::Sucker<T, FlumeSender<types::Request>, FlumeReceiver<types::Response<T>>>;
+type FlumeSourcer<T> =
+    crate::Sourcer<T, FlumeReceiver<types::Request>, FlumeSender<types::Response<T>>>;
+
+/// Internal sender type for flume backend  
+pub struct FlumeSender<T>(flume::Sender<T>);
+/// Internal receiver type for flume backend
+pub struct FlumeReceiver<T>(flume::Receiver<T>);
+
+impl<T> ChannelSender<T> for FlumeSender<T> {
+    fn send(&self, msg: T) -> Result<(), ChannelError> {
+        self.0
+            .send(msg)
+            .map_err(|_| ChannelError::ProducerDisconnected)
+    }
+}
+
+impl<T> ChannelReceiver<T> for FlumeReceiver<T> {
+    fn recv(&self) -> Result<T, ChannelError> {
+        self.0
+            .recv()
+            .map_err(|_| ChannelError::ProducerDisconnected)
+    }
+}
+
+/// Internal channel type for flume backend
+pub struct FlumeChannel;
+
+impl ChannelType for FlumeChannel {
+    type Sender<T> = FlumeSender<T>;
+    type Receiver<T> = FlumeReceiver<T>;
+
+    fn create_request_channel() -> (Self::Sender<types::Request>, Self::Receiver<types::Request>) {
+        let (tx, rx) = flume::unbounded();
+        (FlumeSender(tx), FlumeReceiver(rx))
+    }
+
+    fn create_response_channel<T>() -> (
+        Self::Sender<types::Response<T>>,
+        Self::Receiver<types::Response<T>>,
+    ) {
+        let (tx, rx) = flume::unbounded();
+        (FlumeSender(tx), FlumeReceiver(rx))
+    }
+}
+
+pub struct FlumeSuck<T> {
+    _phantom: std::marker::PhantomData<T>,
+}
+
+impl<T> FlumeSuck<T> {
+    pub fn pair() -> (FlumeSucker<T>, FlumeSourcer<T>)
+    where
+        T: Clone + Send + 'static,
+    {
+        let (request_tx, request_rx) = FlumeChannel::create_request_channel();
+        let (response_tx, response_rx) = FlumeChannel::create_response_channel::<T>();
+
+        let state = Arc::new(crate::types::ValueSource::None);
+        let state = ArcSwap::new(state);
+
+        let sucker = crate::Sucker::new(request_tx, response_rx);
+        let sourcer = crate::Sourcer::new(request_rx, response_tx, state);
+
+        (sucker, sourcer)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::Error;
+    use std::thread;
+
+    #[test]
+    fn test_pre_computed_value() {
+        let (sucker, sourcer) = FlumeSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Ensure consumer gets the value
+        let value = sucker.get().unwrap();
+        assert_eq!(value, 42);
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_closure_value() {
+        let (sucker, sourcer) = FlumeSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = std::thread::spawn(move || {
+            let counter = std::sync::Arc::new(std::sync::Mutex::new(0));
+            let counter_clone = std::sync::Arc::clone(&counter);
+            sourcer
+                .set(move || {
+                    let mut count = counter_clone.lock().unwrap();
+                    *count += 1;
+                    *count
+                })
+                .unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Ensure consumer gets the value
+        let value1 = sucker.get().unwrap();
+        assert_eq!(value1, 1);
+
+        // Ensure consumer gets the next value
+        let value2 = sucker.get().unwrap();
+        assert_eq!(value2, 2);
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_no_source_error() {
+        let (sucker, sourcer) = FlumeSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.run().unwrap();
+        });
+
+        // Consumer should get NoSource error
+        let result = sucker.get();
+        assert!(matches!(result, Err(Error::NoSource)));
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_channel_closed_error() {
+        let (sucker, sourcer) = FlumeSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        let result = sucker.get();
+        assert!(matches!(result, Err(Error::ChannelClosed)));
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_producer_disconnection_error() {
+        let (sucker, sourcer) = FlumeSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            // Simulate crash
+            panic!("Producer crashed!");
+        });
+
+        let result = sucker.get();
+        assert!(matches!(result, Err(Error::ProducerDisconnected)));
+
+        let _ = producer_handle.join();
+    }
+
+    #[test]
+    fn test_is_closed() {
+        let (sucker, sourcer) = FlumeSuck::<i32>::pair();
+
+        assert!(!sucker.is_closed());
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Get one value
+        let _ = sucker.get().unwrap();
+        assert!(!sucker.is_closed());
+
+        // Close and check
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+}

src/sync/mod.rs

@@ -0,0 +1,17 @@
+pub mod traits;
+
+#[cfg(feature = "sync-crossbeam")]
+pub mod crossbeam;
+#[cfg(feature = "sync-flume")]
+pub mod flume;
+#[cfg(feature = "sync-std")]
+pub mod std;
+
+#[cfg(feature = "sync-flume")]
+pub use flume::FlumeSuck;
+
+#[cfg(feature = "sync-crossbeam")]
+pub use crossbeam::CrossbeamSuck;
+
+#[cfg(feature = "sync-std")]
+pub use std::StdSuck;

src/sync/std.rs

@@ -0,0 +1,210 @@
+use arc_swap::ArcSwap;
+
+use crate::sync::traits::{ChannelError, ChannelReceiver, ChannelSender, ChannelType};
+use crate::types;
+use std::sync::Arc;
+#[cfg(feature = "sync-std")]
+use std::sync::mpsc;
+
+type StdSucker<T> = crate::Sucker<T, StdSender<types::Request>, StdReceiver<types::Response<T>>>;
+type StdSourcer<T> = crate::Sourcer<T, StdReceiver<types::Request>, StdSender<types::Response<T>>>;
+
+/// Internal sender type for std backend
+pub struct StdSender<T>(mpsc::Sender<T>);
+/// Internal receiver type for std backend
+pub struct StdReceiver<T>(mpsc::Receiver<T>);
+
+impl<T> ChannelSender<T> for StdSender<T> {
+    fn send(&self, msg: T) -> Result<(), ChannelError> {
+        self.0
+            .send(msg)
+            .map_err(|_| ChannelError::ProducerDisconnected)
+    }
+}
+
+impl<T> ChannelReceiver<T> for StdReceiver<T> {
+    fn recv(&self) -> Result<T, ChannelError> {
+        self.0
+            .recv()
+            .map_err(|_| ChannelError::ProducerDisconnected)
+    }
+}
+
+/// Internal channel type for std backend
+pub struct StdChannel;
+
+impl ChannelType for StdChannel {
+    type Sender<T> = StdSender<T>;
+    type Receiver<T> = StdReceiver<T>;
+
+    fn create_request_channel() -> (Self::Sender<types::Request>, Self::Receiver<types::Request>) {
+        let (tx, rx) = mpsc::channel();
+        (StdSender(tx), StdReceiver(rx))
+    }
+
+    fn create_response_channel<T>() -> (
+        Self::Sender<types::Response<T>>,
+        Self::Receiver<types::Response<T>>,
+    ) {
+        let (tx, rx) = mpsc::channel();
+        (StdSender(tx), StdReceiver(rx))
+    }
+}
+
+pub struct StdSuck<T> {
+    _phantom: std::marker::PhantomData<T>,
+}
+
+impl<T> StdSuck<T> {
+    pub fn pair() -> (StdSucker<T>, StdSourcer<T>)
+    where
+        T: Clone + Send + 'static,
+    {
+        let (request_tx, request_rx) = StdChannel::create_request_channel();
+        let (response_tx, response_rx) = StdChannel::create_response_channel::<T>();
+
+        let state = Arc::new(crate::types::ValueSource::None);
+        let state = ArcSwap::new(state);
+
+        let sucker = crate::Sucker::new(request_tx, response_rx);
+        let sourcer = crate::Sourcer::new(request_rx, response_tx, state);
+
+        (sucker, sourcer)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::Error;
+    use std::thread;
+
+    #[test]
+    fn test_pre_computed_value() {
+        let (sucker, sourcer) = StdSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Ensure consumer gets the value
+        let value = sucker.get().unwrap();
+        assert_eq!(value, 42);
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_closure_value() {
+        let (sucker, sourcer) = StdSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = std::thread::spawn(move || {
+            let counter = std::sync::Arc::new(std::sync::Mutex::new(0));
+            let counter_clone = std::sync::Arc::clone(&counter);
+            sourcer
+                .set(move || {
+                    let mut count = counter_clone.lock().unwrap();
+                    *count += 1;
+                    *count
+                })
+                .unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Ensure consumer gets the value
+        let value1 = sucker.get().unwrap();
+        assert_eq!(value1, 1);
+
+        // Ensure consumer gets the next value
+        let value2 = sucker.get().unwrap();
+        assert_eq!(value2, 2);
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_no_source_error() {
+        let (sucker, sourcer) = StdSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.run().unwrap();
+        });
+
+        // Consumer should get NoSource error
+        let result = sucker.get();
+        assert!(matches!(result, Err(Error::NoSource)));
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_channel_closed_error() {
+        let (sucker, sourcer) = StdSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Close consumer
+        sucker.close().unwrap();
+
+        let result = sucker.get();
+        assert!(matches!(result, Err(Error::ChannelClosed)));
+
+        producer_handle.join().unwrap();
+    }
+
+    #[test]
+    fn test_producer_disconnection_error() {
+        let (sucker, sourcer) = StdSuck::<i32>::pair();
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            // Simulate crash
+            panic!("Producer crashed!");
+        });
+
+        let result = sucker.get();
+        assert!(matches!(result, Err(Error::ProducerDisconnected)));
+
+        let _ = producer_handle.join();
+    }
+
+    #[test]
+    fn test_is_closed() {
+        let (sucker, sourcer) = StdSuck::<i32>::pair();
+
+        assert!(!sucker.is_closed());
+
+        // Start producer
+        let producer_handle = thread::spawn(move || {
+            sourcer.set_static(42).unwrap();
+            sourcer.run().unwrap();
+        });
+
+        // Get one value
+        let _ = sucker.get().unwrap();
+        assert!(!sucker.is_closed());
+
+        // Close and check
+        sucker.close().unwrap();
+
+        producer_handle.join().unwrap();
+    }
+}

src/sync/traits.rs

@@ -0,0 +1,23 @@
+pub use crate::error::Error as ChannelError;
+
+pub trait ChannelSender<T> {
+    fn send(&self, msg: T) -> Result<(), ChannelError>;
+}
+
+pub trait ChannelReceiver<T> {
+    fn recv(&self) -> Result<T, ChannelError>;
+}
+
+pub trait ChannelType {
+    type Sender<T>: ChannelSender<T>;
+    type Receiver<T>: ChannelReceiver<T>;
+
+    fn create_request_channel() -> (
+        Self::Sender<crate::types::Request>,
+        Self::Receiver<crate::types::Request>,
+    );
+    fn create_response_channel<T>() -> (
+        Self::Sender<crate::types::Response<T>>,
+        Self::Receiver<crate::types::Response<T>>,
+    );
+}

src/types.rs

@@ -1,3 +1,7 @@
+use std::sync::Mutex;
+
+use arc_swap::ArcSwap;
+
 /// Request messages sent from consumer to producer
 pub enum Request {
     GetValue,
@@ -12,14 +16,13 @@ pub enum Response<T> {
 }
 
 /// Represents the source of values: either static or dynamic
-pub enum ValueSource<T> {
-    Static(T),
+pub(crate) enum ValueSource<T> {
+    Static { val: T, clone: fn(&T) -> T },
+    DynamicMut(Mutex<Box<dyn FnMut() -> T + Send + Sync + 'static>>),
     Dynamic(Box<dyn Fn() -> T + Send + Sync + 'static>),
-    None,
+    None,    // Never set
+    Cleared, // Was set but cleared (closed)
 }
 
 /// Internal channel state shared between producer and consumer
-pub struct ChannelState<T> {
-    pub source: ValueSource<T>,
-    pub closed: bool,
-}
+pub(crate) type ChannelState<T> = ArcSwap<ValueSource<T>>;