diff --git a/benches/channel.rs b/benches/channel.rs
index 4b4bace..b57c953 100644
--- a/benches/channel.rs
+++ b/benches/channel.rs
@@ -3,22 +3,24 @@ use futures_executor::block_on;
use futures_signals::signal::{channel, SignalExt};
fn criterion_benchmark(c: &mut Criterion) {
- c.bench_function("channel", |b| b.iter(|| {
- let (sender, receiver) = channel(black_box(0));
+ c.bench_function("channel", |b| {
+ b.iter(|| {
+ let (sender, receiver) = channel(black_box(0));
- let handle = std::thread::spawn(move || {
- block_on(receiver.for_each(|value| {
- assert!(value >= 0);
- async move {}
- }));
- });
+ let handle = std::thread::spawn(move || {
+ block_on(receiver.for_each(|value| {
+ assert!(value >= 0);
+ async move {}
+ }));
+ });
- let _ = sender.send(black_box(1));
+ let _ = sender.send(black_box(1));
- drop(sender);
+ drop(sender);
- let _ = handle.join();
- }));
+ let _ = handle.join();
+ })
+ });
}
criterion_group!(benches, criterion_benchmark);
diff --git a/src/atomic.rs b/src/atomic.rs
index 0791417..21d2235 100644
--- a/src/atomic.rs
+++ b/src/atomic.rs
@@ -1,6 +1,5 @@
use std::sync::atomic::{AtomicPtr, Ordering};
-
/*#[derive(Debug)]
pub(crate) struct Atomic {
// TODO only box if the value is too big
@@ -47,7 +46,6 @@ impl Drop for Atomic {
}
}*/
-
/// The same as `Atomic>` except faster and uses less memory.
///
/// This is because it represents `None` as a null pointer, which avoids boxing.
@@ -68,7 +66,6 @@ impl AtomicOption {
fn from_ptr(ptr: *mut A) -> Option {
if ptr.is_null() {
None
-
} else {
// This is safe because we only do this for pointers created with `Box::into_raw`
unsafe { Some(*Box::from_raw(ptr)) }
@@ -106,7 +103,9 @@ impl Drop for AtomicOption {
if !ptr.is_null() {
// This is safe because we only do this for pointers created with `Box::into_raw`
- unsafe { drop(Box::from_raw(ptr)); }
+ unsafe {
+ drop(Box::from_raw(ptr));
+ }
}
}
}
diff --git a/src/future.rs b/src/future.rs
index 6cbc88a..872fa8a 100644
--- a/src/future.rs
+++ b/src/future.rs
@@ -1,13 +1,12 @@
use std::pin::Pin;
// TODO use parking_lot ?
-use std::sync::{Arc, Weak, Mutex};
use std::future::Future;
-use std::task::{Poll, Waker, Context};
+use std::sync::{Arc, Mutex, Weak};
+use std::task::{Context, Poll, Waker};
// TODO use parking_lot ?
-use std::sync::atomic::{AtomicBool, Ordering};
use discard::{Discard, DiscardOnDrop};
use pin_project::pin_project;
-
+use std::sync::atomic::{AtomicBool, Ordering};
#[derive(Debug)]
struct CancelableFutureState {
@@ -15,7 +14,6 @@ struct CancelableFutureState {
waker: Mutex>,
}
-
#[derive(Debug)]
pub struct CancelableFutureHandle {
state: Weak,
@@ -37,7 +35,6 @@ impl Discard for CancelableFutureHandle {
}
}
-
#[pin_project(project = CancelableFutureProj)]
#[derive(Debug)]
#[must_use = "Futures do nothing unless polled"]
@@ -49,14 +46,19 @@ pub struct CancelableFuture {
}
impl Future for CancelableFuture
- where A: Future,
- B: FnOnce() -> A::Output {
-
+where
+ A: Future,
+ B: FnOnce() -> A::Output,
+{
type Output = A::Output;
// TODO should this inline ?
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll {
- let CancelableFutureProj { state, mut future, when_cancelled } = self.project();
+ let CancelableFutureProj {
+ state,
+ mut future,
+ when_cancelled,
+ } = self.project();
// TODO is this correct ?
if state.is_cancelled.load(Ordering::SeqCst) {
@@ -65,27 +67,32 @@ impl Future for CancelableFuture
let callback = when_cancelled.take().unwrap();
// TODO figure out how to call the callback immediately when discard is called, e.g. using two Arc>
Poll::Ready(callback())
-
} else {
match future.as_pin_mut().unwrap().poll(cx) {
Poll::Pending => {
// TODO is this correct ?
*state.waker.lock().unwrap() = Some(cx.waker().clone());
Poll::Pending
- },
+ }
a => a,
}
}
}
}
-
// TODO figure out a more efficient way to implement this
// TODO replace with futures_util::abortable ?
-pub fn cancelable_future(future: A, when_cancelled: B) -> (DiscardOnDrop, CancelableFuture)
- where A: Future,
- B: FnOnce() -> A::Output {
-
+pub fn cancelable_future(
+ future: A,
+ when_cancelled: B,
+) -> (
+ DiscardOnDrop,
+ CancelableFuture,
+)
+where
+ A: Future,
+ B: FnOnce() -> A::Output,
+{
let state = Arc::new(CancelableFutureState {
is_cancelled: AtomicBool::new(false),
waker: Mutex::new(None),
diff --git a/src/internal.rs b/src/internal.rs
index 3f9fcfd..fefcc72 100644
--- a/src/internal.rs
+++ b/src/internal.rs
@@ -1,10 +1,9 @@
use super::signal::Signal;
use std::pin::Pin;
// TODO use parking_lot ?
-use std::sync::{Arc, RwLock, Mutex, MutexGuard, RwLockReadGuard};
-use std::task::{Poll, Context};
use pin_project::pin_project;
-
+use std::sync::{Arc, Mutex, MutexGuard, RwLock, RwLockReadGuard};
+use std::task::{Context, Poll};
#[inline]
pub fn lock_mut(x: &Mutex) -> MutexGuard {
@@ -30,11 +29,14 @@ pub fn unwrap_ref(x: &Option) -> &A {
}
}
-
#[pin_project(project = Map2Proj)]
#[derive(Debug)]
#[must_use = "Signals do nothing unless polled"]
-pub struct Map2 where A: Signal, B: Signal {
+pub struct Map2
+where
+ A: Signal,
+ B: Signal,
+{
#[pin]
signal1: Option,
#[pin]
@@ -45,9 +47,11 @@ pub struct Map2 where A: Signal, B: Signal {
}
impl Map2
- where A: Signal,
- B: Signal,
- C: FnMut(&mut A::Item, &mut B::Item) -> D {
+where
+ A: Signal,
+ B: Signal,
+ C: FnMut(&mut A::Item, &mut B::Item) -> D,
+{
#[inline]
pub fn new(left: A, right: B, callback: C) -> Self {
Self {
@@ -81,42 +85,58 @@ impl Map2
// Pending => None => Pending => Some(right) => Pending
// Pending => None => Pending => None => Pending
impl Signal for Map2
- where A: Signal,
- B: Signal,
- C: FnMut(&mut A::Item, &mut B::Item) -> D {
+where
+ A: Signal,
+ B: Signal,
+ C: FnMut(&mut A::Item, &mut B::Item) -> D,
+{
type Item = D;
// TODO inline this ?
fn poll_change(self: Pin<&mut Self>, cx: &mut Context) -> Poll> {
- let Map2Proj { mut signal1, mut signal2, left, right, callback } = self.project();
+ let Map2Proj {
+ mut signal1,
+ mut signal2,
+ left,
+ right,
+ callback,
+ } = self.project();
let mut changed = false;
- let left_done = match signal1.as_mut().as_pin_mut().map(|signal| signal.poll_change(cx)) {
+ let left_done = match signal1
+ .as_mut()
+ .as_pin_mut()
+ .map(|signal| signal.poll_change(cx))
+ {
None => true,
Some(Poll::Ready(None)) => {
signal1.set(None);
true
- },
+ }
Some(Poll::Ready(a)) => {
*left = a;
changed = true;
false
- },
+ }
Some(Poll::Pending) => false,
};
- let right_done = match signal2.as_mut().as_pin_mut().map(|signal| signal.poll_change(cx)) {
+ let right_done = match signal2
+ .as_mut()
+ .as_pin_mut()
+ .map(|signal| signal.poll_change(cx))
+ {
None => true,
Some(Poll::Ready(None)) => {
signal2.set(None);
true
- },
+ }
Some(Poll::Ready(a)) => {
*right = a;
changed = true;
false
- },
+ }
Some(Poll::Pending) => false,
};
@@ -125,17 +145,14 @@ impl Signal for Map2
left.as_mut().unwrap(),
right.as_mut().unwrap(),
)))
-
} else if left_done && right_done {
Poll::Ready(None)
-
} else {
Poll::Pending
}
}
}
-
// TODO is it possible to avoid the Arc ?
// TODO is it possible to use only a single Mutex ?
pub type PairMut = Arc<(Mutex>, Mutex>)>;
@@ -143,7 +160,11 @@ pub type PairMut = Arc<(Mutex>, Mutex>)>;
#[pin_project(project = MapPairMutProj)]
#[derive(Debug)]
#[must_use = "Signals do nothing unless polled"]
-pub struct MapPairMut where A: Signal, B: Signal {
+pub struct MapPairMut
+where
+ A: Signal,
+ B: Signal,
+{
#[pin]
signal1: Option,
#[pin]
@@ -152,8 +173,10 @@ pub struct MapPairMut where A: Signal, B: Signal {
}
impl MapPairMut
- where A: Signal,
- B: Signal {
+where
+ A: Signal,
+ B: Signal,
+{
#[inline]
pub fn new(left: A, right: B) -> Self {
Self {
@@ -165,13 +188,19 @@ impl MapPairMut
}
impl Signal for MapPairMut
- where A: Signal,
- B: Signal {
+where
+ A: Signal,
+ B: Signal,
+{
type Item = PairMut;
// TODO inline this ?
fn poll_change(self: Pin<&mut Self>, cx: &mut Context) -> Poll> {
- let MapPairMutProj { mut signal1, mut signal2, inner } = self.project();
+ let MapPairMutProj {
+ mut signal1,
+ mut signal2,
+ inner,
+ } = self.project();
let mut changed = false;
@@ -181,47 +210,52 @@ impl Signal for MapPairMut
// TODO is it okay to move this to just above right_done ?
let mut borrow_right = inner.1.lock().unwrap();
- let left_done = match signal1.as_mut().as_pin_mut().map(|signal| signal.poll_change(cx)) {
+ let left_done = match signal1
+ .as_mut()
+ .as_pin_mut()
+ .map(|signal| signal.poll_change(cx))
+ {
None => true,
Some(Poll::Ready(None)) => {
signal1.set(None);
true
- },
+ }
Some(Poll::Ready(a)) => {
*borrow_left = a;
changed = true;
false
- },
+ }
Some(Poll::Pending) => false,
};
- let right_done = match signal2.as_mut().as_pin_mut().map(|signal| signal.poll_change(cx)) {
+ let right_done = match signal2
+ .as_mut()
+ .as_pin_mut()
+ .map(|signal| signal.poll_change(cx))
+ {
None => true,
Some(Poll::Ready(None)) => {
signal2.set(None);
true
- },
+ }
Some(Poll::Ready(a)) => {
*borrow_right = a;
changed = true;
false
- },
+ }
Some(Poll::Pending) => false,
};
if changed {
Poll::Ready(Some(inner.clone()))
-
} else if left_done && right_done {
Poll::Ready(None)
-
} else {
Poll::Pending
}
}
}
-
// TODO is it possible to avoid the Arc ?
// TODO maybe it's faster to use a Mutex ?
pub type Pair = Arc, Option)>>;
@@ -229,7 +263,11 @@ pub type Pair = Arc, Option)>>;
#[pin_project(project = MapPairProj)]
#[derive(Debug)]
#[must_use = "Signals do nothing unless polled"]
-pub struct MapPair where A: Signal, B: Signal {
+pub struct MapPair
+where
+ A: Signal,
+ B: Signal,
+{
#[pin]
signal1: Option,
#[pin]
@@ -238,8 +276,10 @@ pub struct MapPair where A: Signal, B: Signal {
}
impl MapPair
- where A: Signal,
- B: Signal {
+where
+ A: Signal,
+ B: Signal,
+{
#[inline]
pub fn new(left: A, right: B) -> Self {
Self {
@@ -251,52 +291,64 @@ impl MapPair
}
impl Signal for MapPair
- where A: Signal,
- B: Signal {
+where
+ A: Signal,
+ B: Signal,
+{
type Item = Pair;
// TODO inline this ?
fn poll_change(self: Pin<&mut Self>, cx: &mut Context) -> Poll> {
- let MapPairProj { mut signal1, mut signal2, inner } = self.project();
+ let MapPairProj {
+ mut signal1,
+ mut signal2,
+ inner,
+ } = self.project();
let mut changed = false;
let mut borrow = inner.write().unwrap();
- let left_done = match signal1.as_mut().as_pin_mut().map(|signal| signal.poll_change(cx)) {
+ let left_done = match signal1
+ .as_mut()
+ .as_pin_mut()
+ .map(|signal| signal.poll_change(cx))
+ {
None => true,
Some(Poll::Ready(None)) => {
signal1.set(None);
true
- },
+ }
Some(Poll::Ready(a)) => {
borrow.0 = a;
changed = true;
false
- },
+ }
Some(Poll::Pending) => false,
};
- let right_done = match signal2.as_mut().as_pin_mut().map(|signal| signal.poll_change(cx)) {
+ let right_done = match signal2
+ .as_mut()
+ .as_pin_mut()
+ .map(|signal| signal.poll_change(cx))
+ {
None => true,
Some(Poll::Ready(None)) => {
signal2.set(None);
true
- },
+ }
Some(Poll::Ready(a)) => {
borrow.1 = a;
changed = true;
false
- },
+ }
Some(Poll::Pending) => false,
};
if changed {
Poll::Ready(Some(inner.clone()))
-
} else if left_done && right_done {
Poll::Ready(None)
-
} else {
Poll::Pending
}
diff --git a/src/lib.rs b/src/lib.rs
index 44536b6..648822b 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -1,10 +1,9 @@
-#![recursion_limit="128"]
-#![warn(unreachable_pub)]
+//! It is *very highly* recommended to read the tutorial.
+//! It explains all of the concepts you will need to use Signals effectively.
+#![recursion_limit = "128"]
+#![warn(unreachable_pub, missing_debug_implementations, macro_use_extern_crate)]
// missing_docs
-#![deny(warnings, missing_debug_implementations, macro_use_extern_crate)]
-
-///! It is *very highly* recommended to read the tutorial.
-///! It explains all of the concepts you will need to use Signals effectively.
+#![deny(warnings)]
#[cfg(test)]
extern crate futures_executor;
@@ -15,13 +14,12 @@ extern crate futures_executor;
pub mod internal;
pub mod signal;
-pub mod signal_vec;
pub mod signal_map;
+pub mod signal_vec;
mod atomic;
mod future;
-pub use crate::future::{cancelable_future, CancelableFutureHandle, CancelableFuture};
-
+pub use crate::future::{cancelable_future, CancelableFuture, CancelableFutureHandle};
/// # Tutorial
///
diff --git a/src/signal/broadcaster.rs b/src/signal/broadcaster.rs
index 08ac4a8..6f3ef54 100644
--- a/src/signal/broadcaster.rs
+++ b/src/signal/broadcaster.rs
@@ -1,13 +1,12 @@
use super::Signal;
-use std::pin::Pin;
-use std::marker::Unpin;
+use crate::signal::ChangedWaker;
+use futures_util::task::{self, ArcWake};
use std::fmt::Debug;
-use std::sync::{Arc, Mutex, RwLock, Weak};
+use std::marker::Unpin;
+use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
-use std::task::{Poll, Waker, Context};
-use futures_util::task::{self, ArcWake};
-use crate::signal::ChangedWaker;
-
+use std::sync::{Arc, Mutex, RwLock, Weak};
+use std::task::{Context, Poll, Waker};
/// When the Signal changes it will wake up the BroadcasterNotifier, which will
/// then wake up all of the child ChangedWaker.
@@ -35,7 +34,6 @@ impl BroadcasterNotifier {
if let Some(waker) = waker.upgrade() {
waker.wake(false);
true
-
} else {
false
}
@@ -54,10 +52,12 @@ impl ArcWake for BroadcasterNotifier {
}
}
-
/// This will poll the input Signal and keep track of the most recent value.
#[derive(Debug)]
-struct BroadcasterInnerState where A: Signal {
+struct BroadcasterInnerState
+where
+ A: Signal,
+{
// TODO is there a more efficient way to implement this ?
signal: Option>>,
waker: Waker,
@@ -66,7 +66,10 @@ struct BroadcasterInnerState where A: Signal {
epoch: usize,
}
-impl BroadcasterInnerState where A: Signal {
+impl BroadcasterInnerState
+where
+ A: Signal,
+{
fn new(signal: A, waker: Waker) -> Self {
Self {
signal: Some(Box::pin(signal)),
@@ -89,14 +92,14 @@ impl BroadcasterInnerState where A: Signal {
self.value = Some(value);
changed = true;
continue;
- },
+ }
Poll::Ready(None) => {
self.signal = None;
break;
- },
+ }
Poll::Pending => {
break;
- },
+ }
}
}
@@ -107,14 +110,19 @@ impl BroadcasterInnerState where A: Signal {
}
}
-
/// Shared state for the Broadcaster and all child signals.
-struct BroadcasterSharedState where A: Signal {
+struct BroadcasterSharedState
+where
+ A: Signal,
+{
inner: RwLock>,
notifier: Arc,
}
-impl BroadcasterSharedState where A: Signal {
+impl BroadcasterSharedState
+where
+ A: Signal,
+{
fn new(signal: A) -> Self {
let notifier = Arc::new(BroadcasterNotifier::new());
let waker = task::waker(notifier.clone());
@@ -125,14 +133,16 @@ impl BroadcasterSharedState where A: Signal {
}
}
- fn poll(&self, f: F) -> B where F: FnOnce(&BroadcasterInnerState) -> B {
+ fn poll(&self, f: F) -> B
+ where
+ F: FnOnce(&BroadcasterInnerState) -> B,
+ {
if self.notifier.is_changed() {
let mut lock = self.inner.write().unwrap();
lock.poll_signal();
f(&lock)
-
} else {
let lock = self.inner.read().unwrap();
@@ -143,9 +153,10 @@ impl BroadcasterSharedState where A: Signal {
// TODO use derive
impl Debug for BroadcasterSharedState
- where A: Debug + Signal,
- A::Item: Debug {
-
+where
+ A: Debug + Signal,
+ A::Item: Debug,
+{
fn fmt(&self, fmt: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
fmt.debug_struct("BroadcasterSharedState")
.field("inner", &self.inner)
@@ -154,14 +165,19 @@ impl Debug for BroadcasterSharedState
}
}
-
-struct BroadcasterState where A: Signal {
+struct BroadcasterState
+where
+ A: Signal,
+{
epoch: usize,
waker: Arc,
shared_state: Arc>,
}
-impl BroadcasterState where A: Signal {
+impl BroadcasterState
+where
+ A: Signal,
+{
fn new(shared_state: &Arc>) -> Self {
let waker = Arc::new(ChangedWaker::new());
@@ -177,20 +193,25 @@ impl BroadcasterState where A: Signal {
}
}
- fn poll_change(&mut self, cx: &mut Context, f: F) -> Poll> where F: FnOnce(&A::Item) -> B {
- let BroadcasterState { epoch, waker, shared_state } = self;
+ fn poll_change(&mut self, cx: &mut Context, f: F) -> Poll>
+ where
+ F: FnOnce(&A::Item) -> B,
+ {
+ let BroadcasterState {
+ epoch,
+ waker,
+ shared_state,
+ } = self;
shared_state.poll(|state| {
// Value hasn't changed
if state.epoch == *epoch {
if state.signal.is_none() {
Poll::Ready(None)
-
} else {
waker.set_waker(cx);
Poll::Pending
}
-
} else {
*epoch = state.epoch;
Poll::Ready(Some(f(state.value.as_ref().unwrap())))
@@ -201,9 +222,10 @@ impl BroadcasterState where A: Signal {
// TODO use derive
impl Debug for BroadcasterState
- where A: Debug + Signal,
- A::Item: Debug {
-
+where
+ A: Debug + Signal,
+ A::Item: Debug,
+{
fn fmt(&self, fmt: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
fmt.debug_struct("BroadcasterState")
.field("epoch", &self.epoch)
@@ -226,11 +248,17 @@ impl Debug for BroadcasterState
/// If you are using a `Mutable` then you don't need `Broadcaster`, because
/// `Mutable` already supports the `.signal()`, `.signal_cloned()` and
/// `.signal_ref()` methods (they are faster than `Broadcaster`).
-pub struct Broadcaster where A: Signal {
+pub struct Broadcaster
+where
+ A: Signal,
+{
shared_state: Arc>,
}
-impl Broadcaster where A: Signal {
+impl Broadcaster
+where
+ A: Signal,
+{
/// Create a new `Broadcaster`
pub fn new(signal: A) -> Self {
Self {
@@ -240,7 +268,9 @@ impl Broadcaster where A: Signal {
#[inline]
pub fn signal_ref(&self, f: F) -> BroadcasterSignalRef
- where F: FnMut(&A::Item) -> B {
+ where
+ F: FnMut(&A::Item) -> B,
+ {
BroadcasterSignalRef {
state: BroadcasterState::new(&self.shared_state),
callback: f,
@@ -248,7 +278,11 @@ impl Broadcaster where A: Signal {
}
}
-impl Broadcaster where A: Signal, A::Item: Copy {
+impl Broadcaster
+where
+ A: Signal,
+ A::Item: Copy,
+{
/// Returns a new `Signal` which copies values from the input `Signal`
#[inline]
pub fn signal(&self) -> BroadcasterSignal {
@@ -258,7 +292,11 @@ impl Broadcaster where A: Signal, A::Item: Copy {
}
}
-impl Broadcaster where A: Signal, A::Item: Clone {
+impl Broadcaster
+where
+ A: Signal,
+ A::Item: Clone,
+{
/// Returns a new `Signal` which clones values from the input `Signal`
#[inline]
pub fn signal_cloned(&self) -> BroadcasterSignalCloned {
@@ -270,9 +308,10 @@ impl Broadcaster where A: Signal, A::Item: Clone {
// TODO use derive
impl Debug for Broadcaster
- where A: Debug + Signal,
- A::Item: Debug {
-
+where
+ A: Debug + Signal,
+ A::Item: Debug,
+{
fn fmt(&self, fmt: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
fmt.debug_struct("Broadcaster")
.field("shared_state", &self.shared_state)
@@ -280,7 +319,10 @@ impl Debug for Broadcaster
}
}
-impl Clone for Broadcaster where A: Signal {
+impl Clone for Broadcaster
+where
+ A: Signal,
+{
#[inline]
fn clone(&self) -> Self {
Self {
@@ -292,14 +334,18 @@ impl Clone for Broadcaster where A: Signal {
// ---------------------------------------------------------------------------
#[must_use = "Signals do nothing unless polled"]
-pub struct BroadcasterSignal where A: Signal {
+pub struct BroadcasterSignal
+where
+ A: Signal,
+{
state: BroadcasterState,
}
impl Signal for BroadcasterSignal
- where A: Signal,
- A::Item: Copy {
-
+where
+ A: Signal,
+ A::Item: Copy,
+{
type Item = A::Item;
#[inline]
@@ -310,9 +356,10 @@ impl Signal for BroadcasterSignal
// TODO use derive
impl Debug for BroadcasterSignal
- where A: Debug + Signal,
- A::Item: Debug {
-
+where
+ A: Debug + Signal,
+ A::Item: Debug,
+{
fn fmt(&self, fmt: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
fmt.debug_struct("BroadcasterSignal")
.field("state", &self.state)
@@ -323,14 +370,18 @@ impl Debug for BroadcasterSignal
// --------------------------------------------------------------------------
#[must_use = "Signals do nothing unless polled"]
-pub struct BroadcasterSignalCloned where A: Signal {
+pub struct BroadcasterSignalCloned
+where
+ A: Signal,
+{
state: BroadcasterState,
}
impl Signal for BroadcasterSignalCloned
- where A: Signal,
- A::Item: Clone {
-
+where
+ A: Signal,
+ A::Item: Clone,
+{
type Item = A::Item;
#[inline]
@@ -341,9 +392,10 @@ impl Signal for BroadcasterSignalCloned
// TODO use derive
impl Debug for BroadcasterSignalCloned
- where A: Debug + Signal,
- A::Item: Debug {
-
+where
+ A: Debug + Signal,
+ A::Item: Debug,
+{
fn fmt(&self, fmt: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
fmt.debug_struct("BroadcasterSignalCloned")
.field("state", &self.state)
@@ -354,7 +406,10 @@ impl Debug for BroadcasterSignalCloned
// --------------------------------------------------------------------------
#[must_use = "Signals do nothing unless polled"]
-pub struct BroadcasterSignalRef where A: Signal {
+pub struct BroadcasterSignalRef
+where
+ A: Signal,
+{
state: BroadcasterState,
callback: F,
}
@@ -362,9 +417,10 @@ pub struct BroadcasterSignalRef where A: Signal {
impl Unpin for BroadcasterSignalRef where A: Signal {}
impl Signal for BroadcasterSignalRef
- where A: Signal,
- F: FnMut(&A::Item) -> B {
-
+where
+ A: Signal,
+ F: FnMut(&A::Item) -> B,
+{
type Item = B;
#[inline]
@@ -376,9 +432,10 @@ impl Signal for BroadcasterSignalRef
// TODO use derive
impl Debug for BroadcasterSignalRef
- where A: Debug + Signal,
- A::Item: Debug {
-
+where
+ A: Debug + Signal,
+ A::Item: Debug,
+{
fn fmt(&self, fmt: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
fmt.debug_struct("BroadcasterSignalRef")
.field("state", &self.state)
diff --git a/src/signal/channel.rs b/src/signal/channel.rs
index 64c905b..708de3b 100644
--- a/src/signal/channel.rs
+++ b/src/signal/channel.rs
@@ -1,11 +1,10 @@
use super::Signal;
-use std::pin::Pin;
+use crate::atomic::AtomicOption;
use std::marker::Unpin;
-use std::sync::{Arc, Weak};
+use std::pin::Pin;
use std::sync::atomic::{AtomicUsize, Ordering};
-use std::task::{Poll, Context, Waker};
-use crate::atomic::AtomicOption;
-
+use std::sync::{Arc, Weak};
+use std::task::{Context, Poll, Waker};
#[derive(Debug)]
struct Inner {
@@ -44,7 +43,6 @@ impl Inner {
}
}
-
#[derive(Debug)]
pub struct Sender {
inner: Weak>,
@@ -57,11 +55,9 @@ impl Sender {
inner.value.store(Some(value));
inner.notify();
Ok(())
-
} else {
Err(value)
}
-
} else {
Err(value)
}
@@ -106,7 +102,6 @@ impl Drop for Sender {
}
}
-
#[derive(Debug)]
#[must_use = "Signals do nothing unless polled"]
pub struct Receiver {
@@ -125,11 +120,10 @@ impl Signal for Receiver {
if self.inner.has_senders() {
self.inner.waker.store(Some(cx.waker().clone()));
Poll::Pending
-
} else {
Poll::Ready(None)
}
- },
+ }
a => Poll::Ready(a),
}
@@ -147,9 +141,7 @@ pub fn channel(initial_value: A) -> (Sender, Receiver) {
inner: Arc::downgrade(&inner),
};
- let receiver = Receiver {
- inner,
- };
+ let receiver = Receiver { inner };
(sender, receiver)
}
diff --git a/src/signal/macros.rs b/src/signal/macros.rs
index b5869c7..4040a45 100644
--- a/src/signal/macros.rs
+++ b/src/signal/macros.rs
@@ -108,7 +108,6 @@ macro_rules! map_mut {
($($input:tt)*) => { $crate::__internal_map_split!(__internal_map_mut, (), $($input)*) };
}
-
#[doc(hidden)]
#[macro_export]
macro_rules! __internal_map_ref_pairs {
@@ -407,7 +406,6 @@ macro_rules! map_ref {
($($input:tt)*) => { $crate::__internal_map_split!(__internal_map_ref, (), $($input)*) };
}
-
#[doc(hidden)]
#[macro_export]
macro_rules! __internal_map_lets {
diff --git a/src/signal/mutable.rs b/src/signal/mutable.rs
index 37b3c08..4282c48 100644
--- a/src/signal/mutable.rs
+++ b/src/signal/mutable.rs
@@ -1,15 +1,14 @@
use super::Signal;
use std;
use std::fmt;
-use std::pin::Pin;
use std::marker::Unpin;
use std::ops::{Deref, DerefMut};
+use std::pin::Pin;
// TODO use parking_lot ?
-use std::sync::{Arc, Weak, Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard};
+use std::sync::{Arc, Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard, Weak};
// TODO use parking_lot ?
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
-use std::task::{Poll, Waker, Context};
-
+use std::task::{Context, Poll, Waker};
#[derive(Debug)]
pub(crate) struct ChangedWaker {
@@ -50,7 +49,6 @@ impl ChangedWaker {
}
}
-
#[derive(Debug)]
struct MutableLockState {
value: A,
@@ -68,7 +66,6 @@ impl MutableLockState {
if let Some(signal) = signal.upgrade() {
signal.wake(has_changed);
true
-
} else {
false
}
@@ -76,14 +73,12 @@ impl MutableLockState {
}
}
-
#[derive(Debug)]
struct MutableState {
senders: AtomicUsize,
lock: RwLock>,
}
-
#[derive(Debug)]
struct MutableSignalState {
waker: Arc,
@@ -99,17 +94,18 @@ impl MutableSignalState {
}
}
- fn poll_change(&self, cx: &mut Context, f: F) -> Poll> where F: FnOnce(&A) -> B {
+ fn poll_change(&self, cx: &mut Context, f: F) -> Poll>
+ where
+ F: FnOnce(&A) -> B,
+ {
if self.waker.is_changed() {
let value = {
let lock = self.state.lock.read().unwrap();
f(&lock.value)
};
Poll::Ready(Some(value))
-
} else if self.state.senders.load(Ordering::SeqCst) == 0 {
Poll::Ready(None)
-
} else {
self.waker.set_waker(cx);
Poll::Pending
@@ -117,9 +113,11 @@ impl MutableSignalState {
}
}
-
#[derive(Debug)]
-pub struct MutableLockMut<'a, A> where A: 'a {
+pub struct MutableLockMut<'a, A>
+where
+ A: 'a,
+{
mutated: bool,
lock: RwLockWriteGuard<'a, MutableLockState>,
}
@@ -150,9 +148,11 @@ impl<'a, A> Drop for MutableLockMut<'a, A> {
}
}
-
#[derive(Debug)]
-pub struct MutableLockRef<'a, A> where A: 'a {
+pub struct MutableLockRef<'a, A>
+where
+ A: 'a,
+{
lock: RwLockReadGuard<'a, MutableLockState>,
}
@@ -165,7 +165,6 @@ impl<'a, A> Deref for MutableLockRef<'a, A> {
}
}
-
#[repr(transparent)]
pub struct ReadOnlyMutable(Arc>);
@@ -189,7 +188,10 @@ impl ReadOnlyMutable {
}
#[inline]
- pub fn signal_ref(&self, f: F) -> MutableSignalRef where F: FnMut(&A) -> B {
+ pub fn signal_ref(&self, f: F) -> MutableSignalRef
+ where
+ F: FnMut(&A) -> B,
+ {
MutableSignalRef(self.signal_state(), f)
}
}
@@ -225,7 +227,10 @@ impl Clone for ReadOnlyMutable {
}
}
-impl fmt::Debug for ReadOnlyMutable where A: fmt::Debug {
+impl fmt::Debug for ReadOnlyMutable
+where
+ A: fmt::Debug,
+{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
let state = self.0.lock.read().unwrap();
@@ -235,7 +240,6 @@ impl fmt::Debug for ReadOnlyMutable where A: fmt::Debug {
}
}
-
#[repr(transparent)]
pub struct Mutable(ReadOnlyMutable);
@@ -271,7 +275,10 @@ impl Mutable {
value
}
- pub fn replace_with(&self, f: F) -> A where F: FnOnce(&mut A) -> A {
+ pub fn replace_with(&self, f: F) -> A
+ where
+ F: FnOnce(&mut A) -> A,
+ {
let mut state = self.state().lock.write().unwrap();
let new_value = f(&mut state.value);
@@ -301,7 +308,10 @@ impl Mutable {
state.notify(true);
}
- pub fn set_if(&self, value: A, f: F) where F: FnOnce(&A, &A) -> bool {
+ pub fn set_if(&self, value: A, f: F)
+ where
+ F: FnOnce(&A, &A) -> bool,
+ {
let mut state = self.state().lock.write().unwrap();
if f(&state.value, &value) {
@@ -337,28 +347,46 @@ impl Mutable {
}
}
-impl fmt::Debug for Mutable where A: fmt::Debug {
+impl fmt::Debug for Mutable
+where
+ A: fmt::Debug,
+{
fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
let state = self.state().lock.read().unwrap();
- fmt.debug_tuple("Mutable")
- .field(&state.value)
- .finish()
+ fmt.debug_tuple("Mutable").field(&state.value).finish()
}
}
#[cfg(feature = "serde")]
-impl serde::Serialize for Mutable where T: serde::Serialize {
+impl serde::Serialize for Mutable
+where
+ T: serde::Serialize,
+{
#[inline]
- fn serialize(&self, serializer: S) -> Result where S: serde::Serializer {
- self.state().lock.read().unwrap().value.serialize(serializer)
+ fn serialize(&self, serializer: S) -> Result
+ where
+ S: serde::Serializer,
+ {
+ self.state()
+ .lock
+ .read()
+ .unwrap()
+ .value
+ .serialize(serializer)
}
}
#[cfg(feature = "serde")]
-impl<'de, T> serde::Deserialize<'de> for Mutable where T: serde::Deserialize<'de> {
+impl<'de, T> serde::Deserialize<'de> for Mutable
+where
+ T: serde::Deserialize<'de>,
+{
#[inline]
- fn deserialize(deserializer: D) -> Result where D: serde::Deserializer<'de> {
+ fn deserialize(deserializer: D) -> Result
+ where
+ D: serde::Deserializer<'de>,
+ {
T::deserialize(deserializer).map(Mutable::new)
}
}
@@ -398,7 +426,6 @@ impl Drop for Mutable {
}
}
-
// TODO remove it from signals when it's dropped
#[derive(Debug)]
#[repr(transparent)]
@@ -415,7 +442,6 @@ impl Signal for MutableSignal {
}
}
-
// TODO remove it from signals when it's dropped
#[derive(Debug)]
#[must_use = "Signals do nothing unless polled"]
@@ -423,7 +449,10 @@ pub struct MutableSignalRef(MutableSignalState, F);
impl Unpin for MutableSignalRef {}
-impl Signal for MutableSignalRef where F: FnMut(&A) -> B {
+impl Signal for MutableSignalRef
+where
+ F: FnMut(&A) -> B,
+{
type Item = B;
fn poll_change(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll> {
@@ -434,7 +463,6 @@ impl Signal for MutableSignalRef where F: FnMut(&A) -> B {
}
}
-
// TODO it should have a single MutableSignal implementation for both Copy and Clone
// TODO remove it from signals when it's dropped
#[derive(Debug)]
diff --git a/src/signal/signal.rs b/src/signal/signal.rs
index f81eac6..ba09fd3 100644
--- a/src/signal/signal.rs
+++ b/src/signal/signal.rs
@@ -1,1477 +1,1649 @@
-use std::panic::Location;
-use std::fmt::Debug;
-use std::pin::Pin;
-use std::marker::Unpin;
-use std::future::Future;
-use std::task::{Context, Poll};
-use log::trace;
-use futures_core::stream::Stream;
-use futures_util::stream;
-use futures_util::stream::StreamExt;
-use pin_project::pin_project;
-
-use crate::internal::Map2;
-use crate::signal_vec::{VecDiff, SignalVec};
-
-
-// TODO impl for AssertUnwindSafe ?
-// TODO documentation for Signal contract:
-// * a Signal must always return Poll::Ready(Some(...)) the first time it is polled, no exceptions
-// * after the first time it can then return Poll::Ready(None) which means that the Signal is ended (i.e. there won't be any future changes)
-// * or it can return Poll::Pending, which means the Signal hasn't changed from its previous value
-// * whenever the Signal's value has changed, it must call cx.waker().wake_by_ref() which will notify the consumer that the Signal has changed
-// * If wake_by_ref() hasn't been called, then the consumer assumes that nothing has changed, so it won't re-poll the Signal
-// * unlike Streams, the consumer does not poll again if it receives Poll::Ready(Some(...)), it will only repoll if wake_by_ref() is called
-// * If the Signal returns Poll::Ready(None) then the consumer must not re-poll the Signal
-#[must_use = "Signals do nothing unless polled"]
-pub trait Signal {
- type Item;
-
- fn poll_change(self: Pin<&mut Self>, cx: &mut Context) -> Poll>;
-}
-
-
-// Copied from Future in the Rust stdlib
-impl<'a, A> Signal for &'a mut A where A: ?Sized + Signal + Unpin {
- type Item = A::Item;
-
- #[inline]
- fn poll_change(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll> {
- A::poll_change(Pin::new(&mut **self), cx)
- }
-}
-
-// Copied from Future in the Rust stdlib
-impl Signal for Box where A: ?Sized + Signal + Unpin {
- type Item = A::Item;
-
- #[inline]
- fn poll_change(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll> {
- A::poll_change(Pin::new(&mut *self), cx)
- }
-}
-
-// Copied from Future in the Rust stdlib
-impl Signal for Pin
- where A: Unpin + ::std::ops::DerefMut,
- A::Target: Signal {
- type Item = <::Target as Signal>::Item;
-
- #[inline]
- fn poll_change(self: Pin<&mut Self>, cx: &mut Context) -> Poll> {
- Pin::get_mut(self).as_mut().poll_change(cx)
- }
-}
-
-
-// TODO Seal this
-pub trait SignalExt: Signal {
- /// Creates a `Stream` which contains the values of `self`.
- ///
- /// When the output `Stream` is spawned:
- ///
- /// 1. It immediately outputs the current value of `self`.
- ///
- /// 2. Whenever `self` changes it outputs the new value of `self`.
- ///
- /// Like *all* of the `Signal` methods, `to_stream` might skip intermediate changes.
- /// So you ***cannot*** rely upon it containing every intermediate change.
- /// But you ***can*** rely upon it always containing the most recent change.
- ///
- /// # Performance
- ///
- /// This is ***extremely*** efficient: it is *guaranteed* constant time, and it does not do
- /// any heap allocation.
- #[inline]
- fn to_stream(self) -> SignalStream
- where Self: Sized {
- SignalStream {
- signal: self,
- }
- }
-
- // TODO maybe remove this ?
- #[inline]
- fn to_future(self) -> SignalFuture
- where Self: Sized {
- SignalFuture {
- signal: self,
- value: None,
- }
- }
-
- /// Creates a `Signal` which uses a closure to transform the value.
- ///
- /// When the output `Signal` is spawned:
- ///
- /// 1. It calls the closure with the current value of `self`.
- ///
- /// 2. Then it puts the return value of the closure into the output `Signal`.
- ///
- /// 3. Whenever `self` changes it repeats the above steps.
- ///
- /// This happens automatically and efficiently.
- ///
- /// It will call the closure at most once for each change in `self`.
- ///
- /// Like *all* of the `Signal` methods, `map` might skip intermediate changes.
- /// So you ***cannot*** rely upon the closure being called for every intermediate change.
- /// But you ***can*** rely upon it always being called with the most recent change.
- ///
- /// # Examples
- ///
- /// Add `1` to the value:
- ///
- /// ```rust
- /// # use futures_signals::signal::{always, SignalExt};
- /// # let input = always(1);
- /// let mapped = input.map(|value| value + 1);
- /// ```
- ///
- /// `mapped` will always contain the current value of `input`, except with `1` added to it.
- ///
- /// If `input` has the value `10`, then `mapped` will have the value `11`.
- ///
- /// If `input` has the value `5`, then `mapped` will have the value `6`, etc.
- ///
- /// ----
- ///
- /// Formatting to a `String`:
- ///
- /// ```rust
- /// # use futures_signals::signal::{always, SignalExt};
- /// # let input = always(1);
- /// let mapped = input.map(|value| format!("{}", value));
- /// ```
- ///
- /// `mapped` will always contain the current value of `input`, except formatted as a `String`.
- ///
- /// If `input` has the value `10`, then `mapped` will have the value `"10"`.
- ///
- /// If `input` has the value `5`, then `mapped` will have the value `"5"`, etc.
- ///
- /// # Performance
- ///
- /// This is ***extremely*** efficient: it is *guaranteed* constant time, and it does not do
- /// any heap allocation.
- #[inline]
- fn map(self, callback: B) -> Map
- where B: FnMut(Self::Item) -> A,
- Self: Sized {
- Map {
- signal: self,
- callback,
- }
- }
-
- #[inline]
- fn inspect(self, callback: A) -> Inspect
- where A: FnMut(&Self::Item),
- Self: Sized {
- Inspect {
- signal: self,
- callback,
- }
- }
-
- #[inline]
- fn eq(self, value: Self::Item) -> Eq
- where Self::Item: PartialEq,
- Self: Sized {
- Eq {
- signal: self,
- matches: None,
- value,
- }
- }
-
- #[inline]
- fn neq(self, value: Self::Item) -> Neq
- where Self::Item: PartialEq,
- Self: Sized {
- Neq {
- signal: self,
- matches: None,
- value,
- }
- }
-
- /// Creates a `Signal` which uses a closure to transform the value.
- ///
- /// This is exactly the same as `map`, except:
- ///
- /// 1. It calls the closure with a mutable reference to the input value.
- ///
- /// 2. If the new input value is the same as the old input value, it will ***not*** call the closure, instead
- /// it will completely ignore the new value, like as if it never happened.
- ///
- /// It uses the `PartialEq` implementation to determine whether the new value is the same as the old value.
- ///
- /// It only keeps track of the most recent value: that means that it ***won't*** call the closure for consecutive
- /// duplicates, however it ***will*** call the closure for non-consecutive duplicates.
- ///
- /// Because `dedupe_map` has the same behavior as `map`, it is useful solely as a performance optimization.
- ///
- /// # Performance
- ///
- /// The performance is the same as `map`, except with an additional call to `eq`.
- ///
- /// If the `eq` call is fast, then `dedupe_map` can be faster than `map`, because it doesn't call the closure
- /// when the new and old values are the same, and it also doesn't update any child Signals.
- ///
- /// On the other hand, if the `eq` call is slow, then `dedupe_map` is probably slower than `map`.
- #[inline]
- fn dedupe_map(self, callback: B) -> DedupeMap
- // TODO should this use & instead of &mut ?
- where B: FnMut(&mut Self::Item) -> A,
- Self::Item: PartialEq,
- Self: Sized {
- DedupeMap {
- old_value: None,
- signal: self,
- callback,
- }
- }
-
- #[inline]
- fn dedupe(self) -> Dedupe
- where Self::Item: PartialEq,
- Self: Sized {
- Dedupe {
- old_value: None,
- signal: self,
- }
- }
-
- #[inline]
- fn dedupe_cloned(self) -> DedupeCloned
- where Self::Item: PartialEq,
- Self: Sized {
- DedupeCloned {
- old_value: None,
- signal: self,
- }
- }
-
- /// Creates a `Signal` which uses a closure to asynchronously transform the value.
- ///
- /// When the output `Signal` is spawned:
- ///
- /// 1. It calls the closure with the current value of `self`.
- ///
- /// 2. The closure returns a `Future`. It waits for that `Future` to finish, and then
- /// it puts the return value of the `Future` into the output `Signal`.
- ///
- /// 3. Whenever `self` changes it repeats the above steps.
- ///
- /// It will call the closure at most once for each change in `self`.
- ///
- /// Because Signals must always have a current value, if the `Future` is not ready yet, then the
- /// output `Signal` will start with the value `None`. When the `Future` finishes it then changes
- /// to `Some`. This can be used to detect whether the `Future` has finished or not.
- ///
- /// If `self` changes before the old `Future` is finished, it will cancel the old `Future`.
- /// That means if `self` changes faster than the `Future`, then it will never output any values.
- ///
- /// Like *all* of the `Signal` methods, `map_future` might skip intermediate changes.
- /// So you ***cannot*** rely upon the closure being called for every intermediate change.
- /// But you ***can*** rely upon it always being called with the most recent change.
- ///
- /// # Examples
- ///
- /// Call an asynchronous network API whenever the input changes:
- ///
- /// ```rust
- /// # use futures_signals::signal::{always, SignalExt};
- /// # use futures_util::future::{ready, Ready};
- /// # fn call_network_api(value: u32) -> Ready<()> { ready(()) }
- /// # fn main() {
- /// # let input = always(1);
- /// #
- /// let mapped = input.map_future(|value| call_network_api(value));
- /// # }
- /// ```
- ///
- /// # Performance
- ///
- /// This is ***extremely*** efficient: it does not do any heap allocation, and it has *very* little overhead.
- ///
- /// Of course the performance will also depend upon the `Future` which is returned from the closure.
- #[inline]
- fn map_future(self, callback: B) -> MapFuture
- where A: Future,
- B: FnMut(Self::Item) -> A,
- Self: Sized {
- MapFuture {
- signal: Some(self),
- future: None,
- callback,
- first: true,
- }
- }
-
- /// Creates a `Signal` which uses a closure to filter and transform the value.
- ///
- /// When the output `Signal` is spawned:
- ///
- /// 1. The output `Signal` starts with the value `None`.
- ///
- /// 2. It calls the closure with the current value of `self`.
- ///
- /// 3. If the closure returns `Some`, then it puts the return value of the closure into the output `Signal`.
- ///
- /// 4. If the closure returns `None`, then it does nothing.
- ///
- /// 5. Whenever `self` changes it repeats steps 2 - 4.
- ///
- /// The output `Signal` will only be `None` for the initial value. After that it will always be `Some`.
- ///
- /// If the closure returns `Some` for the initial value, then the output `Signal` will never be `None`.
- ///
- /// It will call the closure at most once for each change in `self`.
- ///
- /// Like *all* of the `Signal` methods, `filter_map` might skip intermediate changes.
- /// So you ***cannot*** rely upon the closure being called for every intermediate change.
- /// But you ***can*** rely upon it always being called with the most recent change.
- ///
- /// # Examples
- ///
- /// Add `1` to the value, but only if the value is less than `5`:
- ///
- /// ```rust
- /// # use futures_signals::signal::{always, SignalExt};
- /// # let input = always(1);
- /// let mapped = input.filter_map(|value| {
- /// if value < 5 {
- /// Some(value + 1)
- ///
- /// } else {
- /// None
- /// }
- /// });
- /// ```
- ///
- /// If the initial value of `input` is `5` or greater then `mapped` will be `None`.
- ///
- /// If the current value of `input` is `5` or greater then `mapped` will keep its old value.
- ///
- /// Otherwise `mapped` will be `Some(input + 1)`.
- ///
- /// # Performance
- ///
- /// This is ***extremely*** efficient: it does not do any heap allocation, and it has *very* little overhead.
- #[inline]
- fn filter_map(self, callback: B) -> FilterMap
- where B: FnMut(Self::Item) -> Option,
- Self: Sized {
- FilterMap {
- signal: self,
- callback,
- first: true,
- }
- }
-
- /// Creates a `Signal` which delays updates until a `Future` finishes.
- ///
- /// This can be used to throttle a `Signal` so that it only updates once every X seconds.
- ///
- /// If multiple updates happen while it's being delayed, it will only output the most recent
- /// value.
- ///
- /// # Examples
- ///
- /// Wait 1 second between each update:
- ///
- /// ```rust
- /// # use core::future::Future;
- /// # use futures_signals::signal::{always, SignalExt};
- /// # fn sleep(ms: i32) -> impl Future