title: How can I `flatmap` streams in Rust?
body: <p>I have a <a href="https://docs.rs/rusoto_core/0.36.0/rusoto_core/struct.ByteStream.html" rel="nofollow noreferrer"><code>rusoto_core::ByteStream</code></a> which implements <a href="https://docs.rs/futures/0.1.25/futures/stream/trait.Stream.html" rel="nofollow noreferrer">futures' <code>Stream</code> trait</a>:</p>
  
  <pre><code>let chunks = vec![b"1234".to_vec(), b"5678".to_vec()];
  let stream = ByteStream::new(stream::iter_ok(chunks));
  </code></pre>
  
  <p>I'd like to pass it to <a href="https://docs.rs/actix-web/0.7.18/actix_web/dev/struct.HttpResponseBuilder.html#method.streaming" rel="nofollow noreferrer">actix_web's <code>HttpResponseBuilder::streaming</code></a> method.</p>
  
  <pre><code>use actix_web::dev::HttpResponseBuilder; // 0.7.18
  use rusoto_core::ByteStream; // 0.36.0
  
  fn example(stream: ByteStream, builder: HttpResponseBuilder) {
      builder.streaming(stream);
  }
  </code></pre>
  
  <p>When I try to do it I receive the following error:</p>
  
  <pre class="lang-none prettyprint-override"><code>error[E0271]: type mismatch resolving `&lt;rusoto_core::stream::ByteStream as futures::stream::Stream&gt;::Item == bytes::bytes::Bytes`
   --&gt; src/main.rs:5:13
    |
  5 |     builder.streaming(stream);
    |             ^^^^^^^^^ expected struct `std::vec::Vec`, found struct `bytes::bytes::Bytes`
    |
    = note: expected type `std::vec::Vec&lt;u8&gt;`
               found type `bytes::bytes::Bytes`
  </code></pre>
  
  <p>I believe the reason is that <code>streaming()</code> expects a <code>S: Stream&lt;Item = Bytes, Error&gt;</code> (i.e., <code>Item = Bytes</code>) but my <code>ByteStream</code> has <code>Item = Vec&lt;u8&gt;</code>. How can I fix it?</p>
  
  <p>I think the solution is to <code>flatmap</code> my <code>ByteStream</code> somehow but I couldn't find such a method for streams.</p>
  
  <p>Here's an example how <code>streaming()</code> can be used:</p>
  
  <pre><code>let text = "123";
  let (tx, rx_body) = mpsc::unbounded();
  let _ = tx.unbounded_send(Bytes::from(text.as_bytes()));
  
  HttpResponse::Ok()
      .streaming(rx_body.map_err(|e| error::ErrorBadRequest("bad request")))
  </code></pre>