title: How to run stream to completion in Rust using combinators (other than for_each) and without a while loop?
body: <p>I have a stream which uses combinators and I need to run it to completion. I can use a <code>while</code> loop or the <code>for_each</code> combinator. They will both work, but I think there must be a nicer way.</p>
  <p><a href="https://docs.rs/futures/0.3.6/futures/sink/trait.Sink.html" rel="nofollow noreferrer"><code>Sink</code></a> looks like what I'm looking for, especially <a href="https://docs.rs/futures/0.3.6/futures/sink/fn.drain.html" rel="nofollow noreferrer"><code>sink::drain()</code></a>, but I havn't been able to understand how to use it.</p>
  <p>Using while loop</p>
  <pre class="lang-rust prettyprint-override"><code>use futures::{StreamExt, TryStreamExt}; // 0.3.6
  use tokio; // 0.3.0
  
  #[tokio::main]
  async fn main() {
      let mut stream = Box::pin(
          futures::stream::iter(0..20)
              .map(foo)
              .map_ok(|x| x * 10)
              .and_then(bar)
              .filter(|x| futures::future::ready(x.is_ok())),
      );
  
      while let Some(_) = stream.next().await {
          // Nothing to do here. I just need to run stream.
      }
  }
  
  fn foo(x: i32) -&gt; Result&lt;i32, String&gt; {
      if x != 10 {
          Ok(x)
      } else {
          Err(&quot;eeer&quot;.to_string())
      }
  }
  
  async fn bar(x: i32) -&gt; Result&lt;(), String&gt; {
      async {
          if x == 13 {
              Err(&quot;errr&quot;.to_string())
          } else {
              Ok(())
          }
      }
      .await
  }
  </code></pre>
  <p>Using <code>for_each</code>:</p>
  <pre class="lang-rust prettyprint-override"><code>use futures::{StreamExt, TryStreamExt}; // 0.3.6
  use tokio; // 0.3.0
  
  #[tokio::main]
  async fn main() {
      futures::stream::iter(0..20)
          .map(foo)
          .map_ok(|x| x * 10)
          .and_then(bar)
          .filter(|x| futures::future::ready(x.is_ok()))
          .for_each(|_| futures::future::ready(())) // Nothing to do here, just to run stream
          .await;
  }
  
  fn foo(x: i32) -&gt; Result&lt;i32, String&gt; {
      if x != 10 {
          Ok(x)
      } else {
          Err(&quot;eeer&quot;.to_string())
      }
  }
  
  async fn bar(x: i32) -&gt; Result&lt;(), String&gt; {
      async {
          if x == 13 {
              Err(&quot;errr&quot;.to_string())
          } else {
              Ok(())
          }
      }
      .await
  }
  </code></pre>
  <p>I would like to have something like the following. It's not necessary to use the <code>drain</code> combinator exactly, just some combinator to run the stream:</p>
  <pre class="lang-rust prettyprint-override"><code>use futures::{StreamExt, TryStreamExt}; // 0.3.6
  use tokio; // 0.3.0
  
  #[tokio::main]
  async fn main() {
      futures::stream::iter(0..20)
          .map(foo)
          .map_ok(|x| x * 10)
          .and_then(bar)
          .filter(|x| futures::future::ready(x.is_ok()))
          .forward(futures::sink::drain())
          .await;
  }
  
  fn foo(x: i32) -&gt; Result&lt;i32, String&gt; {
      if x != 10 {
          Ok(x)
      } else {
          Err(&quot;eeer&quot;.to_string())
      }
  }
  
  async fn bar(x: i32) -&gt; Result&lt;(), String&gt; {
      async {
          if x == 13 {
              Err(&quot;errr&quot;.to_string())
          } else {
              Ok(())
          }
      }
      .await
  }
  </code></pre>
  <p>This doesn't work, probably because drain puts some bounds on the <code>Error</code> type:</p>
  <pre class="lang-none prettyprint-override"><code>error[E0271]: type mismatch resolving `&lt;futures::sink::Drain&lt;()&gt; as futures::Sink&lt;()&gt;&gt;::Error == std::string::String`
    --&gt; src/main.rs:11:10
     |
  11 |         .forward(futures::sink::drain())
     |          ^^^^^^^ expected enum `std::convert::Infallible`, found struct `std::string::String`
  
  error[E0271]: type mismatch resolving `&lt;futures::stream::Filter&lt;futures::stream::AndThen&lt;futures::stream::MapOk&lt;futures::stream::Map&lt;futures::stream::Iter&lt;std::ops::Range&lt;i32&gt;&gt;, fn(i32) -&gt; std::result::Result&lt;i32, std::string::String&gt; {foo}&gt;, [closure@src/main.rs:8:17: 8:27]&gt;, impl futures::Future, fn(i32) -&gt; impl futures::Future {bar}&gt;, futures::future::Ready&lt;bool&gt;, [closure@src/main.rs:10:17: 10:54]&gt; as futures::Stream&gt;::Item == std::result::Result&lt;(), std::convert::Infallible&gt;`
    --&gt; src/main.rs:6:5
     |
  6  | /     futures::stream::iter(0..20)
  7  | |         .map(foo)
  8  | |         .map_ok(|x| x * 10)
  9  | |         .and_then(bar)
  10 | |         .filter(|x| futures::future::ready(x.is_ok()))
  11 | |         .forward(futures::sink::drain())
  12 | |         .await;
     | |______________^ expected struct `std::string::String`, found enum `std::convert::Infallible`
     |
     = note: expected enum `std::result::Result&lt;_, std::string::String&gt;`
                found enum `std::result::Result&lt;_, std::convert::Infallible&gt;`
     = note: required because of the requirements on the impl of `futures::Future` for `futures_util::stream::stream::forward::Forward&lt;futures::stream::Filter&lt;futures::stream::AndThen&lt;futures::stream::MapOk&lt;futures::stream::Map&lt;futures::stream::Iter&lt;std::ops::Range&lt;i32&gt;&gt;, fn(i32) -&gt; std::result::Result&lt;i32, std::string::String&gt; {foo}&gt;, [closure@src/main.rs:8:17: 8:27]&gt;, impl futures::Future, fn(i32) -&gt; impl futures::Future {bar}&gt;, futures::future::Ready&lt;bool&gt;, [closure@src/main.rs:10:17: 10:54]&gt;, futures::sink::Drain&lt;()&gt;, ()&gt;`
  </code></pre>