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)>,
 }