Merge pull request #5 from tyranron/support-futures

Support Future and Stream traits

Author: thk1

.github/workflows/rust.yml

@@ -10,6 +10,6 @@ jobs:
     steps:
     - uses: actions/checkout@v1
     - name: Build
-      run: cargo build --verbose
+      run: cargo build --all-features --verbose
     - name: Run tests
-      run: cargo test --verbose
+      run: cargo test --all-features --verbose

Cargo.toml

@@ -1,6 +1,7 @@
 [package]
 name = "send_wrapper"
-version = "0.4.0"
+version = "0.5.0"
+edition = "2018"
 authors = ["Thomas Keh"]
 license = "MIT/Apache-2.0"
 description = """
@@ -13,3 +14,13 @@ readme = "README.md"
 repository = "https://github.com/thk1/send_wrapper"
 documentation = "https://docs.rs/send_wrapper"
 categories = ["rust-patterns"]
+
+[features]
+futures = ["futures-core"]
+
+[dependencies]
+futures-core = { version = "0.3", optional = true }
+
+[dev-dependencies]
+futures-executor = "0.3"
+futures-util = "0.3"

README.md

@@ -17,6 +17,9 @@ can leave the main thread at all, like required in the given
 
 [Documentation](https://docs.rs/send_wrapper)
 
+
+
+
 # Examples
 
 ```rust
@@ -64,12 +67,48 @@ let value = wrapped_value.deref();
 // let mut value: &mut NonSendType = &mut wrapped_value;
 ```
 
+
+## Using with `Future`
+
+To use `SendWrapper` on `Future`s, you should enable `futures` Cargo feature first:
+```toml
+send_wrapper = { version = "0.5", features = ["futures"] }
+```
+
+And then, just wrap your `Future` (or `Stream`):
+```rust
+use futures::{executor, future::{self, BoxFuture}};
+use send_wrapper::SendWrapper;
+
+// `Rc` is a `!Send` type,
+let value = Rc::new(42);
+// so this `Future` is `!Send` too as increments `Rc`'s inner counter.
+let future = future::lazy(|_| value.clone());
+
+// We now wrap the `future` with `SendWrapper` (value is moved inside),
+let wrapped_future = SendWrapper::new(future);
+// so now it's `Send` + `Sync` (`BoxFuture` trait object contains `Send` requirement).
+let boxed_future: BoxFuture<_> = Box::pin(wrapped_future);
+
+let t = thread::spawn(move || {
+	// This would panic (because `future` is polled in wrong thread):
+	// executor::block_on(boxed_future)
+});
+```
+
+
+
+
 # License
 
 `send_wrapper` is distributed under the terms of both the MIT license and the Apache License (Version 2.0).
 
 See LICENSE-APACHE, and LICENSE-MIT for details.
 
+
+
+
+
 [Rust]: https://www.rust-lang.org
 [`Send`]: https://doc.rust-lang.org/std/marker/trait.Send.html
 [`gtk-rs`]: http://gtk-rs.org/

src/futures.rs

@@ -0,0 +1,106 @@
+//! [`Future`] and [`Stream`] support for [`SendWrapper`].
+//!
+//! [`Future`]: std::future::Future
+//! [`Stream`]: futures_core::Stream
+
+use std::{
+	future::Future,
+	ops::{Deref as _, DerefMut as _},
+	pin::Pin,
+	task,
+};
+
+use futures_core::Stream;
+
+use super::SendWrapper;
+
+const POLL_ERROR: &'static str =
+	"Polling SendWrapper<T> variable from a thread different to the one it has been created with.";
+
+impl<F: Future> Future for SendWrapper<F> {
+	type Output = F::Output;
+
+	/// Polls this [`SendWrapper`] [`Future`].
+	///
+	/// # Panics
+	/// Polling panics if it is done from a different thread than the one the [`SendWrapper`]
+	/// instance has been created with.
+	fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
+		if !self.valid() {
+			panic!(POLL_ERROR);
+		}
+		// This is safe as `SendWrapper` itself points to the inner `Future`.
+		// So, as long as `SendWrapper` is pinned, the inner `Future` is pinned too.
+		unsafe { self.map_unchecked_mut(Self::deref_mut) }.poll(cx)
+	}
+}
+
+impl<S: Stream> Stream for SendWrapper<S> {
+	type Item = S::Item;
+
+	/// Polls this [`SendWrapper`] [`Stream`].
+	///
+	/// # Panics
+	/// Polling panics if it is done from a different thread than the one the [`SendWrapper`]
+	/// instance has been created with.
+	fn poll_next(
+		self: Pin<&mut Self>,
+		cx: &mut task::Context<'_>,
+	) -> task::Poll<Option<Self::Item>> {
+		if !self.valid() {
+			panic!(POLL_ERROR);
+		}
+		// This is safe as `SendWrapper` itself points to the inner `Stream`.
+		// So, as long as `SendWrapper` is pinned, the inner `Stream` is pinned too.
+		unsafe { self.map_unchecked_mut(Self::deref_mut) }.poll_next(cx)
+	}
+
+	#[inline]
+	fn size_hint(&self) -> (usize, Option<usize>) {
+		self.deref().size_hint()
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use std::thread;
+
+	use futures_executor as executor;
+	use futures_util::{future, stream, StreamExt};
+
+	use crate::SendWrapper;
+
+	#[test]
+	fn test_future() {
+		let w1 = SendWrapper::new(future::ready(42));
+		let w2 = w1.clone();
+		assert_eq!(
+			format!("{:?}", executor::block_on(w1)),
+			format!("{:?}", executor::block_on(w2)),
+		);
+	}
+
+	#[test]
+	fn test_future_panic() {
+		let w = SendWrapper::new(future::ready(42));
+		let t = thread::spawn(move || executor::block_on(w));
+		assert!(t.join().is_err());
+	}
+
+	#[test]
+	fn test_stream() {
+		let mut w1 = SendWrapper::new(stream::once(future::ready(42)));
+		let mut w2 = SendWrapper::new(stream::once(future::ready(42)));
+		assert_eq!(
+			format!("{:?}", executor::block_on(w1.next())),
+			format!("{:?}", executor::block_on(w2.next())),
+		);
+	}
+
+	#[test]
+	fn test_stream_panic() {
+		let mut w = SendWrapper::new(stream::once(future::ready(42)));
+		let t = thread::spawn(move || executor::block_on(w.next()));
+		assert!(t.join().is_err());
+	}
+}

src/lib.rs

@@ -80,6 +80,9 @@
 //! [`GTK+`]: https://www.gtk.org/
 //! [using `Glib`]: http://gtk-rs.org/docs/glib/source/fn.idle_add.html
 
+#[cfg(feature = "futures")]
+pub mod futures;
+
 use std::fmt;
 use std::marker::Send;
 use std::ops::{Deref, DerefMut, Drop};
@@ -232,13 +235,13 @@ impl<T: Clone> Clone for SendWrapper<T> {
 
 #[cfg(test)]
 mod tests {
-
 	use std::ops::Deref;
 	use std::rc::Rc;
 	use std::sync::mpsc::channel;
 	use std::sync::Arc;
 	use std::thread;
-	use SendWrapper;
+
+	use super::SendWrapper;
 
 	#[test]
 	fn test_deref() {