Streams

A job of a controller follows a stream in a child that owns the subscription: ctx.each starts that child and hands its callback each event with the child’s context. The subscription lives exactly as long as the child: the child ends when the stream is done, and cancelling the child or the job cancels the subscription. This page is about what a controller adds to such a child: the queue it holds, close(), the start rules, and the stream of another controller.

hw belongs to the application these examples come from, and Ready is the state its controller works in, with the field position.

Processing a stream

A controller copies the positions its hardware reports into the state.

The first attempt

// The first attempt: the job ends before the first position.
Job<void> track() => enqueue<Ready, void>(key: 'track', (ctx) async {
  hw.positions.listen((p) => ctx.emit(ctx.state.copyWith(position: p)));
});

The body subscribes and returns, so the job is over before the first position arrives, and the queue moves on. The subscription stays, holding the context of a job that has finished. Each position it delivers throws out of emit into the zone, and the state never changes:

Bad state: Job(track) has already finished, cannot emit

Nothing ends that subscription either: there is no job left to cancel, and close() knows nothing of it. Registering sub.cancel with ctx.onDispose does not repair that: the cleanup runs as soon as the body returns, and it cancels the subscription before the first position arrives.

A child that owns the subscription

Job<void> track() => enqueue<Ready, void>(
  key: 'track',
  (ctx) => ctx.each(hw.positions, (childCtx, p) {
    childCtx.emit(childCtx.state.copyWith(position: p));
  }).value,
);

ctx.each(stream, onData) creates and returns a child SoloJob<void> that owns the subscription. Each event callback receives that child’s context. Use it for state access and cancellation-aware operations. Returning the child’s .value makes stream and callback errors propagate into the parent body. Use .done to inspect the outcome. To stop only this subscription, retain the returned job and call its cancel(), and have the body await .done rather than .value: a Cancelled from .value cancels the parent as well, as below.

The child ends when the source sends onDone, so an open stream with no events is a child still running. The parent waits for its children with or without an explicit await, so it keeps running too, and holds the queue. Accepted parent cancellation, including during close(), cancels the child; a draining close(mode: SoloCloseMode.drain) cancels nothing and waits for the stream to end. The child is cancellable even if the parent is not. It retains the parent’s working type and keepWhile after the parent body returns, but does not repeat canStart. Observers see it as a separate job, and ctx.each names it Job(each) unless the call gives it a key or a describe. Since the child would keep the parent’s working type and keepWhile, ctx.each asks them first, as ctx.state does. After the parent’s own emit of a state they refuse, it ends the parent Cancelled and throws, rather than return a child that never listened.

Cancelling only the child does not directly cancel the parent. An uncaught Cancelled from the child’s .value does cancel the parent through its body. A child whose outcome nobody reads, like the progress in _sync at the top of Children and chains, still holds the parent, but its failure is not the parent’s. The parent ends Done, and the failure goes where a failure nobody read goes, to onUnanswered and by default to the zone, as in Handled and unhandled failures on the errors page. ignoreFailure() on the child keeps it out of both.

Events are delivered one at a time, and a cancellation waits for the callback in flight. That is why waits belong to the callback’s context: childCtx.abandonable ends with the cancellation the moment it arrives and leaves the action running alone, while a plain await ends only when its own future does. Until the callback returns, the child and the parent are still running, the queue is held and close() does not come back. Do not await the child’s own completion or its cancel() inside an event callback: that is a deadlock. The child waits for the callback to return, the callback waits for the child, and the parent waits for the child — nothing moves again. To stop the subscription from inside a callback, call cancel() and do not await it.

The child does not wait for the source: the future the subscription’s cancel() returns is the cleanup of the source, which this job does not own. If that cleanup fails, its error goes to the controller’s onError and on to its onUnanswered, as an error no outcome carries does. A job that must not end before its source has closed waits for that itself, the way Waiting for the source to close on the streams page of async_job does. The rest of that page holds here as well, with job(...) or ctx.child where it writes Job.deferred. There you find what await for and a listen tied to the job do in a body, how a callback in several steps stops for a cancellation, and where the cleanup of one event belongs.

Following another controller

A screen shows whether the user is signed in, and the session is the state of another controller.

The first attempt

// The first attempt: the screen misses the current session.
final class ScreenController extends Solo<Screen> {
  final SoloStream<Session> session;

  ScreenController(this.session) : super(const Screen());

  Job<void> follow() => enqueue<Screen, void>(
    key: 'follow',
    (ctx) => ctx.each(session.stream, (childCtx, next) {
      childCtx.emit(childCtx.state.copyWith(signedIn: next.signedIn));
    }).value,
  );
}

The stream of a controller sends only changes of the state: a subscriber is not sent the current state when it subscribes. A session that is already signed in when follow() starts sends nothing, and the screen goes on showing signedIn: false until the session changes.

The state first, then the stream

Job<void> follow() => enqueue<Screen, void>(key: 'follow', (ctx) async {
  void take(SoloContext<Screen, Screen> target, Session next) =>
      target.emit(target.state.copyWith(signedIn: next.signedIn));

  final following = ctx.each(session.stream, take); // what happens next
  take(ctx, session.currentState); // what has already happened
  await following.value;
});

Here Screen and Session are application states with a signedIn property. The subscription and the read stand on two lines with no await between them, so no change of the session falls between the two. The subscription comes first: take emits, the listeners of the screen hear the write at once, and one of them may change the session right there. Subscribed already, the child hears that change; a subscription made after the read would miss it.

The subscription keeps the following controller’s queue occupied until it ends: no other job of that controller starts before the session closes its stream, and a draining close waits for the same. So this job suits a controller that does nothing but follow. A controller with other jobs to run holds a subscription of its own instead, made in its constructor and cancelled in onClose. For each update that subscription brings, the controller either queues a short job or, when the update is a fact the running work has to hear at once, writes it with externalSetState. External state on the state page shows such a subscription and says which of the two an update calls for.