Bug a bug in StreamQueueTransaction.reject(). (#22)

This would throw a StateError if StreamQueue.rest had been called on
one of the transaction's child queues, because that queue didn't expect
to be canceled later on.

CHANGELOG.md

@@ -3,6 +3,9 @@
 * Add a `collectBytes` function which collects list-of-byte events into
   a single byte list.
 
+* Fix a bug where rejecting a `StreamQueueTransaction` would throw a
+  `StateError` if `StreamQueue.rest` had been called on one of its child queues.
+
 ## 1.12.0
 
 * Add an `AsyncCache` class that caches asynchronous operations for a period of

lib/src/stream_queue.dart

@@ -548,14 +548,14 @@ class _StreamQueue<T> extends StreamQueue<T> {
     if (_isDone) {
       return new Stream<T>.empty();
     }
+    _isDone = true;
 
     if (_subscription == null) {
       return _sourceStream;
     }
 
     var subscription = _subscription;
     _subscription = null;
-    _isDone = true;
 
     var wasPaused = subscription.isPaused;
     var result = new SubscriptionStream<T>(subscription);
@@ -960,7 +960,7 @@ class _HasNextRequest<T> implements _EventRequest<T> {
 /// Request for a [StreamQueue.startTransaction] call.
 ///
 /// This request isn't complete until the user calls
-/// [StreamQueueTransaction.commit] or [StreamQueue.rejectTransaction], at which
+/// [StreamQueueTransaction.commit] or [StreamQueueTransaction.reject], at which
 /// point it manually removes itself from the request queue and calls
 /// [StreamQueue._updateRequests].
 class _TransactionRequest<T> implements _EventRequest<T> {

pubspec.yaml

@@ -1,5 +1,5 @@
 name: async
-version: 1.13.0
+version: 1.13.0-dev
 author: Dart Team <misc@dartlang.org>
 description: Utility functions and classes related to the 'dart:async' library.
 homepage: https://www.github.com/dart-lang/async

test/stream_queue_test.dart

@@ -863,6 +863,30 @@ main() {
         transaction.reject();
       });
 
+      // Regression test.
+      test("pending child rest requests emit no more events", () async {
+        var controller = new StreamController();
+        var events = new StreamQueue(controller.stream);
+        var transaction = events.startTransaction();
+        var queue = transaction.newQueue();
+
+        // This should emit no more events after the transaction is rejected.
+        queue.rest.listen(expectAsync1((_) {}, count: 3),
+            onDone: expectAsync0(() {}, count: 0));
+
+        controller.add(1);
+        controller.add(2);
+        controller.add(3);
+        await flushMicrotasks();
+
+        transaction.reject();
+        await flushMicrotasks();
+
+        // These shouldn't affect the result of `queue.rest.toList()`.
+        controller.add(4);
+        controller.add(5);
+      });
+
       test("child requests' cancel() may still be called explicitly", () async {
         transaction.reject();
         await queue1.cancel();