Repository: alexcrichton/futures-rs
Pull request #585 (2017-09-19 14:45:24)

Add an `unsync` version of the `slot` channel
Also tweak the implementation slightly of both the sync/unsync versions.

cc @tailhook 


Patch:
diff --git a/src/sync/slot.rs b/src/sync/slot.rs
index f0e2813e..06e984d5 100644
--- a/src/sync/slot.rs
+++ b/src/sync/slot.rs
@@ -1,5 +1,6 @@
 //! An unbounded channel that only stores last value sent
 
+use std::mem;
 use std::sync::{Arc, Weak, Mutex};
 
 use task::{self, Task};
@@ -17,7 +18,7 @@ use {Sink, Stream, AsyncSink, Async, Poll, StartSend};
 /// other ones are discarded.
 #[derive(Debug)]
 pub struct Sender<T> {
-    inner: Option<Weak<Mutex<Inner<T>>>>,
+    inner: Weak<Mutex<Inner<T>>>,
 }
 
 /// The receiving end of a channel which preserves only the last value
@@ -45,14 +46,29 @@ impl AssertKindsReceiver for Receiver<u32> {}
 
 impl<T> Sender<T> {
     /// Sets the new new value of the stream and notifies the consumer if any
+    ///
+    /// This function will store the `value` provided as the current value for
+    /// htis channel, replacing any previous value that may have been there. If
+    /// the receiver may still be able to receive this message, then `Ok` is
+    /// returned with the previous value that was in this channel.
+    ///
+    /// If `Ok(Some)` is returned then this value overwrote a previous value,
+    /// and the value was never received by the receiver. If `Ok(None)` is
+    /// returned, then no previous value was found and the `value` is queued up
+    /// to be received by the receiver.
+    ///
+    /// # Errors
+    ///
+    /// This function will return an `Err` if the receiver has gone away and
+    /// it's impossible to send this value to the receiver. The error returned
+    /// retains ownership of the `value` provided and can be extracted, if
+    /// necessary.
     pub fn swap(&self, value: T) -> Result<Option<T>, SendError<T>> {
         let result;
         // Do this step first so that the lock is dropped when
         // `unpark` is called
         let task = {
-            let strong = self.inner.as_ref()
-                .expect("sending to a closed slot");
-            if let Some(ref lock) = strong.upgrade() {
+            if let Some(ref lock) = self.inner.upgrade() {
                 let mut inner = lock.lock().unwrap();
                 result = inner.value.take();
                 inner.value = Some(value);
@@ -82,14 +98,11 @@ impl<T> Sink for Sender<T> {
         // Do this step first so that the lock is dropped *and*
         // weakref is dropped when `unpark` is called
         let task = {
-            if let Some(weak) = self.inner.take() {
-                if let Some(ref lock) = weak.upgrade() {
-                    drop(weak);
-                    let mut inner = lock.lock().unwrap();
-                    inner.task.take()
-                } else {
-                    None
-                }
+            let weak = mem::replace(&mut self.inner, Weak::new());
+            if let Some(ref lock) = weak.upgrade() {
+                drop(weak);
+                let mut inner = lock.lock().unwrap();
+                inner.task.take()
             } else {
                 None
             }
@@ -159,7 +172,7 @@ pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
         value: None,
         task: None,
     }));
-    return (Sender { inner: Some(Arc::downgrade(&inner)) },
+    return (Sender { inner: Arc::downgrade(&inner) },
             Receiver { inner: inner });
 }
 
diff --git a/src/unsync/mod.rs b/src/unsync/mod.rs
index aaa5a707..407b9725 100644
--- a/src/unsync/mod.rs
+++ b/src/unsync/mod.rs
@@ -5,3 +5,4 @@
 
 pub mod mpsc;
 pub mod oneshot;
+pub mod slot;
diff --git a/src/unsync/slot.rs b/src/unsync/slot.rs
new file mode 100644
index 00000000..6a722978
--- /dev/null
+++ b/src/unsync/slot.rs
@@ -0,0 +1,170 @@
+//! An unbounded channel that only stores last value sent
+
+use std::rc::{Rc, Weak};
+use std::cell::RefCell;
+
+use task::{self, Task};
+use {Sink, Stream, AsyncSink, Async, Poll, StartSend};
+
+/// Slot is very similar to unbounded channel but only stores last value sent
+///
+/// I.e. if you want to send some value between from producer to a consumer
+/// and if consumer is slow it should skip old values, the slot is
+/// a structure for the task.
+
+/// The transmission end of a channel which is used to send values
+///
+/// If the receiver is not fast enough only the last value is preserved and
+/// other ones are discarded.
+#[derive(Debug)]
+pub struct Sender<T> {
+    inner: Weak<RefCell<Inner<T>>>,
+}
+
+/// The receiving end of a channel which preserves only the last value
+#[derive(Debug)]
+pub struct Receiver<T> {
+    inner: Rc<RefCell<Inner<T>>>,
+}
+
+/// Error type for sending, used when the receiving end of a channel is
+/// dropped
+#[derive(Clone, PartialEq, Eq, Debug)]
+pub struct SendError<T>(T);
+
+#[derive(Debug)]
+struct Inner<T> {
+    value: Option<T>,
+    task: Option<Task>,
+}
+
+impl<T> Sender<T> {
+    /// Sets the new new value of the stream and notifies the consumer if any.
+    ///
+    /// This function will store the `value` provided as the current value for
+    /// htis channel, replacing any previous value that may have been there. If
+    /// the receiver may still be able to receive this message, then `Ok` is
+    /// returned with the previous value that was in this channel.
+    ///
+    /// If `Ok(Some)` is returned then this value overwrote a previous value,
+    /// and the value was never received by the receiver. If `Ok(None)` is
+    /// returned, then no previous value was found and the `value` is queued up
+    /// to be received by the receiver.
+    ///
+    /// # Errors
+    ///
+    /// This function will return an `Err` if the receiver has gone away and
+    /// it's impossible to send this value to the receiver. The error returned
+    /// retains ownership of the `value` provided and can be extracted, if
+    /// necessary.
+    pub fn swap(&self, value: T) -> Result<Option<T>, SendError<T>> {
+        let result;
+        // Do this step first so that the cell is dropped when
+        // `unpark` is called
+        let task = {
+            if let Some(ref cell) = self.inner.upgrade() {
+                let mut inner = cell.borrow_mut();
+                result = inner.value.take();
+                inner.value = Some(value);
+                inner.task.take()
+            } else {
+                return Err(SendError(value));
+            }
+        };
+        if let Some(task) = task {
+            task.notify();
+        }
+        return Ok(result);
+    }
+}
+
+impl<T> Sink for Sender<T> {
+    type SinkItem = T;
+    type SinkError = SendError<T>;
+    fn start_send(&mut self, item: T) -> StartSend<T, SendError<T>> {
+        self.swap(item)?;
+        Ok(AsyncSink::Ready)
+    }
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        Ok(Async::Ready(()))
+    }
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        // Do this step first so that the cell is dropped *and*
+        // weakref is dropped when `unpark` is called
+        let task = self.inner.upgrade()
+            .and_then(|inner| inner.borrow_mut().task.take());
+        self.inner = Weak::new();
+        // notify on any drop of a sender, so eventually receiver wakes up
+        // when there are no senders and closes the stream
+        if let Some(task) = task {
+            task.notify();
+        }
+        Ok(Async::Ready(()))
+    }
+}
+
+impl<T> Drop for Sender<T> {
+    fn drop(&mut self) {
+        self.close().ok();
+    }
+}
+
+impl<T> Stream for Receiver<T> {
+    type Item = T;
+    type Error = ();  // actually void
+    fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
+        let result = {
+            let mut inner = self.inner.borrow_mut();
+            if inner.value.is_none() {
+                if Rc::weak_count(&self.inner) == 0 {
+                    // no senders, terminate the stream
+                    return Ok(Async::Ready(None));
+                } else {
+                    inner.task = Some(task::current());
+                }
+            }
+            inner.value.take()
+        };
+        match result {
+            Some(value) => Ok(Async::Ready(Some(value))),
+            None => Ok(Async::NotReady),
+        }
+    }
+}
+
+/// Creates an in-memory Stream which only preserves last value
+///
+/// This method is somewhat similar to `channel(1)` but instead of preserving
+/// first value sent (and erroring on sender side) it replaces value if
+/// consumer is not fast enough and preserves last values sent on any
+/// poll of a stream.
+///
+/// # Example
+///
+/// ```
+/// use std::thread;
+/// use futures::prelude::*;
+/// use futures::stream::iter_ok;
+/// use futures::unsync::slot;
+///
+/// let (tx, rx) = slot::channel::<i32>();
+///
+/// tx.send_all(iter_ok(vec![1, 2, 3])).wait();
+///
+/// let received = rx.collect().wait().unwrap();
+/// assert_eq!(received, vec![3]);
+/// ```
+pub fn channel<T>() -> (Sender<T>, Receiver<T>) {
+    let inner = Rc::new(RefCell::new(Inner {
+        value: None,
+        task: None,
+    }));
+    return (Sender { inner: Rc::downgrade(&inner) },
+            Receiver { inner: inner });
+}
+
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Sender<T> {
+        Sender { inner: self.inner.clone() }
+    }
+}


Test patch:
diff --git a/tests/slot-unsync.rs b/tests/slot-unsync.rs
new file mode 100644
index 00000000..d216a6f0
--- /dev/null
+++ b/tests/slot-unsync.rs
@@ -0,0 +1,107 @@
+#![cfg(feature = "use_std")]
+
+extern crate futures;
+
+use futures::prelude::*;
+use futures::future::{lazy, ok};
+use futures::stream::unfold;
+use futures::unsync::slot;
+use futures::unsync::mpsc;
+
+mod support;
+use support::*;
+
+#[test]
+fn send_recv() {
+    let (tx, rx) = slot::channel::<i32>();
+    let mut rx = rx.wait();
+
+    tx.send(1).wait().unwrap();
+
+    assert_eq!(rx.next().unwrap(), Ok(1));
+}
+
+#[test]
+fn swap() {
+    let (tx, rx) = slot::channel::<i32>();
+    let mut rx = rx.wait();
+
+    assert_eq!(tx.swap(1), Ok(None));
+    assert_eq!(tx.swap(2), Ok(Some(1)));
+    assert_eq!(rx.next().unwrap(), Ok(2));
+    assert_eq!(tx.swap(3), Ok(None));
+    assert_eq!(rx.next().unwrap(), Ok(3));
+}
+
+#[test]
+fn send_recv_no_buffer() {
+    let (mut tx, mut rx) = slot::channel::<i32>();
+
+    // Run on a task context
+    lazy(move || {
+        assert!(tx.poll_complete().unwrap().is_ready());
+
+        // Send first message
+
+        let res = tx.start_send(1).unwrap();
+        assert!(is_ready(&res));
+
+        // Send second message
+        let res = tx.start_send(2).unwrap();
+        assert!(is_ready(&res));
+
+        // Take the value
+        assert_eq!(rx.poll().unwrap(), Async::Ready(Some(2)));
+
+        let res = tx.start_send(3).unwrap();
+        assert!(is_ready(&res));
+
+        // Take the value
+        assert_eq!(rx.poll().unwrap(), Async::Ready(Some(3)));
+
+        Ok::<(), ()>(())
+    }).wait().unwrap();
+}
+
+#[test]
+fn send_shared_recv() {
+    let (tx1, rx) = slot::channel::<i32>();
+    let tx2 = tx1.clone();
+    let mut rx = rx.wait();
+
+    tx1.send(1).wait().unwrap();
+    assert_eq!(rx.next().unwrap(), Ok(1));
+
+    tx2.send(2).wait().unwrap();
+    assert_eq!(rx.next().unwrap(), Ok(2));
+}
+
+#[test]
+fn tx_close_gets_none() {
+    let (_, mut rx) = slot::channel::<i32>();
+
+    // Run on a task context
+    lazy(move || {
+        assert_eq!(rx.poll(), Ok(Async::Ready(None)));
+        assert_eq!(rx.poll(), Ok(Async::Ready(None)));
+
+        Ok::<(), ()>(())
+    }).wait().unwrap();
+}
+
+#[test]
+fn spawn_sends_items() {
+    let core = local_executor::Core::new();
+    let stream = unfold(0, |i| Some(ok::<_,u8>((i, i + 1))));
+    let rx = mpsc::spawn(stream, &core, 1);
+    assert_eq!(core.run(rx.take(4).collect()).unwrap(),
+               [0, 1, 2, 3]);
+}
+
+fn is_ready<T>(res: &AsyncSink<T>) -> bool {
+    match *res {
+        AsyncSink::Ready => true,
+        _ => false,
+    }
+}
+