Use more assert_{future,stream} functions

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

@@ -77,6 +77,8 @@ pub use self::shared::Shared;
 
 mod chain;
 pub(crate) use self::chain::Chain;
+use crate::never::Never;
+use crate::stream::assert_stream;
 
 impl<T: ?Sized> FutureExt for T where T: Future {}
 
@@ -173,7 +175,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
@@ -203,7 +205,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.
@@ -228,7 +230,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
@@ -292,7 +294,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
@@ -381,7 +383,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
@@ -435,7 +439,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
@@ -464,7 +468,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.
@@ -478,7 +482,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
@@ -487,7 +491,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
@@ -496,7 +500,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

@@ -102,7 +102,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

@@ -69,6 +69,8 @@ pub(crate) use self::flatten_stream_sink::FlattenStreamSink;
 
 mod try_chain;
 pub(crate) use self::try_chain::{TryChain, TryChainAction};
+use crate::future::assert_future;
+use crate::stream::assert_stream;
 
 impl<Fut: ?Sized + TryFuture> TryFutureExt for Fut {}
 
@@ -157,7 +159,7 @@ pub trait TryFutureExt: TryFuture {
         F: FnOnce(Self::Ok) -> T,
         Self: Sized,
     {
-        MapOk::new(self, f)
+        assert_future::<Result<T, Self::Error>, _>(MapOk::new(self, f))
     }
 
     /// Maps this future's error value to a different value.
@@ -204,7 +206,7 @@ pub trait TryFutureExt: TryFuture {
         F: FnOnce(Self::Error) -> E,
         Self: Sized,
     {
-        MapErr::new(self, f)
+        assert_future::<Result<Self::Ok, E>, _>(MapErr::new(self, f))
     }
 
     /// Maps this future's [`Error`](TryFuture::Error) to a new error type
@@ -234,7 +236,7 @@ pub trait TryFutureExt: TryFuture {
         Self: Sized,
         Self::Error: Into<E>,
     {
-        ErrInto::new(self)
+        assert_future::<Result<Self::Ok, E>, _>(ErrInto::new(self))
     }
 
     /// Executes another future after this one resolves successfully. The
@@ -279,7 +281,7 @@ pub trait TryFutureExt: TryFuture {
         Fut: TryFuture<Error = Self::Error>,
         Self: Sized,
     {
-        AndThen::new(self, f)
+        assert_future::<Result<Fut::Ok, Fut::Error>, _>(AndThen::new(self, f))
     }
 
     /// Executes another future if this one resolves to an error. The
@@ -324,7 +326,7 @@ pub trait TryFutureExt: TryFuture {
         Fut: TryFuture<Ok = Self::Ok>,
         Self: Sized,
     {
-        OrElse::new(self, f)
+        assert_future::<Result<Self::Ok, Fut::Error>, _>(OrElse::new(self, f))
     }
 
     /// Do something with the success value of a future before passing it on.
@@ -350,7 +352,7 @@ pub trait TryFutureExt: TryFuture {
         F: FnOnce(&Self::Ok),
         Self: Sized,
     {
-        InspectOk::new(self, f)
+        assert_future::<Result<Self::Ok, Self::Error>, _>(InspectOk::new(self, f))
     }
 
     /// Do something with the error value of a future before passing it on.
@@ -376,7 +378,7 @@ pub trait TryFutureExt: TryFuture {
         F: FnOnce(&Self::Error),
         Self: Sized,
     {
-        InspectErr::new(self, f)
+        assert_future::<Result<Self::Ok, Self::Error>, _>(InspectErr::new(self, f))
     }
 
     /// Flatten the execution of this future when the successful result of this
@@ -409,7 +411,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
@@ -439,7 +443,7 @@ pub trait TryFutureExt: TryFuture {
         Self: Sized,
         F: FnOnce(Self::Error) -> Self::Ok,
     {
-        UnwrapOrElse::new(self, f)
+        assert_future::<Self::Ok, _>(UnwrapOrElse::new(self, f))
     }
 
     /// Wraps a [`TryFuture`] into a future compatable with libraries using
@@ -477,7 +481,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

@@ -94,3 +94,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
+}