Compare commits

..

1 commit

Author SHA1 Message Date
Callum Leslie
c9c0216b96
chore: release v0.0.3 2025-09-16 08:27:26 +01:00
7 changed files with 26 additions and 71 deletions

View file

@ -6,7 +6,7 @@ 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.3] - 2025-10-14
## [0.0.3] - 2025-09-16
### 🚀 Features

View file

@ -17,7 +17,6 @@ exclude = ["flake.nix", "flake.lock", ".envrc", "cliff.toml", "release-plz.toml"
thiserror = "2.0"
flume = { version = "0.11", optional = true }
crossbeam-channel = { version = "0.5", optional = true }
arc-swap = "1.7.1"
[features]
default = ["all"]

View file

@ -1,5 +1,4 @@
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use crate::error::Error;
use crate::sync::traits::{ChannelReceiver, ChannelSender};
@ -63,47 +62,31 @@ where
impl<T, SR, ST> Sourcer<T, SR, ST>
where
T: Send + 'static,
T: Clone + Send + 'static,
SR: ChannelReceiver<Request>,
ST: ChannelSender<Response<T>>,
{
/// Set a fixed value
pub fn set_static(&self, val: T) -> Result<(), Error>
where
T: Clone,
{
self.state.swap(Arc::new(ValueSource::Static {
val,
clone: T::clone,
}));
pub fn set_static(&self, value: T) -> Result<(), Error> {
let mut state = self.state.lock().map_err(|_| Error::InternalError)?;
*state = ValueSource::Static(value);
Ok(())
}
/// Set a closure that implements [Fn]
/// Set a closure
pub fn set<F>(&self, closure: F) -> Result<(), Error>
where
F: Fn() -> T + Send + Sync + 'static,
{
self.state
.swap(Arc::new(ValueSource::Dynamic(Box::new(closure))));
Ok(())
}
/// 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,
)))));
let mut state = self.state.lock().map_err(|_| Error::InternalError)?;
*state = ValueSource::Dynamic(Box::new(closure));
Ok(())
}
/// Close the channel
pub fn close(&self) -> Result<(), Error> {
self.state.swap(Arc::new(ValueSource::Cleared));
let mut state = self.state.lock().map_err(|_| Error::InternalError)?;
*state = ValueSource::Cleared;
Ok(())
}
@ -120,7 +103,8 @@ where
}
Ok(Request::Close) => {
// Close channel
self.close()?;
let mut state = self.state.lock().map_err(|_| Error::InternalError)?;
*state = ValueSource::Cleared;
break;
}
Err(_) => {
@ -133,26 +117,12 @@ where
}
fn handle_get_value(&self) -> Result<Response<T>, Error> {
let state = self.state.load();
let state = self.state.lock().map_err(|_| Error::InternalError)?;
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
}
}
match &*state {
ValueSource::Static(value) => Ok(Response::Value(value.clone())),
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);
let value = self.execute_closure_safely(closure);
match value {
Ok(v) => Ok(Response::Value(v)),
Err(_) => Ok(Response::NoSource), // Closure execution failed
@ -165,7 +135,7 @@ where
fn execute_closure_safely(
&self,
closure: &mut dyn FnMut() -> T,
closure: &dyn Fn() -> T,
) -> Result<T, Box<dyn std::any::Any + Send>> {
std::panic::catch_unwind(std::panic::AssertUnwindSafe(closure))
}

View file

@ -1,9 +1,6 @@
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> =
@ -65,10 +62,10 @@ impl<T> CrossbeamSuck<T> {
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 state = std::sync::Arc::new(std::sync::Mutex::new(crate::types::ValueSource::None));
let sucker = crate::Sucker::new(request_tx, response_rx);
let sourcer = crate::Sourcer::new(request_rx, response_tx, state);
let sourcer = crate::Sourcer::new(request_rx, response_tx, std::sync::Arc::clone(&state));
(sucker, sourcer)
}

View file

@ -1,9 +1,6 @@
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> =
@ -65,11 +62,10 @@ impl<T> FlumeSuck<T> {
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 state = std::sync::Arc::new(std::sync::Mutex::new(crate::types::ValueSource::None));
let sucker = crate::Sucker::new(request_tx, response_rx);
let sourcer = crate::Sourcer::new(request_rx, response_tx, state);
let sourcer = crate::Sourcer::new(request_rx, response_tx, std::sync::Arc::clone(&state));
(sucker, sourcer)
}

View file

@ -1,8 +1,5 @@
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;
@ -63,11 +60,10 @@ impl<T> StdSuck<T> {
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 state = std::sync::Arc::new(std::sync::Mutex::new(crate::types::ValueSource::None));
let sucker = crate::Sucker::new(request_tx, response_rx);
let sourcer = crate::Sourcer::new(request_rx, response_tx, state);
let sourcer = crate::Sourcer::new(request_rx, response_tx, std::sync::Arc::clone(&state));
(sucker, sourcer)
}

View file

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