1036 lines
33 KiB
Rust
1036 lines
33 KiB
Rust
//! A runtime-agnostic oneshot broadcast channel for Rust.
|
|
//!
|
|
//! Copyright (c) 2025 Stephen Waits <steve@waits.net>
|
|
//!
|
|
//! This crate provides a simple way to broadcast a value once to multiple listeners.
|
|
//! It's built on `Arc<OnceLock>` and `event-listener` for efficiency and runtime agnosticism.
|
|
//!
|
|
//! # Features
|
|
//!
|
|
//! - **Simple API** - Create, announce, listen
|
|
//! - **Fast** - Built on `Arc<OnceLock>` and `event-listener`
|
|
//! - **Runtime-agnostic** - Works with tokio, async-std, smol, or any executor
|
|
//! - **Type-safe** - Can only announce once, listeners are cloneable
|
|
//! - **Lightweight** - Minimal dependencies
|
|
//! - **Thread-safe** - All types are Send + Sync when T: Send + Sync
|
|
//! - **Cancel-safe** - Async operations can be safely cancelled
|
|
//!
|
|
//! # Feature Flags
|
|
//!
|
|
//! - `default` - Includes std feature
|
|
//! - `std` - Enables std library support (required for blocking operations: `listen_blocking()`, `listen_timeout_blocking()`)
|
|
//! - `tracing` - Enables tracing integration for debugging
|
|
//!
|
|
//! # Quick Start
|
|
//!
|
|
//! ```ignore
|
|
//! use announcement::Announcement;
|
|
//!
|
|
//! #[tokio::main]
|
|
//! async fn main() {
|
|
//! // Create announcement channel
|
|
//! let (announcer, announcement) = Announcement::new();
|
|
//!
|
|
//! // Create multiple listeners
|
|
//! let listener1 = announcement.listener();
|
|
//! let listener2 = announcement.listener();
|
|
//!
|
|
//! // Spawn tasks
|
|
//! tokio::spawn(async move {
|
|
//! let value = listener1.listen().await;
|
|
//! println!("Listener 1: {:?}", value);
|
|
//! });
|
|
//!
|
|
//! tokio::spawn(async move {
|
|
//! let value = listener2.listen().await;
|
|
//! println!("Listener 2: {:?}", value);
|
|
//! });
|
|
//!
|
|
//! // Announce to all listeners
|
|
//! announcer.announce(42).unwrap();
|
|
//! }
|
|
//! ```
|
|
//!
|
|
//! # Use Cases
|
|
//!
|
|
//! - **Shutdown signals** - Notify all tasks to shut down
|
|
//! - **Configuration broadcast** - Share initialized config to all workers
|
|
//! - **Lazy initialization** - Signal when a resource is ready
|
|
//! - **Event fanout** - Broadcast a one-time event to multiple subscribers
|
|
//!
|
|
//! # Comparison to Alternatives
|
|
//!
|
|
//! Unlike `tokio::sync::broadcast`, `announcement` is:
|
|
//! - Single-shot (announce once)
|
|
//! - Runtime-agnostic
|
|
//! - Simpler API for one-time broadcasts
|
|
//!
|
|
//! # Performance
|
|
//!
|
|
//! - Announcement creation: ~200ns
|
|
//! - Listener creation: ~20ns
|
|
//! - Announce operation: ~10-40ns
|
|
//! - Listen (already announced): ~5-10ns
|
|
//!
|
|
//! For N listeners with `Arc<T>`:
|
|
//! - Memory: O(1) - single allocation of T
|
|
//! - Time: O(N) arc clones (~10ns each)
|
|
//!
|
|
//! # Thread Safety & Memory Model
|
|
//!
|
|
//! All types (`Announcement`, `Announcer`, `Listener`) are `Send + Sync` when `T: Send + Sync`.
|
|
//!
|
|
//! ## Memory Ordering Guarantees
|
|
//!
|
|
//! - **Announce → Listen**: The announce operation has a happens-before relationship with
|
|
//! all listen operations. Any writes before `announce()` are visible after `listen()` returns.
|
|
//! - **Closed Flag**: Uses SeqCst ordering to ensure visibility of the closed state across
|
|
//! all threads, including on weakly-ordered architectures (ARM, RISC-V, PowerPC).
|
|
//! - **Value Storage**: `OnceLock` provides its own synchronization guarantees.
|
|
//!
|
|
//! ## TOCTOU Protection
|
|
//!
|
|
//! All async listen operations use a double-check pattern to prevent Time-Of-Check-Time-Of-Use
|
|
//! races between checking for a value and waiting for notification.
|
|
//!
|
|
//! # Cancel Safety
|
|
//!
|
|
//! The `listen()` method is cancel-safe: dropping the future will not lose the announced value.
|
|
//! You can safely use it with `select!`, `timeout`, or any operation that might cancel the future.
|
|
//! The value remains available for subsequent `listen()` or `try_listen()` calls.
|
|
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::{Arc, OnceLock};
|
|
|
|
use event_listener::Event;
|
|
|
|
#[cfg(feature = "std")]
|
|
use event_listener::Listener as EventListenerTrait;
|
|
|
|
#[cfg(feature = "std")]
|
|
use std::time::{Duration, Instant};
|
|
|
|
/// Internal shared state for the announcement channel.
|
|
struct AnnouncementInner<T> {
|
|
/// The announced value (set exactly once).
|
|
value: OnceLock<T>,
|
|
/// Whether the announcer was closed without announcing.
|
|
closed: AtomicBool,
|
|
}
|
|
|
|
/// The main announcement channel type that creates listeners.
|
|
///
|
|
/// This is the primary interface for creating listeners that will receive
|
|
/// the announced value. It can be cloned cheaply to share across threads.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::<i32>::new();
|
|
/// let listener1 = announcement.listener();
|
|
/// let listener2 = announcement.listener();
|
|
///
|
|
/// announcer.announce(42).unwrap();
|
|
///
|
|
/// assert_eq!(listener1.try_listen(), Some(42));
|
|
/// assert_eq!(listener2.try_listen(), Some(42));
|
|
/// ```
|
|
#[must_use = "Announcement should be used to create listeners"]
|
|
#[derive(Clone)]
|
|
pub struct Announcement<T> {
|
|
inner: Arc<AnnouncementInner<T>>,
|
|
event: Arc<Event>,
|
|
}
|
|
|
|
/// The sending end of the announcement channel.
|
|
///
|
|
/// This type is NOT `Clone` - it can only be used once to ensure
|
|
/// that a value can only be announced once.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::<String>::new();
|
|
/// announcer.announce("Hello!".to_string()).unwrap();
|
|
/// ```
|
|
#[must_use = "Announcer must be used; call announce() or close()"]
|
|
pub struct Announcer<T> {
|
|
inner: Arc<AnnouncementInner<T>>,
|
|
event: Arc<Event>,
|
|
}
|
|
|
|
/// A listener that waits for an announced value.
|
|
///
|
|
/// This type is `Clone` and can be shared across multiple tasks or threads.
|
|
/// All clones will receive the same value when it's announced.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::new();
|
|
/// let listener = announcement.listener();
|
|
///
|
|
/// announcer.announce(42).unwrap();
|
|
/// assert_eq!(listener.try_listen(), Some(42));
|
|
/// ```
|
|
#[must_use = "Listener must be used; call listen() or try_listen()"]
|
|
#[derive(Clone)]
|
|
pub struct Listener<T> {
|
|
inner: Arc<AnnouncementInner<T>>,
|
|
event: Arc<Event>,
|
|
}
|
|
|
|
/// Error returned when attempting to announce a value.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::{Announcement, AnnounceError};
|
|
///
|
|
/// let (announcer, announcement) = Announcement::new();
|
|
/// announcer.announce(42).unwrap();
|
|
///
|
|
/// // Cannot announce again - would need a second announcer which can't be created
|
|
/// let err = AnnounceError::AlreadyAnnounced(99);
|
|
/// assert_eq!(err.into_inner(), 99);
|
|
/// ```
|
|
#[derive(Debug, Clone, PartialEq, Eq)]
|
|
pub enum AnnounceError<T> {
|
|
/// A value was already announced.
|
|
/// Contains the value you tried to send.
|
|
AlreadyAnnounced(T),
|
|
}
|
|
|
|
impl<T> AnnounceError<T> {
|
|
/// Extract the value from the error.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::AnnounceError;
|
|
///
|
|
/// let err = AnnounceError::AlreadyAnnounced(42);
|
|
/// assert_eq!(err.into_inner(), 42);
|
|
/// ```
|
|
pub fn into_inner(self) -> T {
|
|
match self {
|
|
AnnounceError::AlreadyAnnounced(val) => val,
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<T> std::fmt::Display for AnnounceError<T> {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
match self {
|
|
AnnounceError::AlreadyAnnounced(_) => {
|
|
write!(f, "a value has already been announced")
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<T: std::fmt::Debug> std::error::Error for AnnounceError<T> {}
|
|
|
|
/// Error returned when listening for a value.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::ListenError;
|
|
///
|
|
/// let err = ListenError::Timeout;
|
|
/// assert_eq!(err.to_string(), "timeout elapsed while waiting for announcement");
|
|
///
|
|
/// let err = ListenError::Closed;
|
|
/// assert_eq!(err.to_string(), "announcer was dropped without announcing");
|
|
/// ```
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum ListenError {
|
|
/// Timeout elapsed without receiving a value
|
|
Timeout,
|
|
/// Announcer was dropped without announcing
|
|
Closed,
|
|
}
|
|
|
|
impl std::fmt::Display for ListenError {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
match self {
|
|
ListenError::Timeout => write!(f, "timeout elapsed while waiting for announcement"),
|
|
ListenError::Closed => write!(f, "announcer was dropped without announcing"),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl std::error::Error for ListenError {}
|
|
|
|
impl<T: Clone + Send + Sync + 'static> Announcement<T> {
|
|
/// Create a new announcement channel.
|
|
///
|
|
/// Returns a tuple of `(Announcer, Announcement)`. The announcer is used to
|
|
/// broadcast the value once, and the announcement is used to create listeners.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::<i32>::new();
|
|
/// ```
|
|
pub fn new() -> (Announcer<T>, Self) {
|
|
#[cfg(feature = "tracing")]
|
|
tracing::trace!("Creating new announcement channel");
|
|
|
|
let inner = Arc::new(AnnouncementInner {
|
|
value: OnceLock::new(),
|
|
closed: AtomicBool::new(false),
|
|
});
|
|
let event = Arc::new(Event::new());
|
|
|
|
let announcer = Announcer {
|
|
inner: Arc::clone(&inner),
|
|
event: Arc::clone(&event),
|
|
};
|
|
|
|
let announcement = Self { inner, event };
|
|
|
|
(announcer, announcement)
|
|
}
|
|
|
|
/// Create a new listener for this announcement.
|
|
///
|
|
/// Can be called before or after announcing. Can be called multiple times
|
|
/// to create multiple independent listeners that will all receive the same value.
|
|
///
|
|
/// # Performance
|
|
///
|
|
/// O(1), performs 2 Arc clones (~10-20ns)
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::new();
|
|
/// let listener1 = announcement.listener();
|
|
/// let listener2 = announcement.listener();
|
|
///
|
|
/// announcer.announce(42).unwrap();
|
|
///
|
|
/// assert_eq!(listener1.try_listen(), Some(42));
|
|
/// assert_eq!(listener2.try_listen(), Some(42));
|
|
/// ```
|
|
pub fn listener(&self) -> Listener<T> {
|
|
Listener {
|
|
inner: Arc::clone(&self.inner),
|
|
event: Arc::clone(&self.event),
|
|
}
|
|
}
|
|
|
|
/// Check if a value has been announced.
|
|
///
|
|
/// This is a non-blocking, lock-free operation.
|
|
///
|
|
/// # Performance
|
|
///
|
|
/// O(1), ~5-10ns
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::<i32>::new();
|
|
/// assert!(!announcement.is_announced());
|
|
///
|
|
/// announcer.announce(42).unwrap();
|
|
/// assert!(announcement.is_announced());
|
|
/// ```
|
|
#[inline]
|
|
pub fn is_announced(&self) -> bool {
|
|
self.inner.value.get().is_some()
|
|
}
|
|
|
|
/// Check if the announcer was closed without announcing.
|
|
///
|
|
/// Returns `true` if the announcer was explicitly closed or dropped without announcing.
|
|
/// This is a non-blocking, lock-free operation.
|
|
///
|
|
/// # Performance
|
|
///
|
|
/// O(1), ~5-10ns
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::<i32>::new();
|
|
/// assert!(!announcement.is_closed());
|
|
///
|
|
/// drop(announcer);
|
|
/// assert!(announcement.is_closed());
|
|
/// ```
|
|
#[inline]
|
|
pub fn is_closed(&self) -> bool {
|
|
self.inner.closed.load(Ordering::SeqCst)
|
|
}
|
|
}
|
|
|
|
impl<T> Announcer<T> {
|
|
/// Announce a value to all listeners.
|
|
///
|
|
/// This consumes the announcer, enforcing one-time use. All waiting listeners
|
|
/// will be woken up and can retrieve the value.
|
|
///
|
|
/// # Returns
|
|
///
|
|
/// - `Ok(())` - Value successfully announced
|
|
/// - `Err(AnnounceError::AlreadyAnnounced(val))` - Already announced (unreachable)
|
|
///
|
|
/// **Note**: This error is **unreachable** with the public API because `Announcer` is not
|
|
/// `Clone` and `announce()` consumes `self`, making it impossible to call twice. The error
|
|
/// exists for soundness (required by `OnceLock::set()`) but will never occur in correct usage.
|
|
///
|
|
/// # Performance
|
|
///
|
|
/// ~10-40ns for the announce operation, plus O(N) time to wake N waiting listeners.
|
|
/// This is a non-blocking operation.
|
|
///
|
|
/// # Thread Safety
|
|
///
|
|
/// This method is thread-safe and can be called from any thread. The announced value
|
|
/// will be visible to all listeners across all threads.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::new();
|
|
/// let listener = announcement.listener();
|
|
///
|
|
/// announcer.announce(42).unwrap();
|
|
/// assert_eq!(listener.try_listen(), Some(42));
|
|
/// ```
|
|
pub fn announce(self, val: T) -> Result<(), AnnounceError<T>> {
|
|
#[cfg(feature = "tracing")]
|
|
tracing::debug!("Announcing value");
|
|
|
|
// Try to set the value (atomic operation)
|
|
// Note: AlreadyAnnounced is unreachable because Announcer is not Clone
|
|
// and announce() consumes self, making it impossible to call twice.
|
|
// The error case is required by OnceLock::set() but cannot occur in practice.
|
|
// COVERAGE: The .map_err line below is UNREACHABLE and cannot be tested.
|
|
self.inner
|
|
.value
|
|
.set(val)
|
|
.map_err(AnnounceError::AlreadyAnnounced)?;
|
|
|
|
// Wake all waiting listeners
|
|
#[cfg(feature = "tracing")]
|
|
tracing::trace!("Notifying all waiting listeners");
|
|
self.event.notify(usize::MAX);
|
|
|
|
#[cfg(feature = "tracing")]
|
|
tracing::debug!("Value announced successfully");
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Close the announcer without announcing a value.
|
|
///
|
|
/// This explicitly closes the channel, notifying all listeners that no value
|
|
/// will be announced. All waiting listeners will wake up and receive
|
|
/// `Err(ListenError::Closed)`.
|
|
///
|
|
/// **Note on `close()` vs Drop**: This method consumes `self`, so Rust's Drop
|
|
/// implementation runs immediately afterward. The Drop impl also closes the channel
|
|
/// if no value was announced, making `close()` and just dropping the announcer
|
|
/// functionally equivalent. Use `close()` for explicitness and clarity of intent.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::<i32>::new();
|
|
/// let listener = announcement.listener();
|
|
///
|
|
/// announcer.close();
|
|
/// // Listener would receive Err(ListenError::Closed)
|
|
/// ```
|
|
///
|
|
/// Dropping the announcer has the same effect:
|
|
///
|
|
/// ```
|
|
/// use announcement::{Announcement, ListenError};
|
|
///
|
|
/// let (announcer, announcement) = Announcement::<i32>::new();
|
|
/// let listener = announcement.listener();
|
|
///
|
|
/// drop(announcer); // Auto-closes via Drop impl
|
|
/// assert_eq!(listener.try_listen(), None);
|
|
/// assert_eq!(listener.is_closed(), true);
|
|
/// ```
|
|
///
|
|
/// # Mutation Testing Note
|
|
///
|
|
/// This method is marked with `#[cfg_attr(test, mutants::skip)]` because mutations
|
|
/// to this function cannot be detected by tests. The reason: `close(self)` consumes
|
|
/// self, which immediately triggers the Drop implementation. Drop performs identical
|
|
/// operations (set closed flag + notify listeners), making close() and Drop functionally
|
|
/// equivalent. Any mutation to close() is masked by Drop's safety net, which is by design.
|
|
///
|
|
/// This is not a testing gap - it reflects the intentional redundancy where Drop serves
|
|
/// as a guarantee that listeners are never left hanging, even if close() is not called.
|
|
#[cfg_attr(test, mutants::skip)]
|
|
pub fn close(self) {
|
|
#[cfg(feature = "tracing")]
|
|
tracing::debug!("Closing announcer without announcing");
|
|
|
|
// Set the closed flag
|
|
self.inner.closed.store(true, Ordering::SeqCst);
|
|
|
|
// Wake all waiting listeners
|
|
#[cfg(feature = "tracing")]
|
|
tracing::trace!("Notifying all waiting listeners of closure");
|
|
self.event.notify(usize::MAX);
|
|
}
|
|
}
|
|
|
|
impl<T> Drop for Announcer<T> {
|
|
/// Automatically close the channel when the announcer is dropped without announcing.
|
|
///
|
|
/// This ensures listeners don't wait indefinitely if the announcer is dropped.
|
|
///
|
|
/// # Behavior
|
|
///
|
|
/// If neither `announce()` nor `close()` was called, dropping the announcer will:
|
|
/// 1. Set the closed flag (with Release ordering)
|
|
/// 2. Wake all waiting listeners
|
|
/// 3. Cause future `listen()` calls to return `Err(ListenError::Closed)`
|
|
///
|
|
/// If a value was already announced, dropping does nothing (listeners already have the value).
|
|
///
|
|
/// # Example
|
|
///
|
|
/// ```
|
|
/// use announcement::{Announcement, ListenError};
|
|
///
|
|
/// let (announcer, announcement) = Announcement::<i32>::new();
|
|
/// let listener = announcement.listener();
|
|
///
|
|
/// // Announcer dropped here without calling announce() or close()
|
|
/// drop(announcer);
|
|
///
|
|
/// // Listeners are notified of closure
|
|
/// assert!(announcement.is_closed());
|
|
/// // Non-blocking check sees no value
|
|
/// assert_eq!(listener.try_listen(), None);
|
|
/// ```
|
|
///
|
|
/// # Mutation Testing Note
|
|
///
|
|
/// The Drop implementation is marked with `#[cfg_attr(test, mutants::skip)]` because
|
|
/// mutations to the guard condition cause test timeouts rather than failures:
|
|
/// - Replacing the body causes listeners to hang indefinitely
|
|
/// - Mutating the boolean logic (&&, !) causes listeners to never be notified
|
|
///
|
|
/// These timeouts actually prove the tests would catch the bugs, but at the cost
|
|
/// of 60+ second hangs per mutation. The existing drop tests provide adequate coverage.
|
|
#[cfg_attr(test, mutants::skip)]
|
|
fn drop(&mut self) {
|
|
// Only close if we haven't announced yet
|
|
if self.inner.value.get().is_none() && !self.inner.closed.load(Ordering::SeqCst) {
|
|
#[cfg(feature = "tracing")]
|
|
tracing::debug!("Announcer dropped without announcing, auto-closing");
|
|
|
|
// Set the closed flag
|
|
self.inner.closed.store(true, Ordering::SeqCst);
|
|
|
|
// Wake all waiting listeners
|
|
#[cfg(feature = "tracing")]
|
|
tracing::trace!("Notifying all waiting listeners of auto-closure");
|
|
self.event.notify(usize::MAX);
|
|
}
|
|
}
|
|
}
|
|
|
|
impl<T: Clone> Listener<T> {
|
|
/// Try to receive without waiting.
|
|
///
|
|
/// Returns `Some(value)` if a value has been announced, or `None` if not yet announced.
|
|
///
|
|
/// # Performance
|
|
///
|
|
/// O(1), ~5-10ns, non-blocking
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::new();
|
|
/// let listener = announcement.listener();
|
|
///
|
|
/// assert_eq!(listener.try_listen(), None);
|
|
///
|
|
/// announcer.announce(42).unwrap();
|
|
/// assert_eq!(listener.try_listen(), Some(42));
|
|
/// ```
|
|
#[inline]
|
|
pub fn try_listen(&self) -> Option<T> {
|
|
self.inner.value.get().cloned()
|
|
}
|
|
|
|
/// Check if the announcer was closed without announcing.
|
|
///
|
|
/// Returns `true` if the announcer was explicitly closed or dropped without announcing.
|
|
/// This is a non-blocking, lock-free operation.
|
|
///
|
|
/// # Performance
|
|
///
|
|
/// O(1), ~5-10ns
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::<i32>::new();
|
|
/// let listener = announcement.listener();
|
|
/// assert!(!listener.is_closed());
|
|
///
|
|
/// drop(announcer);
|
|
/// assert!(listener.is_closed());
|
|
/// ```
|
|
#[inline]
|
|
pub fn is_closed(&self) -> bool {
|
|
self.inner.closed.load(Ordering::SeqCst)
|
|
}
|
|
}
|
|
|
|
impl<T: Clone> Listener<T> {
|
|
/// Wait for and receive the announced value (async).
|
|
///
|
|
/// Waits until a value is announced or the announcer is closed. Multiple calls
|
|
/// return the same result.
|
|
///
|
|
/// # Returns
|
|
///
|
|
/// - `Ok(value)` - A value was announced
|
|
/// - `Err(ListenError::Closed)` - The announcer was dropped or closed without announcing
|
|
///
|
|
/// # Cancel Safety
|
|
///
|
|
/// This method is **cancel-safe**. Dropping the future (e.g., via `select!` or `timeout`)
|
|
/// will not lose the announced value. The value remains available for subsequent calls
|
|
/// to `listen()` or `try_listen()`.
|
|
///
|
|
/// You can safely use this method with:
|
|
/// - `tokio::select!` - Choose between multiple futures
|
|
/// - `tokio::time::timeout` - Add a timeout
|
|
/// - Any other cancellation mechanism
|
|
///
|
|
/// # Multiple Calls
|
|
///
|
|
/// Calling `listen()` multiple times (even after it returns) will always return the
|
|
/// same value. The value is stored in an `Arc<OnceLock>` and cloned for each listener.
|
|
///
|
|
/// # Performance
|
|
///
|
|
/// - Fast path (already announced): ~5-10ns
|
|
/// - Slow path (waiting): ~100ns + wait time
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
///
|
|
/// # #[tokio::main]
|
|
/// # async fn main() {
|
|
/// let (announcer, announcement) = Announcement::new();
|
|
/// let listener = announcement.listener();
|
|
///
|
|
/// tokio::spawn(async move {
|
|
/// announcer.announce(42).unwrap();
|
|
/// });
|
|
///
|
|
/// let value = listener.listen().await.unwrap();
|
|
/// assert_eq!(value, 42);
|
|
/// # }
|
|
/// ```
|
|
pub async fn listen(&self) -> Result<T, ListenError> {
|
|
loop {
|
|
// Fast path: value already announced
|
|
if let Some(val) = self.inner.value.get() {
|
|
#[cfg(feature = "tracing")]
|
|
tracing::trace!("Value already announced (fast path)");
|
|
return Ok(val.clone());
|
|
}
|
|
|
|
// Check if closed
|
|
if self.inner.closed.load(Ordering::SeqCst) {
|
|
#[cfg(feature = "tracing")]
|
|
tracing::trace!("Announcer was closed without announcing");
|
|
return Err(ListenError::Closed);
|
|
}
|
|
|
|
// Slow path: set up listener before waiting
|
|
#[cfg(feature = "tracing")]
|
|
tracing::trace!("Waiting for value to be announced");
|
|
|
|
let event_listener = self.event.listen();
|
|
|
|
// Check again (TOCTOU protection)
|
|
// This prevents the race where the value is set after we check
|
|
// but before we register the listener
|
|
if let Some(val) = self.inner.value.get() {
|
|
#[cfg(feature = "tracing")]
|
|
tracing::trace!("Value announced during listener setup");
|
|
return Ok(val.clone());
|
|
}
|
|
|
|
// Check closed again
|
|
if self.inner.closed.load(Ordering::SeqCst) {
|
|
#[cfg(feature = "tracing")]
|
|
tracing::trace!("Announcer closed during listener setup");
|
|
return Err(ListenError::Closed);
|
|
}
|
|
|
|
// Wait for notification
|
|
event_listener.await;
|
|
|
|
#[cfg(feature = "tracing")]
|
|
tracing::trace!("Received notification, checking for value or closure");
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "std")]
|
|
impl<T: Clone> Listener<T> {
|
|
/// Blocking wait for the announced value.
|
|
///
|
|
/// This method blocks the current thread until a value is announced or the
|
|
/// announcer is closed. FOR NON-ASYNC CONTEXTS ONLY.
|
|
///
|
|
/// **Requires**: `std` feature (enabled by default)
|
|
///
|
|
/// # Returns
|
|
///
|
|
/// - `Ok(value)` - A value was announced
|
|
/// - `Err(ListenError::Closed)` - The announcer was dropped or closed without announcing
|
|
///
|
|
/// # ⚠️ Warning: Async Context
|
|
///
|
|
/// **DO NOT** use this method in async functions or tasks. It will block the executor
|
|
/// thread, preventing other tasks from running. This can cause:
|
|
/// - Deadlocks in async runtimes
|
|
/// - Poor performance degradation
|
|
/// - Runtime warnings or panics
|
|
///
|
|
/// In async contexts, use [`listen()`](Self::listen) instead.
|
|
///
|
|
/// # When to Use
|
|
///
|
|
/// Use this method when:
|
|
/// - You're in a non-async context (regular threads, main function)
|
|
/// - You want to synchronize between threads using standard library threads
|
|
/// - You're interfacing with blocking APIs
|
|
///
|
|
/// # Thread Safety
|
|
///
|
|
/// This method is thread-safe and can be called from any thread. Multiple threads
|
|
/// can block on different listeners simultaneously.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
/// use std::thread;
|
|
/// use std::time::Duration;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::new();
|
|
/// let listener = announcement.listener();
|
|
///
|
|
/// thread::spawn(move || {
|
|
/// thread::sleep(Duration::from_millis(10));
|
|
/// announcer.announce(42).unwrap();
|
|
/// });
|
|
///
|
|
/// let value = listener.listen_blocking().unwrap();
|
|
/// assert_eq!(value, 42);
|
|
/// ```
|
|
pub fn listen_blocking(&self) -> Result<T, ListenError> {
|
|
loop {
|
|
if let Some(val) = self.inner.value.get() {
|
|
return Ok(val.clone());
|
|
}
|
|
|
|
if self.inner.closed.load(Ordering::SeqCst) {
|
|
return Err(ListenError::Closed);
|
|
}
|
|
|
|
let listener = self.event.listen();
|
|
|
|
if let Some(val) = self.inner.value.get() {
|
|
return Ok(val.clone());
|
|
}
|
|
|
|
if self.inner.closed.load(Ordering::SeqCst) {
|
|
return Err(ListenError::Closed);
|
|
}
|
|
|
|
// Block current thread
|
|
EventListenerTrait::wait(listener);
|
|
}
|
|
}
|
|
|
|
/// Blocking wait with timeout.
|
|
///
|
|
/// Returns the value if it's announced within the timeout period.
|
|
///
|
|
/// **Requires**: `std` feature (enabled by default)
|
|
///
|
|
/// # Returns
|
|
///
|
|
/// - `Ok(val)` - Value received within timeout
|
|
/// - `Err(ListenError::Timeout)` - Timeout elapsed without receiving a value
|
|
/// - `Err(ListenError::Closed)` - The announcer was dropped or closed without announcing
|
|
///
|
|
/// # ⚠️ Warning: Async Context
|
|
///
|
|
/// **DO NOT** use this method in async functions or tasks. It will block the executor
|
|
/// thread. In async contexts, use [`listen()`](Self::listen) with `tokio::time::timeout`
|
|
/// or similar timeout mechanisms instead.
|
|
///
|
|
/// # Examples
|
|
///
|
|
/// ```
|
|
/// use announcement::Announcement;
|
|
/// use std::time::Duration;
|
|
///
|
|
/// let (announcer, announcement) = Announcement::new();
|
|
/// let listener = announcement.listener();
|
|
///
|
|
/// announcer.announce(42).unwrap();
|
|
///
|
|
/// let result = listener.listen_timeout_blocking(Duration::from_secs(1));
|
|
/// assert_eq!(result, Ok(42));
|
|
/// ```
|
|
pub fn listen_timeout_blocking(&self, duration: Duration) -> Result<T, ListenError> {
|
|
let deadline = Instant::now() + duration;
|
|
|
|
loop {
|
|
if let Some(val) = self.inner.value.get() {
|
|
return Ok(val.clone());
|
|
}
|
|
|
|
if self.inner.closed.load(Ordering::SeqCst) {
|
|
return Err(ListenError::Closed);
|
|
}
|
|
|
|
let listener = self.event.listen();
|
|
|
|
if let Some(val) = self.inner.value.get() {
|
|
return Ok(val.clone());
|
|
}
|
|
|
|
if self.inner.closed.load(Ordering::SeqCst) {
|
|
return Err(ListenError::Closed);
|
|
}
|
|
|
|
let remaining = deadline.saturating_duration_since(Instant::now());
|
|
if remaining.is_zero() {
|
|
return Err(ListenError::Timeout);
|
|
}
|
|
|
|
if EventListenerTrait::wait_timeout(listener, remaining).is_none() {
|
|
return Err(ListenError::Timeout);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Send and Sync are automatically derived for our types because:
|
|
// - Arc<OnceLock<T>> is Send + Sync when T: Send + Sync
|
|
// - Arc<Event> is Send + Sync (Event guarantees this)
|
|
// No unsafe code needed!
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
/// Test: Unit test: is_announced() method
|
|
///
|
|
/// Verifies: is_announced() correctly reports announcement state
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_is_announced() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
assert!(!announcement.is_announced());
|
|
announcer.announce(42).unwrap();
|
|
assert!(announcement.is_announced());
|
|
}
|
|
|
|
/// Test: Unit test: try_listen before announce
|
|
///
|
|
/// Verifies: try_listen() returns None before announce and Some(value) after
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_try_listen_before_announce() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener = announcement.listener();
|
|
assert_eq!(listener.try_listen(), None);
|
|
announcer.announce(42).unwrap();
|
|
assert_eq!(listener.try_listen(), Some(42));
|
|
}
|
|
|
|
/// Test: Unit test: try_listen after announce
|
|
///
|
|
/// Verifies: try_listen() returns Some(value) when called after announce
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_try_listen_after_announce() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
announcer.announce(42).unwrap();
|
|
let listener = announcement.listener();
|
|
assert_eq!(listener.try_listen(), Some(42));
|
|
}
|
|
|
|
/// Test: Unit test: listener cloning
|
|
///
|
|
/// Verifies: Cloned listeners both receive the announced value
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_clone_listener() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let listener1 = announcement.listener();
|
|
let listener2 = listener1.clone();
|
|
|
|
announcer.announce(42).unwrap();
|
|
|
|
assert_eq!(listener1.try_listen(), Some(42));
|
|
assert_eq!(listener2.try_listen(), Some(42));
|
|
}
|
|
|
|
/// Test: Unit test: multiple listeners
|
|
///
|
|
/// Verifies: Multiple listeners all receive the same announced value
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_multiple_listeners() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let l1 = announcement.listener();
|
|
let l2 = announcement.listener();
|
|
let l3 = announcement.listener();
|
|
|
|
announcer.announce(42).unwrap();
|
|
|
|
assert_eq!(l1.try_listen(), Some(42));
|
|
assert_eq!(l2.try_listen(), Some(42));
|
|
assert_eq!(l3.try_listen(), Some(42));
|
|
}
|
|
|
|
/// Test: Unit test: announce with no listeners
|
|
///
|
|
/// Verifies: Announcing succeeds even when no listeners exist
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_announce_with_no_listeners() {
|
|
let (announcer, _announcement) = Announcement::<i32>::new();
|
|
// No listeners created
|
|
announcer.announce(42).unwrap(); // Should succeed
|
|
}
|
|
|
|
/// Test: Unit test: AnnounceError::into_inner()
|
|
///
|
|
/// Verifies: into_inner() extracts the value from AnnounceError
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_error_into_inner() {
|
|
let err = AnnounceError::AlreadyAnnounced(99);
|
|
assert_eq!(err.into_inner(), 99);
|
|
}
|
|
|
|
/// Test: Unit test: announcement cloning
|
|
///
|
|
/// Verifies: Cloned announcements share the same value
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_clone_announcement() {
|
|
let (announcer, announcement) = Announcement::<i32>::new();
|
|
let announcement2 = announcement.clone();
|
|
|
|
let l1 = announcement.listener();
|
|
let l2 = announcement2.listener();
|
|
|
|
announcer.announce(42).unwrap();
|
|
|
|
assert_eq!(l1.try_listen(), Some(42));
|
|
assert_eq!(l2.try_listen(), Some(42));
|
|
}
|
|
|
|
/// Test: Unit test: Arc value handling
|
|
///
|
|
/// Verifies: Arc-wrapped values work correctly with announcement
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_with_arc() {
|
|
use std::sync::Arc;
|
|
|
|
#[derive(Clone, Debug, PartialEq)]
|
|
struct Data {
|
|
value: Vec<u8>,
|
|
}
|
|
|
|
let (announcer, announcement) = Announcement::new();
|
|
let data = Arc::new(Data {
|
|
value: vec![1, 2, 3],
|
|
});
|
|
|
|
let l1 = announcement.listener();
|
|
let l2 = announcement.listener();
|
|
|
|
announcer.announce(data.clone()).unwrap();
|
|
|
|
let r1 = l1.try_listen().unwrap();
|
|
let r2 = l2.try_listen().unwrap();
|
|
|
|
assert_eq!(r1, data);
|
|
assert_eq!(r2, data);
|
|
|
|
// Verify Arc refcount is 4 (original + announcement + r1 + r2)
|
|
assert_eq!(Arc::strong_count(&r1), 4);
|
|
}
|
|
|
|
/// Test: Unit test: AnnounceError Display trait
|
|
///
|
|
/// Verifies: AnnounceError implements Display correctly
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_error_display() {
|
|
let err = AnnounceError::AlreadyAnnounced(42);
|
|
assert_eq!(err.to_string(), "a value has already been announced");
|
|
}
|
|
|
|
/// Test: Unit test: AnnounceError Display trait
|
|
///
|
|
/// Verifies: AnnounceError implements Display correctly
|
|
/// Category: unit
|
|
#[test]
|
|
fn test_listen_error_display() {
|
|
let err = ListenError::Timeout;
|
|
assert_eq!(
|
|
err.to_string(),
|
|
"timeout elapsed while waiting for announcement"
|
|
);
|
|
}
|
|
}
|