Stream transformation functions should all accept futures

[nex3]

The Stream API has numerous transformation functions that take blocks, mirroring the Iterable transformation functions. Of these, only mappedBy and reduce allow their blocks to return Futures. It would be more consistent and more useful if all the transformation functions worked with asynchronous blocks.

[madsager]

cc @floitschG.
Removed Area-IO label.
Added Area-Library label.

[lrhn]

I don't see mappedBy or reduce accepting Futures either (they accept futures as any other value, but don't treat them specially).

The problem with handling futures is that it isn't obvious what to do about them
You can either pause the incoming stream until the future completes, and then use the result as the next stream event, or you can just let the original stream continue report the future results in the order they complete in (that is, not necessarily preserve the order of the original events). Both makes sense in different cases.

[nex3]

You're right, I misread the lack of type annotations to indicate that they could return Futures.

The driving use case here isn't map/reduce (where it sometimes makes sense to return literal Futures rather than resolving them) but other methods that return e.g. booleans or ints, such as filter(), any(), or max().

[nex3]

Added Library-Async label.

[lrhn]

Removed Type-Defect label.
Added Type-Enhancement label.

[natebosch]

Now that we have FutureOr we could express this in the type signature.

We introduced Stream.asyncMap to get around this, but as pointed out some of the other methods would be very useful. asyncWhere is one I had to implement as a StreamTransformer.

I'm not sure if changing any of these would be breaking.

[lrhn]

It is breaking to change the signature of any public instance member on any public class, so no need to wonder if any particular change is breaking. It always is.

Even if it wasn't breaking, I still don't think these operations are worth making asynchronous.
We could potentially make every asynchronous function (anything returning a stream or future) allow asynchronous functions for all callbacks, but it's an overhead on every call, even when you don't use it. I'd rather just use async functions to write the code directly instead of using the stream methods to with closures to modify a stream by an async function or predicate.

[matanlurey]

@lrhn:

It is breaking to change the signature of any public instance member on any public class, so no need to wonder if any particular change is breaking. It always is.

That's true. Consider:

abstract class StreamLike<T> {
  StreamLike<S> map<S>(S Function(T) map);
}

... changing it to ...

abstract class StreamLike<T> {
  StreamLike<S> map<S>(FutureOr<S> Function(T) map);
}

... means:

• Users that implement StreamLike need to update their map function.
• Tear-ing off StreamLike.map now has a different type signature.

Even if it wasn't breaking, I still don't think these operations are worth making asynchronous.

I think @natebosch is more asking, make it possible for them to be either async or sync:

abstract class StreamLike<T> {
  StreamLike<S> map<S>(FutureOr<S> Function(T) map) async* {
    // Hypothetical.
    for (var element in this) {
      var result = await map(element);
      yield result;
    }
  }
}

But it's an overhead on every call, even when you don't use it.

That part is true. Could we detect the function signature, i.e.:

if (map is Future<S> Function(T)) {
  return _mapAsync(...);
} else {
  return _mapSync(...);
}