Use more assert_{future,stream} functions (#2020)

Use `assert_future` and `assert_stream` annotations in almost all
`FutureExt`, `TryFutureExt`, `StreamExt` and `TryStreamExt` trait
fns.

My own style of working with third-party code is doing Cmd-Click
in Idea and look at signature/implementation.

I found these existing `assert_` annotations very helpful in
understanding what this particular function does, what it expects,
what it returns, much faster than reading documentation or opening
actual future/stream implementation.

For example, there's `TryStreamExt::and_then` function:

```
    fn and_then<Fut, F>(self, f: F) -> AndThen<Self, Fut, F>
    where
        F: FnMut(Self::Ok) -> Fut,
        Fut: TryFuture<Error = Self::Error>,
        Self: Sized,
    { ... }
```

What does it do? Does it return a future or a stream?  Is stream
item original stream item or returned from future?  Is error
converted? Is it even `Try` object returned from this function?

Now with the function body:

```
        assert_stream::<Result<Fut::Ok, Fut::Error>, _>(...)
```

it's easier to infer that function returns a stream with the same
items as provided future.

So this diff extends `assert_` function to almost all functions of
these four traits.

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

@@ -3,20 +3,21 @@
 //! This module contains a number of functions for working with `Future`s,
 //! including the `FutureExt` trait which adds methods to `Future` types.
 
-use super::{assert_future, Either};
 #[cfg(feature = "alloc")]
 use alloc::boxed::Box;
 use core::pin::Pin;
 
+use crate::future::{assert_future, Either};
+use crate::stream::assert_stream;
+use crate::fns::{inspect_fn, into_fn, ok_fn, InspectFn, IntoFn, OkFn};
+use crate::never::Never;
 #[cfg(feature = "alloc")]
 use futures_core::future::{BoxFuture, LocalBoxFuture};
 use futures_core::{
     future::Future,
     stream::Stream,
     task::{Context, Poll},
 };
-use crate::never::Never;
-use crate::fns::{OkFn, ok_fn, IntoFn, into_fn, InspectFn, inspect_fn};
 use pin_utils::pin_mut;
 
 // Combinators
@@ -223,7 +224,7 @@ pub trait FutureExt: Future {
         B: Future<Output = Self::Output>,
         Self: Sized,
     {
-        Either::Left(self)
+        assert_future::<Self::Output, _>(Either::Left(self))
     }
 
     /// Wrap this future in an `Either` future, making it the right-hand variant
@@ -253,7 +254,7 @@ pub trait FutureExt: Future {
         A: Future<Output = Self::Output>,
         Self: Sized,
     {
-        Either::Right(self)
+        assert_future::<Self::Output, _>(Either::Right(self))
     }
 
     /// Convert this future into a single element stream.
@@ -278,7 +279,7 @@ pub trait FutureExt: Future {
     where
         Self: Sized,
     {
-        IntoStream::new(self)
+        assert_stream::<Self::Output, _>(IntoStream::new(self))
     }
 
     /// Flatten the execution of this future when the output of this
@@ -342,7 +343,7 @@ pub trait FutureExt: Future {
         Self::Output: Stream,
         Self: Sized,
     {
-        FlattenStream::new(self)
+        assert_stream::<<Self::Output as Stream>::Item, _>(FlattenStream::new(self))
     }
 
     /// Fuse a future such that `poll` will never again be called once it has
@@ -431,7 +432,9 @@ pub trait FutureExt: Future {
     where
         Self: Sized + ::std::panic::UnwindSafe,
     {
-        CatchUnwind::new(self)
+        assert_future::<Result<Self::Output, Box<dyn std::any::Any + Send>>, _>(CatchUnwind::new(
+            self,
+        ))
     }
 
     /// Create a cloneable handle to this future where all handles will resolve
@@ -485,7 +488,7 @@ pub trait FutureExt: Future {
         Self: Sized,
         Self::Output: Clone,
     {
-        Shared::new(self)
+        assert_future::<Self::Output, _>(Shared::new(self))
     }
 
     /// Turn this future into a future that yields `()` on completion and sends
@@ -515,7 +518,7 @@ pub trait FutureExt: Future {
     where
         Self: Sized + Send + 'a,
     {
-        Box::pin(self)
+        assert_future::<Self::Output, _>(Box::pin(self))
     }
 
     /// Wrap the future in a Box, pinning it.
@@ -529,7 +532,7 @@ pub trait FutureExt: Future {
     where
         Self: Sized + 'a,
     {
-        Box::pin(self)
+        assert_future::<Self::Output, _>(Box::pin(self))
     }
 
     /// Turns a [`Future<Output = T>`](Future) into a
@@ -538,7 +541,7 @@ pub trait FutureExt: Future {
     where
         Self: Sized,
     {
-        UnitError::new(self)
+        assert_future::<Result<Self::Output, ()>, _>(UnitError::new(self))
     }
 
     /// Turns a [`Future<Output = T>`](Future) into a
@@ -547,7 +550,7 @@ pub trait FutureExt: Future {
     where
         Self: Sized,
     {
-        NeverError::new(self)
+        assert_future::<Result<Self::Output, Never>, _>(NeverError::new(self))
     }
 
     /// A convenience for calling `Future::poll` on `Unpin` future types.

futures-util/src/future/mod.rs

@@ -109,7 +109,7 @@ cfg_target_has_atomic! {
 
 // Just a helper function to ensure the futures we're returning all have the
 // right implementations.
-fn assert_future<T, F>(future: F) -> F
+pub(crate) fn assert_future<T, F>(future: F) -> F
 where
     F: Future<Output = T>,
 {

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

@@ -14,13 +14,13 @@ use futures_core::{
 #[cfg(feature = "sink")]
 use futures_sink::Sink;
 
-use super::assert_future;
-use crate::future::{Map, Inspect};
 use crate::fns::{
-    MapOkFn, map_ok_fn, MapErrFn, map_err_fn, MapOkOrElseFn,
-    map_ok_or_else_fn, IntoFn, UnwrapOrElseFn, unwrap_or_else_fn, InspectOkFn, inspect_ok_fn, InspectErrFn,
-    inspect_err_fn, into_fn
+    inspect_err_fn, inspect_ok_fn, into_fn, map_err_fn, map_ok_fn, map_ok_or_else_fn,
+    unwrap_or_else_fn, InspectErrFn, InspectOkFn, IntoFn, MapErrFn, MapOkFn, MapOkOrElseFn,
+    UnwrapOrElseFn,
 };
+use crate::future::{assert_future, Inspect, Map};
+use crate::stream::assert_stream;
 
 // Combinators
 mod into_future;
@@ -230,7 +230,7 @@ pub trait TryFutureExt: TryFuture {
     /// The provided closure `f` will only be called if this future is resolved
     /// to an [`Ok`]. If it resolves to an [`Err`], panics, or is dropped, then
     /// the provided closure will never be invoked.
-    /// 
+    ///
     /// The provided closure `e` will only be called if this future is resolved
     /// to an [`Err`]. If it resolves to an [`Ok`], panics, or is dropped, then
     /// the provided closure will never be invoked.
@@ -247,13 +247,13 @@ pub trait TryFutureExt: TryFuture {
     /// let future = async { Ok::<i32, i32>(5) };
     /// let future = future.map_ok_or_else(|x| x * 2, |x| x + 3);
     /// assert_eq!(future.await, 8);
-    /// 
+    ///
     /// let future = async { Err::<i32, i32>(5) };
     /// let future = future.map_ok_or_else(|x| x * 2, |x| x + 3);
     /// assert_eq!(future.await, 10);
     /// # });
     /// ```
-    /// 
+    ///
     fn map_ok_or_else<T, E, F>(self, e: E, f: F) -> MapOkOrElse<Self, F, E>
     where
         F: FnOnce(Self::Ok) -> T,
@@ -534,7 +534,9 @@ pub trait TryFutureExt: TryFuture {
         Self::Ok: TryStream<Error = Self::Error>,
         Self: Sized,
     {
-        TryFlattenStream::new(self)
+        assert_stream::<Result<<Self::Ok as TryStream>::Ok, Self::Error>, _>(TryFlattenStream::new(
+            self,
+        ))
     }
 
     /// Unwraps this future's ouput, producing a future with this future's
@@ -603,7 +605,7 @@ pub trait TryFutureExt: TryFuture {
     where
         Self: Sized,
     {
-        IntoFuture::new(self)
+        assert_future::<Result<Self::Ok, Self::Error>, _>(IntoFuture::new(self))
     }
 
     /// A convenience method for calling [`TryFuture::try_poll`] on [`Unpin`]

futures-util/src/stream/mod.rs

@@ -100,3 +100,12 @@ cfg_target_has_atomic! {
     #[cfg(feature = "alloc")]
     pub use self::select_all::{select_all, SelectAll};
 }
+
+// Just a helper function to ensure the futures we're returning all have the
+// right implementations.
+pub(crate) fn assert_stream<T, S>(stream: S) -> S
+    where
+        S: Stream<Item = T>,
+{
+    stream
+}