Returning stream of futures from function

user_14923227 0
Viewed 56

I am trying to write a function which reads a response from a server and returns a stream of futures which when awaited should return a type which implements the tokio AsyncRead trait:

pub async fn connect_peers(url: &str) -> impl futures::stream::Stream<Item = impl AsyncRead> {
    let response = reqwest::get(url).await.unwrap().bytes().await.unwrap();
    let peers: &[u8] = // parse response ...

    peers.array_chunks().map(|x: &[u8; 6]| {
        let (&ip, port) = x.split_array_ref::<4>();
        let addr = SocketAddr::new(IpAddr::from(ip), BigEndian::read_u16(port));
        tokio::net::TcpStream::connect(addr)
    }).collect()
}

However, I get the following compile error:

error[E0277]: a value of type `impl Stream<Item = impl tokio::io::AsyncRead>` cannot be built from an iterator over elements of type `impl futures::Future<Output = Result<tokio::net::TcpStream, std::io::Error>>`
    --> src/lib.rs:350:12
     |
350  |         }).collect()
     |            ^^^^^^^ value of type `impl Stream<Item = impl tokio::io::AsyncRead>` cannot be built from `std::iter::Iterator<Item=impl futures::Future<Output = Result<tokio::net::TcpStream, std::io::Error>>>`
     |
     = help: the trait `FromIterator<impl futures::Future<Output = Result<tokio::net::TcpStream, std::io::Error>>>` is not implemented for `impl Stream<Item = impl tokio::io::AsyncRead>`

I am not sure how to create a futures::stream::Stream from the iterator returned by peers.array_chunks().map(|x: &[u8; 6]| { ... }).

This blog post show that something like this is possible:

fn get_pages_futures() -> impl Stream<Item = impl Future<Output = Vec<usize>>> {
    stream::iter(0..).map(|i| get_page(i))
}

where get_page is an async function returning Vec<usize>. But I am not sure how this would translate to my use case.

I have tried the following return type signature (adding the future as output):

impl futures::stream::Stream<Item = impl futures::Future<Output = impl AsyncRead>>

which gives the following compile error:

error[E0277]: a value of type `impl Stream<Item = impl futures::Future<Output = impl tokio::io::AsyncRead>>` cannot be built from an iterator over elements of type `impl futures::Future<Output = Result<tokio::net::TcpStream, std::io::Error>>`
    --> src/lib.rs:350:12
     |
350  |         }).collect()
     |            ^^^^^^^ value of type `impl Stream<Item = impl futures::Future<Output = impl tokio::io::AsyncRead>>` cannot be built from `std::iter::Iterator<Item=impl futures::Future<Output = Result<tokio::net::TcpStream, std::io::Error>>>`
     |
     = help: the trait `FromIterator<impl futures::Future<Output = Result<tokio::net::TcpStream, std::io::Error>>>` is not implemented for `impl Stream<Item = impl futures::Future<Output = impl tokio::io::AsyncRead>>`

How do I return an asynchronous stream from the connect_peers function?

rust
1 Answers

The use of stream::iter() is on the right track - it converts an iterator to a stream. Since you already have an iterator, you need to replace xxx.collect() with stream::iter(xxx):

futures::stream::iter(peers.array_chunks().map(|x: &[u8; 6]| {
    let (&ip, port) = x.split_array_ref::<4>();
    let addr = SocketAddr::new(IpAddr::from(ip), BigEndian::read_u16(port));
    tokio::net::TcpStream::connect(addr)
}))

However, that results in an error:

error[E0277]: the trait bound `impl futures::Future<Output = Result<tokio::net::TcpStream, std::io::Error>>: tokio::io::AsyncRead` is not satisfied

If you take a more careful look at the code, the error makes sense: we promised to return a stream of something that implements AsyncRead (such as TcpStream), but connect() doesn't return a TcpStream, it returns a future that will, once it completes, return a TcpStream.

To fix this, you need to turn your stream that returns futures into a stream that returns TcpStreams - something like:

futures::stream::iter(peers.array_chunks().map(|x: &[u8; 6]| {
    // ...
    tokio::net::TcpStream::connect(addr)
}))
.map(|connect| async { connect.await.unwrap() })

However StreamExt::map() takes an ordinary closure, and we need an async version. The async equivalent of map() is, somewhat surprisingly called then(), and if we change map() to then(), the code compiles:

pub async fn connect_peers(url: &str) -> impl futures::stream::Stream<Item = impl AsyncRead> {
    let response = reqwest::get(url).await.unwrap().bytes().await.unwrap();
    let peers: &[u8] = b"foo";

    futures::stream::iter(peers.array_chunks().map(|x: &[u8; 6]| {
        let (&ip, port) = x.split_array_ref::<4>();
        let addr = SocketAddr::new(IpAddr::from(ip), BigEndian::read_u16(port));
        tokio::net::TcpStream::connect(addr)
    }))
    .then(|connect| async { connect.await.unwrap() })
}

Playground

user_16008980
Related