Merge pull request #546 from tmccombs/unsync-stream-spawn

Unsync stream spawn

Author: alexcrichton

src/lib.rs

@@ -200,6 +200,8 @@ pub use future::{
 mod lock;
 mod task_impl;
 
+mod resultstream;
+
 pub mod task;
 pub mod executor;
 #[cfg(feature = "use_std")]

src/resultstream.rs

@@ -0,0 +1,46 @@
+// This should really be in the the stream module,
+// but `pub(crate)` isn't available until Rust 1.18,
+// and pre-1.18 there isn't a really good way to have a sub-module
+// available to the crate, but not without it.
+use core::marker::PhantomData;
+
+use {Poll, Async};
+use stream::Stream;
+
+
+/// A stream combinator used to convert a `Stream<Item=T,Error=E>`
+/// to a `Stream<Item=Result<T,E>>`.
+///
+/// A poll on this stream will never return an `Err`. As such the
+/// actual error type is parameterized, so it can match whatever error
+/// type is needed.
+///
+/// This structure is produced by the `Sream::results` method.
+#[derive(Debug)]
+#[must_use = "streams do nothing unless polled"]
+pub struct Results<S: Stream, E> {
+    inner: S,
+    phantom: PhantomData<E>
+}
+
+pub fn new<S, E>(s: S) -> Results<S, E> where S: Stream {
+    Results {
+        inner: s,
+        phantom: PhantomData
+    }
+}
+
+impl<S: Stream, E> Stream for Results<S, E> {
+    type Item = Result<S::Item, S::Error>;
+    type Error = E;
+
+    fn poll(&mut self) -> Poll<Option<Result<S::Item, S::Error>>, E> {
+        match self.inner.poll() {
+            Ok(Async::Ready(Some(item))) => Ok(Async::Ready(Some(Ok(item)))),
+            Err(e) => Ok(Async::Ready(Some(Err(e)))),
+            Ok(Async::Ready(None)) => Ok(Async::Ready(None)),
+            Ok(Async::NotReady) => Ok(Async::NotReady)
+        }
+    }
+}
+

src/sync/mpsc/mod.rs

@@ -80,6 +80,7 @@ use sync::mpsc::queue::{Queue, PopResult};
 use task::{self, Task};
 use future::Executor;
 use sink::SendAll;
+use resultstream::{self, Results};
 use {Async, AsyncSink, Future, Poll, StartSend, Sink, Stream};
 
 mod queue;
@@ -841,11 +842,9 @@ pub struct SpawnHandle<Item, Error> {
 
 /// Type of future which `Executor` instances must be able to execute for `spawn`.
 pub struct Execute<S: Stream> {
-    inner: SendAll<Sender<Result<S::Item, S::Error>>, ResultStream<S>>
+    inner: SendAll<Sender<Result<S::Item, S::Error>>, Results<S, SendError<Result<S::Item, S::Error>>>>
 }
 
-struct ResultStream<S: Stream>(S);
-
 /// Spawns a `stream` onto the instance of `Executor` provided, `executor`,
 /// returning a handle representing the remote stream.
 ///
@@ -869,7 +868,7 @@ pub fn spawn<S, E>(stream: S, executor: &E, buffer: usize) -> SpawnHandle<S::Ite
 {
     let (tx, rx) = channel(buffer);
     executor.execute(Execute {
-        inner: tx.send_all(ResultStream(stream))
+        inner: tx.send_all(resultstream::new(stream))
     }).expect("failed to spawn stream");
     SpawnHandle {
         rx: rx
@@ -900,7 +899,7 @@ pub fn spawn_unbounded<S, E>(stream: S, executor: &E) -> SpawnHandle<S::Item, S:
 {
     let (tx, rx) = channel2(None);
     executor.execute(Execute {
-        inner: tx.send_all(ResultStream(stream))
+        inner: tx.send_all(resultstream::new(stream))
     }).expect("failed to spawn stream");
     SpawnHandle {
         rx: rx
@@ -948,19 +947,6 @@ impl<S: Stream> fmt::Debug for Execute<S> {
     }
 }
 
-impl<S: Stream> Stream for ResultStream<S> {
-    type Item = Result<S::Item, S::Error>;
-    type Error = SendError<Self::Item>;
-    fn poll(&mut self) -> Poll<Option<Result<S::Item, S::Error>>, Self::Error> {
-        match self.0.poll() {
-            Ok(Async::Ready(Some(item))) => Ok(Async::Ready(Some(Ok(item)))),
-            Err(e) => Ok(Async::Ready(Some(Err(e)))),
-            Ok(Async::Ready(None)) => Ok(Async::Ready(None)),
-            Ok(Async::NotReady) => Ok(Async::NotReady)
-        }
-    }
-}
-
 /*
  *
  * ===== impl Inner =====

src/unsync/mpsc.rs

@@ -13,7 +13,10 @@ use std::mem;
 use std::rc::{Rc, Weak};
 
 use task::{self, Task};
-use {Async, AsyncSink, Poll, StartSend, Sink, Stream};
+use future::Executor;
+use sink::SendAll;
+use resultstream::{self, Results};
+use {Async, AsyncSink, Future, Poll, StartSend, Sink, Stream};
 
 /// Creates a bounded in-memory channel with buffered storage.
 ///
@@ -330,3 +333,125 @@ impl<T> SendError<T> {
         self.0
     }
 }
+
+/// Handle returned from the `spawn` function.
+///
+/// This handle is a stream that proxies a stream on a seperate `Executor`.
+/// Created through the `mpsc::spawn` function, this handle will produce
+/// the same values as the proxied stream, as they are produced in the executor,
+/// and uses a limited buffer to exert back-pressure on the remote stream.
+///
+/// If this handle is dropped, then the stream will no longer be polled and is
+/// scheduled to be dropped.
+pub struct SpawnHandle<Item, Error> {
+    inner: Receiver<Result<Item, Error>>
+}
+
+/// Type of future which `Executor` instances must be able to execut for `spawn`.
+pub struct Execute<S: Stream> {
+    inner: SendAll<Sender<Result<S::Item, S::Error>>, Results<S, SendError<Result<S::Item, S::Error>>>>
+}
+
+/// Spawns a `stream` onto the instance of `Executor` provided, `executor`,
+/// returning a handle representing the remote stream.
+///
+/// The `stream` will be canceled if the `SpawnHandle` is dropped.
+///
+/// The `SpawnHandle` returned is a stream that is a proxy for `stream` itself.
+/// When `stream` has additional items available, then the `SpawnHandle`
+/// will have those same items available.
+///
+/// At most `buffer + 1` elements will be buffered at a time. If the buffer
+/// is full, then `stream` will stop progressing until more space is available.
+/// This allows the `SpawnHandle` to exert backpressure on the `stream`.
+///
+/// # Panics
+///
+/// This function will panic if `executor` is unable spawn a `Future` containing
+/// the entirety of the `stream`.
+pub fn spawn<S, E>(stream: S, executor: &E, buffer: usize) -> SpawnHandle<S::Item, S::Error>
+    where S: Stream,
+          E: Executor<Execute<S>>
+{
+    let (tx, rx) = channel(buffer);
+    executor.execute(Execute {
+        inner: tx.send_all(resultstream::new(stream))
+    }).expect("failed to spawn stream");
+    SpawnHandle {
+        inner: rx
+    }
+}
+
+/// Spawns a `stream` onto the instance of `Executor` provided, `executor`,
+/// returning a handle representing the remote stream, with unbounded buffering.
+///
+/// The `stream` will be canceled if the `SpawnHandle` is dropped.
+///
+/// The `SpawnHandle` returned is a stream that is a proxy for `stream` itself.
+/// When `stream` has additional items available, then the `SpawnHandle`
+/// will have those same items available.
+///
+/// An unbounded buffer is used, which means that values will be buffered as
+/// fast as `stream` can produce them, without any backpressure. Therefore, if
+/// `stream` is an infinite stream, it can use an unbounded amount of memory, and
+/// potentially hog CPU resources. In particular, if `stream` is infinite
+/// and doesn't ever yield (by returnin `Async::NotReady` from `poll`), it
+/// will result in an infinite loop.
+///
+/// # Panics
+///
+/// This function will panic if `executor` is unable spawn a `Future` containing
+/// the entirety of the `stream`.
+pub fn spawn_unbounded<S,E>(stream: S, executor: &E) -> SpawnHandle<S::Item, S::Error>
+    where S: Stream,
+          E: Executor<Execute<S>>
+{
+    let (tx, rx) = channel_(None);
+    executor.execute(Execute {
+        inner: tx.send_all(resultstream::new(stream))
+    }).expect("failed to spawn stream");
+    SpawnHandle {
+        inner: rx
+    }
+}
+
+impl<I, E> Stream for SpawnHandle<I, E> {
+    type Item = I;
+    type Error = E;
+
+    fn poll(&mut self) -> Poll<Option<I>, E> {
+        match self.inner.poll() {
+            Ok(Async::Ready(Some(Ok(t)))) => Ok(Async::Ready(Some(t.into()))),
+            Ok(Async::Ready(Some(Err(e)))) => Err(e),
+            Ok(Async::Ready(None)) => Ok(Async::Ready(None)),
+            Ok(Async::NotReady) => Ok(Async::NotReady),
+            Err(_) => unreachable!("mpsc::Receiver should never return Err"),
+        }
+    }
+}
+
+impl<I, E> fmt::Debug for SpawnHandle<I, E> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("SpawnHandle")
+            .finish()
+    }
+}
+
+impl<S: Stream> Future for Execute<S> {
+    type Item = ();
+    type Error = ();
+
+    fn poll(&mut self) -> Poll<(), ()> {
+        match self.inner.poll() {
+            Ok(Async::NotReady) => Ok(Async::NotReady),
+            _ => Ok(Async::Ready(()))
+        }
+    }
+}
+
+impl<S: Stream> fmt::Debug for Execute<S> {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        f.debug_struct("Execute")
+         .finish()
+    }
+}

tests/unsync.rs

@@ -8,7 +8,7 @@ use futures::{Future, Stream, Sink, Async};
 use futures::unsync::oneshot;
 use futures::unsync::mpsc::{self, SendError};
 use futures::future::lazy;
-use futures::stream::iter;
+use futures::stream::{iter, unfold};
 
 use support::local_executor::Core;
 
@@ -112,3 +112,12 @@ fn mpsc_send_unpark() {
     core.spawn(lazy(move || { let _ = rx; Ok(()) }));
     core.run(donerx).unwrap();
 }
+
+#[test]
+fn spawn_sends_items() {
+    let core = 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]);
+}