_StreamQueue._cancel() causes Stream StateError

[gamebox]

In the process of building an incremental test runner for Flutter, I've come across an interesting bug. I use the AnsiTerminal terminal defined as a global getter inside the flutter_tools package. Being an AnsiTerminal it is bound to stdin.

We use the test running facilities provided in dart-lang/test via package:test_core/src/executable.dart. Their io libary binds to stdin lazily through a StreamQueue(after a transform that splits the lines). We never cause the binding to happen, and things work well during the invocation. But when we nicely close out all of our resources when we go to shutdown the process, it blows up with a StateError do to these two lines in _StreamQueue._cancel()

if (_subscription == null) _subscription = _sourceStream.listen(null);
var future = _subscription.cancel();

I'm trying to understand the rationale behind this, and maybe what strategy I can use to work around it. Removing those lines seems to have no affect on this narrow case, but I'm sure that this has some meaningful semantics in other context.

[kevmoo]

CC @lrhn

[gamebox]

For the moment, I'm going to just wrap the call in a try/catch/finally. Again, this is happening when the process is closing down.

[natebosch]

The stream needs to be listened to once in order to forward the return value from StreamSubscription.cancel().

Here is one test that fails if we don't listen first:

async/test/stream_queue_test.dart

Lines 484 to 490 in 30e9ccc

	test("cancels underlying subscription when called before any event",
	() async {
	var cancelFuture = new Future.value(42);
	var controller = new StreamController<int>(onCancel: () => cancelFuture);
	var events = new StreamQueue<int>(controller.stream);
	expect(await events.cancel(), 42);
	});

I think in practice very few onCancel callbacks return a useful value, and the documentation for StreamSubscription.cancel says the future should only ever complete with null.

Changing this behavior would be breaking. Given the current documentation it might be preferable to change it, but I'm not sure if it's worth the breaking change.

I think, though, in general we don't really want to support the use case of creating multiple StreamQueue instances off the same non-broadcast stream as long as you only ever request events from one of them... We should probably take a look at whether we can do something different in the test package.

[gamebox]

Yeah, my approach described above works for now, but it's definitely a hack.

An alternative that may make sense is to change the implementation of stdin in dart:io package to be a BroadcastStream(which, for what it's worth should be the default IMO). I can't think of bad behavior that could be caused by multiple consumption of STDIN. If someone can think of some, please let me know.

I talked to @mit-mit about having someone from Dart lang team look at this.

[lrhn]

So the problem is that you have two different clients trying to use stdin, you are just lucky that only one of them actually do something with it.

I guess the test code could be changed to not cancel the stream unless it has actually been used (it would then have to record that somewhere).
Personally, I'd probably initialize the StreamQueue lazily the first time it's actually needed, instead of creating it ahead of time. Then you would only need to cancel the queue if it exists.

[gamebox]

@lrhn

Yeah, lazy initialization is not very pretty in dart, but I've got a patch to do exactly this.

But I'm still confused by why you think we would be 'lucky' if someone used the STDIN stream only once. To be honest, I'm actually confused by the semantics of having a stream be only listened to once period. To me, that's like saying you can only only iterate a list once. The current API requires I know way too much about the implementation of every other thing in not only my program, but also all of it's dependencies - possibly even the Dart standard library.

Developers should be able to use a Stream and not worry about StateErrors that that stem from behavior outside of their own application code, especially since these sort of Effects are not transparent in the type system.

The following should probably be cross posted in an issue on dart-lang/sdk:

Some of this would be easier if you could ask a Stream whether it's been listened to or not, but with the current API, I need to wrap a call to listen() for any stream I do not explicitly control with try/catch.

A more sensible API would be to have listen() return a subscription if the listener is added, and null if not. That's a very simple, straightforward way for a developer to at least handle the case - which will be more elegant once we have NNTBD, and therefore the nullability will be encoded in the type signature. What's more, single subscription streams should probably be the exception rather than the default. Alternatively, Stream should be the name of the abstract class and the current Stream class should be called SinglecastStream or something similar.

@natebosch

I don't really want to create multiple StreamQueue's off of a Singlecast stream. I just want the semantics of the cancel method to be maintained in the close method of StreamQueue, direct from the docs for StreamSubscription.cancel:

A returned future completes with a null value. If the cleanup throws, which it really shouldn't, the returned future completes with that error.

So, in my mind that means return the future from cancelling the current subscription if one exists, or a Future otherwise. So something roughly like this:

Future _cancel() {
    if (_isDone) return null;
    var future = _subscription != null ? Future.value(null) : _subscription.cancel();
    _close();
    return future;
}

[natebosch]

just want the semantics of the cancel method to be maintained in the close method of StreamQueue, direct from the docs for StreamSubscription.cancel

I think that would be the right approach too - but I also don't think it's worth the breaking change.

[ened]

@natebosch has there been development or alternative ideas on this issue?

[natebosch]

@ened - I'll tag with "next breaking release" so that if we have some other compelling reason to bump the version here we can fix this along with it - as I mentioned earlier I don't think it's worth a major version release for just this change.

The recommendation is still to avoid wrapping a single subscriber stream in a StreamQueue multiple times, and closing multiple times, when you must only listen once.

[lrhn]

For the record, the behavior is intentional. We want the stream to be cleaned up when you close the stream queue.
In other situations, we've had requests to do exactly that because they would otherwise leak resources when a stream was not listened to and closed, so generally we let stream consumers always consume the stream, even if you read no events at all.

[lrhn]

No change planned.