send_wrapper 0.6.0

This Rust library implements a wrapper type called SendWrapper which allows you to move around non-Send types between threads, as long as you access the contained value only from within the original thread. You also have to make sure that the wrapper is dropped from within the original thread. If any of these constraints is violated, a panic occurs.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99

//! [`Future`] and [`Stream`] support for [`SendWrapper`].
use std::{
	future::Future,
	ops::{Deref as _, DerefMut as _},
	pin::Pin,
	task,
};

use futures_core::Stream;

use crate::SendWrapper;

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.
	#[track_caller]
	fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
		self.assert_valid_for_poll();
		// 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.
	#[track_caller]
	fn poll_next(
		self: Pin<&mut Self>,
		cx: &mut task::Context<'_>,
	) -> task::Poll<Option<Self::Item>> {
		self.assert_valid_for_poll();
		// 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());
	}
}