Q:

How to write len methods for futures::Stream?

Here's the code I tried (used a fold() to implement len() for streams but there're some weird type errors I can't understand (do(row: Row) returns a Result<(), Error>):
let rows_count = rows
    .for_each(|row| parse_row(&row))
    .fold(0, |acc, a| futures::future::ok(acc + 1))
    .wait()
    .unwrap();

Thanks!

A:

Per default fold on Streams returns a Future. If you want to block on the Stream and get the number of elements you need to wait on it.
Furthermore if any element in the Stream is an error, it will panic.
Possible blocking implementation:
use futures::stream::Stream;
use futures::future::{Future, ok};

trait StreamExt {
    fn len(self) -> usize;
}

impl<T: Stream> StreamExt for T {
    fn len(self) -> usize {
        self.fold(0, |a, _| ok(a + 1))
            .wait()
            .map_err(|_| ())
            .unwrap()
    }
}

Playground

