Update Future combinators to preserve Clone

This updates most `Future` combinators to preserve `Clone` when available on the input `Future`.

For motivation, imagine you have some complicated `Future` that is not `Clone` and requires `.shared()` to properly share it. Then imagine you have a library function that is meant to bundle together a bunch of combinators to fulfill some semantic purpose. That library funciton will have to call `.shared()` if it wants to try to guarantee the return `Future` is `Clone`, but this might be suboptimal if the input `Future` was already `Clone`, plus it has the ability to obfuscate and hide the `.shared()` allocation. With this change, you can instead require `Future + Clone` on the input `Future` and have a guarantee the output will be `Clone` as well.

The hold-out `Future` implementations are:
- `Remote` / `RemoteHandle` due to their use of `futures_channel::oneshot::{Sender, Receiver}`. This seems like it is by design that these cannot be `Clone`.
- `JoinAll` / `TryJoinAll` due to their use of `Stream` combinators, but also that it seems unlikely that people would expect them to offer `Clone` since they are used to performa a  potentially costly sync barrier that would probably be desired to happen only once.

For the hold-outs, the existing pattern of using `.shared()` allows for `Clone`, and follows the intended semantics of those combinators.

Some combinators that might not make the most sense to preserve `Clone`:
- `IntoStream`
- `TryFlattenStream`

