each<T> method

@override Job<void> each<T>(Stream<T> stream, FutureOr<void> onData(JobContext ctx, T event), { Object? key, String describe()?, })

Starts a child job that processes stream one event at a time.

Returns the child at once without observing its outcome; the parent waits for it even when the body does not, so an open stream keeps the parent alive until the stream ends or the child is cancelled. onData gets the child's context, and an asynchronous callback is awaited before the next event. Throws synchronously if a child cannot be started, as run does. key and describe name the child as they name any job, see A job's name; without describe its description is each. The child's cleanup stack unwinds when the child ends, not after an event: what the callback registers piles up for as long as the stream lasts.

final listening = ctx.each<int>(events, (child, event) {
  child.log(event);
});
await listening.cancel();

See Processing a stream on the streams page.

Implementation

@override
Job<void> each<T>(
  Stream<T> stream,
  FutureOr<void> Function(JobContext ctx, T event) onData, {
  Object? key,
  String Function()? describe,
}) {
  // Asked here, with its own verb, before `startChild` asks again with
  // one about children: the caller followed a stream, and the message
  // names what they did.
  _checkCanRun('follow a stream');
  final child = createEachJob(
    (child) => child._followStream(stream, (event) => onData(child, event)),
    key: key,
    describe: describe ?? () => 'each',
  );
  startChild(child);
  return child;
}