Why does Stream.first need to wait for the subscription cancel?

[natebosch]

sdk/sdk/lib/async/stream.dart

Lines 1167 to 1173 in 3dcb5d1

	Future<T> get first {
	_Future<T> future = new _Future<T>();
	StreamSubscription subscription;
	subscription = this.listen(
	(T value) {
	_cancelAndValue(subscription, future, value);
	},

sdk/sdk/lib/async/stream_pipe.dart

Lines 58 to 65 in 3dcb5d1

	void _cancelAndValue(StreamSubscription subscription, _Future future, value) {
	var cancelFuture = subscription.cancel();
	if (cancelFuture is Future && !identical(cancelFuture, Future._nullFuture)) {
	cancelFuture.whenComplete(() => future._complete(value));
	} else {
	future._complete(value);
	}
	}

We don't handle or forward errors from cancelFuture - why do we need to wait for it at all before we can return the first value? This leads to some confusing behavior like:

Stream<int> stuff() async* {
  yield 1;
  await Future.delayed(Duration(seconds: 5));
}

void main() {
  stuff().first.then((v) => print(v)); // takes 5 seconds to print, even though there is a value right away.
}

This was added when StreamSubscription.cancel started returning a Future in https://codereview.chromium.org//18915008 - but a scan of the comments didn't help me find the reasoning for APIs like first or elementAt to wait for the cancelation to be finished.

cc @lrhn

[lrhn]

First of all, the example should not wait 5 seconds as written. That's an implementation bug - it should deliver its yield value to the stream listener before checking whether it's been cancelled, so you have a window to pause or cancel immediately when you receive an event to prevent the async function from continuing work.

That said, you could just put the delay into a finally block, then you'd see the the same behavior even if everybody follows the spec.

The reason for waiting for the cancel to finish is really that we prefer any operation that cancels a stream to wait for the stream cancel. It ensures the user that the stream operation is completely done and its resources are released, which makes code less prone to race conditions.
If the cancel does do a lot of work, likely releasing resources, and reports that in its return future, it's probably for a reason. Ignoring it seems like the wrong thing to do.

You could argue that we should then also propagate an error during the cancel.
That might even be true - I can see arguments for either behavior (all boiling down to errors in cancel operations being unexpected anyway).

[natebosch]

That's an implementation bug

Do we have tracking for it?

I'm a little confused by what the next statements mean...

it should deliver its yield value to the stream listener before checking whether it's been cancelled

The yield does deliver the value immediately, and at that moment it hasn't been canceled.

so you have a window to pause or cancel immediately when you receive an event to prevent the async function from continuing work.

I think I understand this to mean that before continuing work after a yield the async* function should check if it's been canceled - whereas today it's checking if it is canceled only immediately before the next yield. Is that correct?

The rest of the explanation for waiting for the cancel is satisfying to me. Should we close this issue or turn it into a tracking issue for changing the implementation of async*?

[lrhn]

Yes, that is a correct summary, at least for some implementations - I'm not sure all implementations act the same way wrt. yield.

The way it's stated, the event of the yield is not considered "delivered" until the listener callback has been called.
That callback can then cancel or pause the subscription, and if it does so immediately (synchronously, not in a later microtask), the yield implementation should not continue with the following code.
If cancelled, it should return at that yield, if paused, it should stay at that yield until resumed.

That's how the language currently specifies the behavior.
This was clarified in the Dart 2 revision (it was fairly ambiguous about the scheduling before).

I'll close this, and check if we have an open issue for the current behavior.