Using both tokio and futures StreamExt

help
1

Hello,

I have a code style question. According to tokio_stream docs when using both futures and tokio StreamExt, one should prefer import one and use other with fullly quallified syntax.

I would like to use both in stream adaptors and keep the code readable. Are there are any tips or good practices?

Here's a code example I have right now. I chose to shadow producer_stream because I couldn't use .chunks_timeout directly.

use futures::stream::{self, StreamExt};
use rand::{distributions::Uniform, Rng};
use tokio::time::{self, Duration};

#[tokio::main]
async fn main() {
    let duration = Duration::from_millis(200);
    let interval = time::interval(duration);

    let mut rng = rand::thread_rng();
    let value_range = Uniform::new(0.0, 10.0);

    let producer_stream = tokio_stream::wrappers::IntervalStream::new(interval)
        .map(|_| {
            // A random vector
            let n: usize = rng.gen_range(1..5);
            (0..n)
                .map(|_| rng.sample(&value_range))
                .collect::<Vec<f64>>()
        })
        .flat_map(|v| stream::iter(v));
    let producer_stream =
        tokio_stream::StreamExt::chunks_timeout(producer_stream, 10, Duration::from_millis(500));

    tokio::pin!(producer_stream);

    while let Some(items) = producer_stream.next().await {
        println!("{:?}", items);
    }
}
2

What you're doing looks fine to me.

3

Ok, thank you for the confirmation. I was just wondering whether there's a leaner solution.

On a similar note, are there any general recomendations on consuming the streams?

Assume that I want to call a(n async) function on each element of the stream. Is it preferable to do it in while let loop

while let Some(v) = stream.next().await {
  do_async_work(v).await;  
}

or add .then method?

let stream = some_stream.then(do_async_work);
4

I think while loops are generally a good idea, because there are some cases where they are more lenient borrow-checker wise.