fix: remove Unpin requirement (#12)

Co-authored-by: teor <[email protected]>

Author: thomaseizinger

CHANGELOG.md

@@ -1,9 +1,9 @@
 ## Next release
 
 - Allow iterating through future and stream collections.
-  Some types may need to be [pinned](https://docs.rs/rustc-std-workspace-std/latest/std/pin/macro.pin.html)
-  before being passed to `try_push`. This is a breaking change.
-  See [PR 11](https://github.com/thomaseizinger/rust-futures-bounded/pull/11).
+  See [PR 10](https://github.com/thomaseizinger/rust-futures-bounded/pull/10),
+  [PR 11](https://github.com/thomaseizinger/rust-futures-bounded/pull/11)
+  and [PR 12](https://github.com/thomaseizinger/rust-futures-bounded/pull/12).
 
 ## 0.3.0
 

src/futures_map.rs

@@ -127,15 +127,22 @@ where
     /// This iterator returns futures in an arbitrary order, which may change.
     ///
     /// If downcasting a future to `T` fails it will be skipped in the iterator.
-    pub fn iter_of_type<T>(&self) -> impl Iterator<Item = (&ID, &T)>
+    pub fn iter_of_type<T>(&self) -> impl Iterator<Item = (&ID, Pin<&T>)>
     where
         T: 'static,
     {
         self.inner.iter().filter_map(|a| {
-            let pin = a.inner.inner.as_ref();
-            let any = Pin::into_inner(pin) as &(dyn Any + Send);
+            let pointer = a.inner.inner.as_ref().get_ref();
+
+            let any = pointer as &(dyn Any + Send);
+            // SAFETY: this returns `None` and drops a `&T`, which is safe because dropping a reference is trivial.
             let inner = any.downcast_ref::<T>()?;
-            Some((&a.tag, inner))
+
+            // Safety: The pointer is already pinned, and will remain pinned for its entire lifetime,
+            // because we return a `Pin<&T>`.
+            let pinned = unsafe { Pin::new_unchecked(inner) };
+
+            Some((&a.tag, pinned))
         })
     }
 
@@ -144,15 +151,22 @@ where
     /// This iterator returns futures in an arbitrary order, which may change.
     ///
     /// If downcasting a future to `T` fails it will be skipped in the iterator.
-    pub fn iter_mut_of_type<T>(&mut self) -> impl Iterator<Item = (&ID, &mut T)>
+    pub fn iter_mut_of_type<T>(&mut self) -> impl Iterator<Item = (&ID, Pin<&mut T>)>
     where
         T: 'static,
     {
         self.inner.iter_mut().filter_map(|a| {
             let pin = a.inner.inner.as_mut();
-            let any = Pin::into_inner(pin) as &mut (dyn Any + Send);
+
+            // Safety: We are only temporarily manipulating the pointer and pinning it again further down.
+            let pointer = unsafe { Pin::into_inner_unchecked(pin) };
+            let any = pointer as &mut (dyn Any + Send);
             let inner = any.downcast_mut::<T>()?;
-            Some((&a.tag, inner))
+
+            // Safety: The pointer is already pinned.
+            let pinned = unsafe { Pin::new_unchecked(inner) };
+
+            Some((&a.tag, pinned))
         })
     }
 }
@@ -329,7 +343,7 @@ mod tests {
         }
         assert!(!sender.iter().any(|tx| tx.is_canceled()));
 
-        for (_, rx) in futures.iter_mut_of_type::<oneshot::Receiver<()>>() {
+        for (_, mut rx) in futures.iter_mut_of_type::<oneshot::Receiver<()>>() {
             rx.close();
         }
         assert!(sender.iter().all(|tx| tx.is_canceled()));

src/futures_set.rs

@@ -1,4 +1,7 @@
-use std::task::{ready, Context, Poll};
+use std::{
+    pin::Pin,
+    task::{ready, Context, Poll},
+};
 
 use crate::{AnyFuture, BoxFuture, Delay, FuturesMap, PushError, Timeout};
 
@@ -63,7 +66,7 @@ where
     /// This iterator returns futures in an arbitrary order, which may change.
     ///
     /// If downcasting a future to `T` fails it will be skipped in the iterator.
-    pub fn iter_of_type<T>(&self) -> impl Iterator<Item = &T>
+    pub fn iter_of_type<T>(&self) -> impl Iterator<Item = Pin<&T>>
     where
         T: 'static,
     {
@@ -75,7 +78,7 @@ where
     /// This iterator returns futures in an arbitrary order, which may change.
     ///
     /// If downcasting a future to `T` fails it will be skipped in the iterator.
-    pub fn iter_mut_of_type<T>(&mut self) -> impl Iterator<Item = &mut T>
+    pub fn iter_mut_of_type<T>(&mut self) -> impl Iterator<Item = Pin<&mut T>>
     where
         T: 'static,
     {

src/futures_tuple_set.rs

@@ -1,4 +1,5 @@
 use std::collections::HashMap;
+use std::pin::Pin;
 use std::task::{ready, Context, Poll};
 
 use crate::{AnyFuture, BoxFuture, Delay, FuturesMap, PushError, Timeout};
@@ -70,7 +71,7 @@ where
     /// This iterator returns futures in an arbitrary order, which may change.
     ///
     /// If downcasting a future to `T` fails it will be skipped in the iterator.
-    pub fn iter_of_type<T>(&self) -> impl Iterator<Item = (&T, &D)>
+    pub fn iter_of_type<T>(&self) -> impl Iterator<Item = (Pin<&T>, &D)>
     where
         T: 'static,
     {
@@ -84,7 +85,7 @@ where
     /// This iterator returns futures in an arbitrary order, which may change.
     ///
     /// If downcasting a future to `T` fails it will be skipped in the iterator.
-    pub fn iter_mut_of_type<T>(&mut self) -> impl Iterator<Item = (&mut T, &mut D)>
+    pub fn iter_mut_of_type<T>(&mut self) -> impl Iterator<Item = (Pin<&mut T>, &mut D)>
     where
         T: 'static,
     {
@@ -150,7 +151,7 @@ mod tests {
         }
         assert!(!sender.iter().any(|tx| tx.is_canceled()));
 
-        for (rx, _) in futures.iter_mut_of_type::<oneshot::Receiver<()>>() {
+        for (mut rx, _) in futures.iter_mut_of_type::<oneshot::Receiver<()>>() {
             rx.close();
         }
         assert!(sender.iter().all(|tx| tx.is_canceled()));

src/lib.rs

@@ -59,15 +59,15 @@ impl<T> fmt::Debug for PushError<T> {
 impl std::error::Error for Timeout {}
 
 #[doc(hidden)]
-pub trait AnyStream: futures_util::Stream + Any + Unpin + Send {}
+pub trait AnyStream: futures_util::Stream + Any + Send {}
 
-impl<T> AnyStream for T where T: futures_util::Stream + Any + Unpin + Send {}
+impl<T> AnyStream for T where T: futures_util::Stream + Any + Send {}
 
 type BoxStream<T> = Pin<Box<dyn AnyStream<Item = T> + Send>>;
 
 #[doc(hidden)]
-pub trait AnyFuture: std::future::Future + Any + Unpin + Send {}
+pub trait AnyFuture: std::future::Future + Any + Send {}
 
-impl<T> AnyFuture for T where T: std::future::Future + Any + Unpin + Send {}
+impl<T> AnyFuture for T where T: std::future::Future + Any + Send {}
 
 type BoxFuture<T> = Pin<Box<dyn AnyFuture<Output = T> + Send>>;

src/stream_map.rs

@@ -119,15 +119,20 @@ where
     /// This iterator returns streams in an arbitrary order, which may change.
     ///
     /// If downcasting a stream to `T` fails it will be skipped in the iterator.
-    pub fn iter_of_type<T>(&self) -> impl Iterator<Item = (&ID, &T)>
+    pub fn iter_of_type<T>(&self) -> impl Iterator<Item = (&ID, Pin<&T>)>
     where
         T: 'static,
     {
         self.inner.iter().filter_map(|a| {
-            let pin = a.inner.inner.as_ref();
-            let any = Pin::into_inner(pin) as &(dyn Any + Send);
+            let pointer = a.inner.inner.as_ref().get_ref();
+
+            let any = pointer as &(dyn Any + Send);
             let inner = any.downcast_ref::<T>()?;
-            Some((&a.key, inner))
+
+            // Safety: The pointer is already pinned.
+            let pinned = unsafe { Pin::new_unchecked(inner) };
+
+            Some((&a.key, pinned))
         })
     }
 
@@ -136,15 +141,22 @@ where
     /// This iterator returns streams in an arbitrary order, which may change.
     ///
     /// If downcasting a stream to `T` fails it will be skipped in the iterator.
-    pub fn iter_mut_of_type<T>(&mut self) -> impl Iterator<Item = (&ID, &mut T)>
+    pub fn iter_mut_of_type<T>(&mut self) -> impl Iterator<Item = (&ID, Pin<&mut T>)>
     where
         T: 'static,
     {
         self.inner.iter_mut().filter_map(|a| {
             let pin = a.inner.inner.as_mut();
-            let any = Pin::into_inner(pin) as &mut (dyn Any + Send);
+
+            // Safety: We are only temporarily manipulating the pointer and pinning it again further down.
+            let pointer = unsafe { Pin::into_inner_unchecked(pin) };
+            let any = pointer as &mut (dyn Any + Send);
             let inner = any.downcast_mut::<T>()?;
-            Some((&a.key, inner))
+
+            // Safety: The pointer is already pinned.
+            let pinned = unsafe { Pin::new_unchecked(inner) };
+
+            Some((&a.key, pinned))
         })
     }
 }
@@ -355,7 +367,7 @@ mod tests {
         }
         assert!(!sender.iter().any(|tx| tx.is_closed()));
 
-        for (_, rx) in streams.iter_mut_of_type::<mpsc::Receiver<()>>() {
+        for (_, mut rx) in streams.iter_mut_of_type::<mpsc::Receiver<()>>() {
             rx.close();
         }
         assert!(sender.iter().all(|tx| tx.is_closed()));

src/stream_set.rs

@@ -1,4 +1,7 @@
-use std::task::{ready, Context, Poll};
+use std::{
+    pin::Pin,
+    task::{ready, Context, Poll},
+};
 
 use crate::{AnyStream, BoxStream, Delay, PushError, StreamMap, Timeout};
 
@@ -63,7 +66,7 @@ where
     /// This iterator returns streams in an arbitrary order, which may change.
     ///
     /// If downcasting a stream to `T` fails it will be skipped in the iterator.
-    pub fn iter_of_type<T>(&self) -> impl Iterator<Item = &T>
+    pub fn iter_of_type<T>(&self) -> impl Iterator<Item = Pin<&T>>
     where
         T: 'static,
     {
@@ -75,7 +78,7 @@ where
     /// This iterator returns streams in an arbitrary order, which may change.
     ///
     /// If downcasting a stream to `T` fails it will be skipped in the iterator.
-    pub fn iter_mut_of_type<T>(&mut self) -> impl Iterator<Item = &mut T>
+    pub fn iter_mut_of_type<T>(&mut self) -> impl Iterator<Item = Pin<&mut T>>
     where
         T: 'static,
     {