Alternative implementation of StreamQueue.fork which has no impact if it

test/stream_queue_test.dart

 // Copyright (c) 2015, the Dart project authors.  Please see the AUTHORS file
 // for details. All rights reserved. Use of this source code is governed by a
 // BSD-style license that can be found in the LICENSE filevents.
 
 import "dart:async";
 
 import "package:async/async.dart" show StreamQueue;
 import "package:test/test.dart";
 
 import "utils.dart";
@@ skipping 854 matching lines @@
         expect(await queue2.hasNext, isFalse);
       });
 
       test("the underlying stream is only canceled once all forks are canceled",
           () async {
         var controller = new StreamController();
         var queue1 = new StreamQueue<int>(controller.stream);
         var queue2 = queue1.fork();
 
         await flushMicrotasks();
-        expect(controller.hasListener, isFalse);
+        expect(controller.hasListener, isTrue);
 
         expect(queue1.next, completion(1));
         await flushMicrotasks();
         expect(controller.hasListener, isTrue);
 
         queue2.cancel();
         await flushMicrotasks();
         expect(controller.hasListener, isTrue);
 
         controller.add(1);
@@ skipping 49 matching lines @@
           expect(await queue2.hasNext, isFalse);
         });
 
         test("the underlying stream is only canceled once all forks are "
             "canceled", () async {
           var controller = new StreamController();
           var queue1 = new StreamQueue<int>(controller.stream);
           var queue2 = queue1.fork();
 
           await flushMicrotasks();
-          expect(controller.hasListener, isFalse);
+          expect(controller.hasListener, isTrue);
 
           expect(queue1.next, throwsStateError);
           await flushMicrotasks();
           expect(controller.hasListener, isTrue);
 
           queue2.cancel(immediate: true);
           await flushMicrotasks();
           expect(controller.hasListener, isTrue);
 
           queue1.cancel(immediate: true);
@@ skipping 60 matching lines @@
         expect(await queue2.hasNext, isFalse);
       });
 
       test("the underlying stream is only paused when all forks are paused",
           () async {
         var controller = new StreamController();
         var queue1 = new StreamQueue<int>(controller.stream);
         var queue2 = queue1.fork();
 
         await flushMicrotasks();
-        expect(controller.hasListener, isFalse);
+        expect(controller.hasListener, isTrue);
 
         var sub1 = queue1.rest.listen(null);
         await flushMicrotasks();
         expect(controller.hasListener, isTrue);
         expect(controller.isPaused, isFalse);
 
         sub1.pause();
         await flushMicrotasks();
         expect(controller.isPaused, isTrue);
 
@@ skipping 85 matching lines @@
     controller.add(4);
     await flushMicrotasks();
     controller.close();
   }();
   return controller.stream;
 }
 
 Stream<int> createLongStream(int eventCount) async* {
   for (int i = 0; i < eventCount; i++) yield i;
 }