cleslie/suck
1
0
Fork 0
mirror of https://github.com/callumio/suck.git synced 2026-03-21 22:18:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor: reorganize channel modules and implement async/sync structures

Browse source
This commit is contained in:
Callum Leslie 2026-03-04 21:01:46 +00:00
parent 8366421ede
commit a8d73eba4b
9 changed files with 46 additions and 30 deletions
Download patch file Download diff file Expand all files Collapse all files

0
src/async_channel.rs → src/asynchronous/channel.rs
View file

1
src/asynchronous/mod.rs
View file

@ -1,3 +1,4 @@
pub mod channel;
pub mod traits; pub mod traits;
#[cfg(feature = "async-tokio")] #[cfg(feature = "async-tokio")]

19
src/asynchronous/tokio.rs
View file

@ -9,10 +9,16 @@ use crate::asynchronous::traits::{
}; };
use crate::types; use crate::types;
type TokioSucker<T> = type TokioSucker<T> = crate::asynchronous::channel::AsyncSucker<
crate::AsyncSucker<T, TokioSender<types::Request>, TokioReceiver<types::Response<T>>>; T,
type TokioSourcer<T> = TokioSender<types::Request>,
crate::AsyncSourcer<T, TokioReceiver<types::Request>, TokioSender<types::Response<T>>>; TokioReceiver<types::Response<T>>,
>;
type TokioSourcer<T> = crate::asynchronous::channel::AsyncSourcer<
T,
TokioReceiver<types::Request>,
TokioSender<types::Response<T>>,
>;
pub struct TokioSender<T>(mpsc::UnboundedSender<T>); pub struct TokioSender<T>(mpsc::UnboundedSender<T>);
pub struct TokioReceiver<T>(Mutex<mpsc::UnboundedReceiver<T>>); pub struct TokioReceiver<T>(Mutex<mpsc::UnboundedReceiver<T>>);
@ -70,8 +76,9 @@ impl<T> TokioSuck<T> {
let (response_tx, response_rx) = TokioChannel::create_response_channel::<T>(); let (response_tx, response_rx) = TokioChannel::create_response_channel::<T>();
let state = ArcSwap::new(Arc::new(crate::types::ValueSource::None)); let state = ArcSwap::new(Arc::new(crate::types::ValueSource::None));
let sucker = crate::AsyncSucker::new(request_tx, response_rx); let sucker = crate::asynchronous::channel::AsyncSucker::new(request_tx, response_rx);
let sourcer = crate::AsyncSourcer::new(request_rx, response_tx, state); let sourcer =
crate::asynchronous::channel::AsyncSourcer::new(request_rx, response_tx, state);
(sucker, sourcer) (sucker, sourcer)
} }

10
src/lib.rs
View file

@ -1,10 +1,6 @@
#![doc = include_str!("../README.md")] #![doc = include_str!("../README.md")]
#![cfg_attr(docsrs, feature(doc_auto_cfg))] #![cfg_attr(docsrs, feature(doc_auto_cfg))]
#[cfg(feature = "async")]
pub mod async_channel;
#[cfg(feature = "sync")]
pub mod channel;
pub mod error; pub mod error;
#[cfg(feature = "async")] #[cfg(feature = "async")]
@ -15,7 +11,7 @@ pub mod sync;
pub mod types; pub mod types;
#[cfg(feature = "async")] #[cfg(feature = "async")]
pub use async_channel::{AsyncSourcer, AsyncSucker}; pub use asynchronous::channel::{AsyncSourcer, AsyncSucker};
#[cfg(feature = "sync")]
pub use channel::{Sourcer, Sucker};
pub use error::Error; pub use error::Error;
#[cfg(feature = "sync")]
pub use sync::channel::{Sourcer, Sucker};

0
src/channel.rs → src/sync/channel.rs
View file

18
src/sync/crossbeam.rs
View file

@ -6,10 +6,16 @@ use crate::types;
use arc_swap::ArcSwap; use arc_swap::ArcSwap;
use crossbeam_channel; use crossbeam_channel;
type CrossbeamSucker<T> = type CrossbeamSucker<T> = crate::sync::channel::Sucker<
crate::Sucker<T, CrossbeamSender<types::Request>, CrossbeamReceiver<types::Response<T>>>; T,
type CrossbeamSourcer<T> = CrossbeamSender<types::Request>,
crate::Sourcer<T, CrossbeamReceiver<types::Request>, CrossbeamSender<types::Response<T>>>; CrossbeamReceiver<types::Response<T>>,
>;
type CrossbeamSourcer<T> = crate::sync::channel::Sourcer<
T,
CrossbeamReceiver<types::Request>,
CrossbeamSender<types::Response<T>>,
>;
/// Internal sender type for crossbeam backend /// Internal sender type for crossbeam backend
pub struct CrossbeamSender<T>(crossbeam_channel::Sender<T>); pub struct CrossbeamSender<T>(crossbeam_channel::Sender<T>);
@ -67,8 +73,8 @@ impl<T> CrossbeamSuck<T> {
let state = ArcSwap::new(Arc::new(crate::types::ValueSource::None)); let state = ArcSwap::new(Arc::new(crate::types::ValueSource::None));
let sucker = crate::Sucker::new(request_tx, response_rx); let sucker = crate::sync::channel::Sucker::new(request_tx, response_rx);
let sourcer = crate::Sourcer::new(request_rx, response_tx, state); let sourcer = crate::sync::channel::Sourcer::new(request_rx, response_tx, state);
(sucker, sourcer) (sucker, sourcer)
} }