If these changes make sense, I think it would also make sense to apply them to `Stream` combinators as well (although I don't see myself utilizing this property as much with them). If that is the case, these `Future` -> `Stream` combinators make sense to preserve `Clone`.

Tested:
- `cargo doc`.
- `cargo fmt`.
- `cargo test --all-features`.

Author: zjijz

futures-util/src/future/future/catch_unwind.rs

@@ -8,7 +8,7 @@ use pin_project_lite::pin_project;
 
 pin_project! {
     /// Future for the [`catch_unwind`](super::FutureExt::catch_unwind) method.
-    #[derive(Debug)]
+    #[derive(Clone, Debug)]
     #[must_use = "futures do nothing unless you `.await` or poll them"]
     pub struct CatchUnwind<Fut> {
         #[pin]

futures-util/src/future/future/flatten.rs

@@ -9,7 +9,7 @@ use pin_project_lite::pin_project;
 
 pin_project! {
     #[project = FlattenProj]
-    #[derive(Debug)]
+    #[derive(Clone, Debug)]
     pub enum Flatten<Fut1, Fut2> {
         First { #[pin] f: Fut1 },
         Second { #[pin] f: Fut2 },

futures-util/src/future/future/fuse.rs

@@ -5,7 +5,7 @@ use pin_project_lite::pin_project;
 
 pin_project! {
     /// Future for the [`fuse`](super::FutureExt::fuse) method.
-    #[derive(Debug)]
+    #[derive(Clone, Debug)]
     #[must_use = "futures do nothing unless you `.await` or poll them"]
     pub struct Fuse<Fut> {
         #[pin]

futures-util/src/future/future/map.rs

@@ -10,7 +10,7 @@ pin_project! {
     /// Internal Map future
     #[project = MapProj]
     #[project_replace = MapProjReplace]
-    #[derive(Debug)]
+    #[derive(Clone, Debug)]
     #[must_use = "futures do nothing unless you `.await` or poll them"]
     pub enum Map<Fut, F> {
         Incomplete {

futures-util/src/future/future/mod.rs

@@ -30,15 +30,15 @@ delegate_all!(
     /// Future for the [`flatten`](super::FutureExt::flatten) method.
     Flatten<F>(
         flatten::Flatten<F, <F as Future>::Output>
-    ): Debug + Future + FusedFuture + New[|x: F| flatten::Flatten::new(x)]
+    ): Clone + Debug + Future + FusedFuture + New[|x: F| flatten::Flatten::new(x)]
     where F: Future
 );
 
 delegate_all!(
     /// Stream for the [`flatten_stream`](FutureExt::flatten_stream) method.
     FlattenStream<F>(
         flatten::Flatten<F, <F as Future>::Output>
-    ): Debug + Sink + Stream + FusedStream + New[|x: F| flatten::Flatten::new(x)]
+    ): Clone + Debug + Sink + Stream + FusedStream + New[|x: F| flatten::Flatten::new(x)]
     where F: Future
 );
 
@@ -49,49 +49,49 @@ delegate_all!(
     /// Future for the [`map`](super::FutureExt::map) method.
     Map<Fut, F>(
         map::Map<Fut, F>
-    ): Debug + Future + FusedFuture + New[|x: Fut, f: F| map::Map::new(x, f)]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut, f: F| map::Map::new(x, f)]
 );
 
 delegate_all!(
     /// Stream for the [`into_stream`](FutureExt::into_stream) method.
     IntoStream<F>(
         crate::stream::Once<F>
-    ): Debug + Stream + FusedStream + New[|x: F| crate::stream::Once::new(x)]
+    ): Clone + Debug + Stream + FusedStream + New[|x: F| crate::stream::Once::new(x)]
 );
 
 delegate_all!(
     /// Future for the [`map_into`](FutureExt::map_into) combinator.
     MapInto<Fut, T>(
         Map<Fut, IntoFn<T>>
-    ): Debug + Future + FusedFuture + New[|x: Fut| Map::new(x, into_fn())]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut| Map::new(x, into_fn())]
 );
 
 delegate_all!(
     /// Future for the [`then`](FutureExt::then) method.
     Then<Fut1, Fut2, F>(
         flatten::Flatten<Map<Fut1, F>, Fut2>
-    ): Debug + Future + FusedFuture + New[|x: Fut1, y: F| flatten::Flatten::new(Map::new(x, y))]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut1, y: F| flatten::Flatten::new(Map::new(x, y))]
 );
 
 delegate_all!(
     /// Future for the [`inspect`](FutureExt::inspect) method.
     Inspect<Fut, F>(
         map::Map<Fut, InspectFn<F>>
-    ): Debug + Future + FusedFuture + New[|x: Fut, f: F| map::Map::new(x, inspect_fn(f))]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut, f: F| map::Map::new(x, inspect_fn(f))]
 );
 
 delegate_all!(
     /// Future for the [`never_error`](super::FutureExt::never_error) combinator.
     NeverError<Fut>(
         Map<Fut, OkFn<Infallible>>
-    ): Debug + Future + FusedFuture + New[|x: Fut| Map::new(x, ok_fn())]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut| Map::new(x, ok_fn())]
 );
 
 delegate_all!(
     /// Future for the [`unit_error`](super::FutureExt::unit_error) combinator.
     UnitError<Fut>(
         Map<Fut, OkFn<()>>
-    ): Debug + Future + FusedFuture + New[|x: Fut| Map::new(x, ok_fn())]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut| Map::new(x, ok_fn())]
 );
 
 #[cfg(feature = "std")]

futures-util/src/future/join.rs

@@ -20,6 +20,18 @@ impl<Fut1: Future, Fut2: Future> Join<Fut1, Fut2> {
     }
 }
 
+impl<Fut1, Fut2> Clone for Join<Fut1, Fut2>
+where
+    Fut1: Future + Clone,
+    Fut1::Output: Clone,
+    Fut2: Future + Clone,
+    Fut2::Output: Clone,
+{
+    fn clone(&self) -> Self {
+        Self { fut1: self.fut1.clone(), fut2: self.fut2.clone() }
+    }
+}
+
 impl<Fut1, Fut2> fmt::Debug for Join<Fut1, Fut2>
 where
     Fut1: Future + fmt::Debug,

futures-util/src/future/lazy.rs

@@ -4,7 +4,7 @@ use futures_core::future::{FusedFuture, Future};
 use futures_core::task::{Context, Poll};
 
 /// Future for the [`lazy`] function.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct Lazy<F> {
     f: Option<F>,

futures-util/src/future/maybe_done.rs

@@ -10,7 +10,7 @@ use futures_core::task::{Context, Poll};
 /// A future that may have completed.
 ///
 /// This is created by the [`maybe_done()`] function.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub enum MaybeDone<Fut: Future> {
     /// A not-yet-completed future
     Future(/* #[pin] */ Fut),

futures-util/src/future/poll_fn.rs

@@ -8,6 +8,7 @@ use futures_core::task::{Context, Poll};
 
 /// Future for the [`poll_fn`] function.
 #[must_use = "futures do nothing unless you `.await` or poll them"]
+#[derive(Clone)]
 pub struct PollFn<F> {
     f: F,
 }

futures-util/src/future/select.rs

@@ -6,7 +6,7 @@ use futures_core::task::{Context, Poll};
 
 /// Future for the [`select()`] function.
 #[must_use = "futures do nothing unless you `.await` or poll them"]
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct Select<A, B> {
     inner: Option<(A, B)>,
 }

futures-util/src/future/select_all.rs

@@ -8,7 +8,7 @@ use futures_core::future::Future;
 use futures_core::task::{Context, Poll};
 
 /// Future for the [`select_all`] function.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct SelectAll<Fut> {
     inner: Vec<Fut>,

futures-util/src/future/select_ok.rs

@@ -8,7 +8,7 @@ use futures_core::future::{Future, TryFuture};
 use futures_core::task::{Context, Poll};
 
 /// Future for the [`select_ok`] function.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 #[must_use = "futures do nothing unless you `.await` or poll them"]
 pub struct SelectOk<Fut> {
     inner: Vec<Fut>,

futures-util/src/future/try_future/into_future.rs

@@ -5,7 +5,7 @@ use pin_project_lite::pin_project;
 
 pin_project! {
     /// Future for the [`into_future`](super::TryFutureExt::into_future) method.
-    #[derive(Debug)]
+    #[derive(Clone, Debug)]
     #[must_use = "futures do nothing unless you `.await` or poll them"]
     pub struct IntoFuture<Fut> {
         #[pin]

futures-util/src/future/try_future/mod.rs

@@ -31,21 +31,21 @@ delegate_all!(
     /// Future for the [`try_flatten`](TryFutureExt::try_flatten) method.
     TryFlatten<Fut1, Fut2>(
         try_flatten::TryFlatten<Fut1, Fut2>
-    ): Debug + Future + FusedFuture + New[|x: Fut1| try_flatten::TryFlatten::new(x)]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut1| try_flatten::TryFlatten::new(x)]
 );
 
 delegate_all!(
     /// Future for the [`try_flatten_err`](TryFutureExt::try_flatten_err) method.
     TryFlattenErr<Fut1, Fut2>(
         try_flatten_err::TryFlattenErr<Fut1, Fut2>
-    ): Debug + Future + FusedFuture + New[|x: Fut1| try_flatten_err::TryFlattenErr::new(x)]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut1| try_flatten_err::TryFlattenErr::new(x)]
 );
 
 delegate_all!(
     /// Future for the [`try_flatten_stream`](TryFutureExt::try_flatten_stream) method.
     TryFlattenStream<Fut>(
         try_flatten::TryFlatten<Fut, Fut::Ok>
-    ): Debug + Sink + Stream + FusedStream + New[|x: Fut| try_flatten::TryFlatten::new(x)]
+    ): Clone + Debug + Sink + Stream + FusedStream + New[|x: Fut| try_flatten::TryFlatten::new(x)]
     where Fut: TryFuture
 );
 
@@ -55,49 +55,49 @@ delegate_all!(
     #[cfg_attr(docsrs, doc(cfg(feature = "sink")))]
     FlattenSink<Fut, Si>(
         try_flatten::TryFlatten<Fut, Si>
-    ): Debug + Sink + Stream + FusedStream + New[|x: Fut| try_flatten::TryFlatten::new(x)]
+    ): Clone + Debug + Sink + Stream + FusedStream + New[|x: Fut| try_flatten::TryFlatten::new(x)]
 );
 
 delegate_all!(
     /// Future for the [`and_then`](TryFutureExt::and_then) method.
     AndThen<Fut1, Fut2, F>(
         TryFlatten<MapOk<Fut1, F>, Fut2>
-    ): Debug + Future + FusedFuture + New[|x: Fut1, f: F| TryFlatten::new(MapOk::new(x, f))]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut1, f: F| TryFlatten::new(MapOk::new(x, f))]
 );
 
 delegate_all!(
     /// Future for the [`or_else`](TryFutureExt::or_else) method.
     OrElse<Fut1, Fut2, F>(
         TryFlattenErr<MapErr<Fut1, F>, Fut2>
-    ): Debug + Future + FusedFuture + New[|x: Fut1, f: F| TryFlattenErr::new(MapErr::new(x, f))]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut1, f: F| TryFlattenErr::new(MapErr::new(x, f))]
 );
 
 delegate_all!(
     /// Future for the [`err_into`](TryFutureExt::err_into) method.
     ErrInto<Fut, E>(
         MapErr<Fut, IntoFn<E>>
-    ): Debug + Future + FusedFuture + New[|x: Fut| MapErr::new(x, into_fn())]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut| MapErr::new(x, into_fn())]
 );
 
 delegate_all!(
     /// Future for the [`ok_into`](TryFutureExt::ok_into) method.
     OkInto<Fut, E>(
         MapOk<Fut, IntoFn<E>>
-    ): Debug + Future + FusedFuture + New[|x: Fut| MapOk::new(x, into_fn())]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut| MapOk::new(x, into_fn())]
 );
 
 delegate_all!(
     /// Future for the [`inspect_ok`](super::TryFutureExt::inspect_ok) method.
     InspectOk<Fut, F>(
         Inspect<IntoFuture<Fut>, InspectOkFn<F>>
-    ): Debug + Future + FusedFuture + New[|x: Fut, f: F| Inspect::new(IntoFuture::new(x), inspect_ok_fn(f))]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut, f: F| Inspect::new(IntoFuture::new(x), inspect_ok_fn(f))]
 );
 
 delegate_all!(
     /// Future for the [`inspect_err`](super::TryFutureExt::inspect_err) method.
     InspectErr<Fut, F>(
         Inspect<IntoFuture<Fut>, InspectErrFn<F>>
-    ): Debug + Future + FusedFuture + New[|x: Fut, f: F| Inspect::new(IntoFuture::new(x), inspect_err_fn(f))]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut, f: F| Inspect::new(IntoFuture::new(x), inspect_err_fn(f))]
 );
 
 #[allow(unreachable_pub)] // https://github.com/rust-lang/rust/issues/57411
