description

# Stream<T>.fromFutures constructor

Stream<T>.fromFutures(

1. Iterable<Future<T>> futures

)

Create a single-subscription stream from a group of futures.

The stream reports the results of the futures on the stream in the order in which the futures complete. Each future provides either a data event or an error event, depending on how the future completes.

If some futures have already completed when `Stream.fromFutures` is called, their results will be emitted in some unspecified order.

When all futures have completed, the stream is closed.

If `futures` is empty, the stream closes as soon as possible.

Example:

    Future<int> waitTask() async {
      await Future.delayed(const Duration(seconds: 2));
      return 10;
    }

    Future<String> doneTask() async {
      await Future.delayed(const Duration(seconds: 5));
      return 'Future complete';
    }

    final stream = Stream<Object>.fromFutures([doneTask(), waitTask()]);
    stream.listen(print, onDone: () => print('Done'), onError: print);

    // Outputs:
    // 10 after 'waitTask' finished.
    // "Future complete" after 'doneTask' finished.
    // "Done" when stream completed.

## Implementation

    factory Stream.fromFutures(Iterable<Future<T>> futures) {
      _StreamController<T> controller = _SyncStreamController<T>(
        null,
        null,
        null,
        null,
      );
      int count = 0;
      // Declare these as variables holding closures instead of as
      // function declarations.
      // This avoids creating a new closure from the functions for each future.
      void onValue(T value) {
        if (!controller.isClosed) {
          controller._add(value);
          if (--count == 0) controller._closeUnchecked();
        }
      }

      void onError(Object error, StackTrace stack) {
        if (!controller.isClosed) {
          controller._addError(error, stack);
          if (--count == 0) controller._closeUnchecked();
        }
      }

      // The futures are already running, so start listening to them immediately
      // (instead of waiting for the stream to be listened on).
      // If we wait, we might not catch errors in the futures in time.
      for (var future in futures) {
        count++;
        future.then(onValue, onError: onError);
      }
      // Use schedule microtask since controller is sync.
      if (count == 0) scheduleMicrotask(controller.close);
      return controller.stream;
    }