13
src/sync/flume.rs
View file

@ -7,9 +7,12 @@ use arc_swap::ArcSwap;
use flume; use flume;
type FlumeSucker<T> = type FlumeSucker<T> =
crate::Sucker<T, FlumeSender<types::Request>, FlumeReceiver<types::Response<T>>>; crate::sync::channel::Sucker<T, FlumeSender<types::Request>, FlumeReceiver<types::Response<T>>>;
type FlumeSourcer<T> = type FlumeSourcer<T> = crate::sync::channel::Sourcer<
crate::Sourcer<T, FlumeReceiver<types::Request>, FlumeSender<types::Response<T>>>; T,
FlumeReceiver<types::Request>,
FlumeSender<types::Response<T>>,
>;
/// Internal sender type for flume backend /// Internal sender type for flume backend
pub struct FlumeSender<T>(flume::Sender<T>); pub struct FlumeSender<T>(flume::Sender<T>);
@ -68,8 +71,8 @@ impl<T> FlumeSuck<T> {
let state = Arc::new(crate::types::ValueSource::None); let state = Arc::new(crate::types::ValueSource::None);
let state = ArcSwap::new(state); let state = ArcSwap::new(state);
let sucker = crate::Sucker::new(request_tx, response_rx); let sucker = crate::sync::channel::Sucker::new(request_tx, response_rx);
let sourcer = crate::Sourcer::new(request_rx, response_tx, state); let sourcer = crate::sync::channel::Sourcer::new(request_rx, response_tx, state);
(sucker, sourcer) (sucker, sourcer)
} }

1
src/sync/mod.rs
View file

@ -1,3 +1,4 @@
pub mod channel;
pub mod traits; pub mod traits;
#[cfg(feature = "sync-crossbeam")] #[cfg(feature = "sync-crossbeam")]

14
src/sync/std.rs
View file

@ -6,8 +6,10 @@ use std::sync::Arc;
#[cfg(feature = "sync-std")] #[cfg(feature = "sync-std")]
use std::sync::mpsc; use std::sync::mpsc;
type StdSucker<T> = crate::Sucker<T, StdSender<types::Request>, StdReceiver<types::Response<T>>>; type StdSucker<T> =
type StdSourcer<T> = crate::Sourcer<T, StdReceiver<types::Request>, StdSender<types::Response<T>>>; crate::sync::channel::Sucker<T, StdSender<types::Request>, StdReceiver<types::Response<T>>>;
type StdSourcer<T> =
crate::sync::channel::Sourcer<T, StdReceiver<types::Request>, StdSender<types::Response<T>>>;
/// Internal sender type for std backend /// Internal sender type for std backend
pub struct StdSender<T>(mpsc::Sender<T>); pub struct StdSender<T>(mpsc::Sender<T>);
@ -66,8 +68,8 @@ impl<T> StdSuck<T> {
let state = Arc::new(crate::types::ValueSource::None); let state = Arc::new(crate::types::ValueSource::None);
let state = ArcSwap::new(state); let state = ArcSwap::new(state);
let sucker = crate::Sucker::new(request_tx, response_rx); let sucker = crate::sync::channel::Sucker::new(request_tx, response_rx);
let sourcer = crate::Sourcer::new(request_rx, response_tx, state); let sourcer = crate::sync::channel::Sourcer::new(request_rx, response_tx, state);
(sucker, sourcer) (sucker, sourcer)
} }
@ -236,7 +238,7 @@ mod tests {
let state = Arc::new(crate::types::ValueSource::None); let state = Arc::new(crate::types::ValueSource::None);
let state = ArcSwap::new(state); let state = ArcSwap::new(state);
let sourcer = crate::Sourcer::new(request_rx, response_tx, state); let sourcer = crate::sync::channel::Sourcer::new(request_rx, response_tx, state);
sourcer.set_static(42).unwrap(); sourcer.set_static(42).unwrap();
let producer_handle = thread::spawn(move || sourcer.run().unwrap()); let producer_handle = thread::spawn(move || sourcer.run().unwrap());
@ -254,7 +256,7 @@ mod tests {
let state = Arc::new(crate::types::ValueSource::None); let state = Arc::new(crate::types::ValueSource::None);
let state = ArcSwap::new(state); let state = ArcSwap::new(state);
let sourcer = crate::Sourcer::new(request_rx, response_tx, state); let sourcer = crate::sync::channel::Sourcer::new(request_rx, response_tx, state);
sourcer.run().unwrap(); sourcer.run().unwrap();
} }