@@ -107,28 +107,28 @@ delegate_all!(
     /// Future for the [`map_ok`](TryFutureExt::map_ok) method.
     MapOk<Fut, F>(
         Map<IntoFuture<Fut>, MapOkFn<F>>
-    ): Debug + Future + FusedFuture + New[|x: Fut, f: F| Map::new(IntoFuture::new(x), map_ok_fn(f))]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut, f: F| Map::new(IntoFuture::new(x), map_ok_fn(f))]
 );
 
 delegate_all!(
     /// Future for the [`map_err`](TryFutureExt::map_err) method.
     MapErr<Fut, F>(
         Map<IntoFuture<Fut>, MapErrFn<F>>
-    ): Debug + Future + FusedFuture + New[|x: Fut, f: F| Map::new(IntoFuture::new(x), map_err_fn(f))]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut, f: F| Map::new(IntoFuture::new(x), map_err_fn(f))]
 );
 
 delegate_all!(
     /// Future for the [`map_ok_or_else`](TryFutureExt::map_ok_or_else) method.
     MapOkOrElse<Fut, F, G>(
         Map<IntoFuture<Fut>, MapOkOrElseFn<F, G>>
-    ): Debug + Future + FusedFuture + New[|x: Fut, f: F, g: G| Map::new(IntoFuture::new(x), map_ok_or_else_fn(f, g))]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut, f: F, g: G| Map::new(IntoFuture::new(x), map_ok_or_else_fn(f, g))]
 );
 
 delegate_all!(
     /// Future for the [`unwrap_or_else`](TryFutureExt::unwrap_or_else) method.
     UnwrapOrElse<Fut, F>(
         Map<IntoFuture<Fut>, UnwrapOrElseFn<F>>
-    ): Debug + Future + FusedFuture + New[|x: Fut, f: F| Map::new(IntoFuture::new(x), unwrap_or_else_fn(f))]
+    ): Clone + Debug + Future + FusedFuture + New[|x: Fut, f: F| Map::new(IntoFuture::new(x), unwrap_or_else_fn(f))]
 );
 
 impl<Fut: ?Sized + TryFuture> TryFutureExt for Fut {}

futures-util/src/future/try_future/try_flatten.rs

@@ -9,7 +9,7 @@ use pin_project_lite::pin_project;
 
 pin_project! {
     #[project = TryFlattenProj]
-    #[derive(Debug)]
+    #[derive(Clone, Debug)]
     pub enum TryFlatten<Fut1, Fut2> {
         First { #[pin] f: Fut1 },
         Second { #[pin] f: Fut2 },

futures-util/src/future/try_future/try_flatten_err.rs

@@ -6,7 +6,7 @@ use pin_project_lite::pin_project;
 
 pin_project! {
     #[project = TryFlattenErrProj]
-    #[derive(Debug)]
+    #[derive(Clone, Debug)]
     pub enum TryFlattenErr<Fut1, Fut2> {
         First { #[pin] f: Fut1 },
         Second { #[pin] f: Fut2 },

futures-util/src/future/try_join.rs

@@ -20,6 +20,18 @@ impl<Fut1: TryFuture, Fut2: TryFuture> TryJoin<Fut1, Fut2> {
     }
 }
 
+impl<Fut1, Fut2> Clone for TryJoin<Fut1, Fut2>
+where
+    Fut1: TryFuture + Clone,
+    Fut1::Ok: Clone,
+    Fut2: TryFuture + Clone,
+    Fut2::Ok: Clone,
+{
+    fn clone(&self) -> Self {
+        Self { fut1: self.fut1.clone(), fut2: self.fut2.clone() }
+    }
+}
+
 impl<Fut1, Fut2> fmt::Debug for TryJoin<Fut1, Fut2>
 where
     Fut1: TryFuture + fmt::Debug,

futures-util/src/future/try_maybe_done.rs

@@ -10,7 +10,7 @@ use futures_core::task::{Context, Poll};
 /// A future that may have completed with an error.
 ///
 /// This is created by the [`try_maybe_done()`] function.
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub enum TryMaybeDone<Fut: TryFuture> {
     /// A not-yet-completed future
     Future(/* #[pin] */ Fut),

futures-util/src/future/try_select.rs

@@ -5,7 +5,7 @@ use futures_core::task::{Context, Poll};
 
 /// Future for the [`try_select()`] function.
 #[must_use = "futures do nothing unless you `.await` or poll them"]
-#[derive(Debug)]
+#[derive(Clone, Debug)]
 pub struct TrySelect<A, B> {
     inner: Option<(A, B)>,
 }

futures-util/src/lib.rs

@@ -253,6 +253,13 @@ macro_rules! delegate_all {
             delegate_sink!(inner, _Item);
         }
     };
+    (@trait Clone $name:ident < $($arg:ident),* > ($t:ty) $(where $($bound:tt)*)*) => {
+        impl<$($arg),*> Clone for $name<$($arg),*> where $t: Clone $(, $($bound)*)* {
+            fn clone(&self) -> Self {
+                Self { inner: Clone::clone(&self.inner) }
+            }
+        }
+    };
     (@trait Debug $name:ident < $($arg:ident),* > ($t:ty) $(where $($bound:tt)*)*) => {
         impl<$($arg),*> core::fmt::Debug for $name<$($arg),*> where $t: core::fmt::Debug $(, $($bound)*)* {
             fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {