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

lib/src/stream_queue.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 file.
 
 library async.stream_events;
 
 import 'dart:async';
 import 'dart:collection';
 
-import "forkable_stream.dart";
+import "subscription_fork.dart";
 import "subscription_stream.dart";
 import "stream_completer.dart";
 import "../result.dart";
 
 /// An asynchronous pull-based interface for accessing stream events.
 ///
 /// Wraps a stream and makes individual events available on request.
 ///
 /// You can request (and reserve) one or more events from the stream,
 /// and after all previous requests have been fulfilled, stream events
@@ skipping 51 matching lines @@
   // again when new events are available.
   // The request can remove events that it uses, or keep them in the event
   // queue until it has all that it needs.
   //
   // This model is very flexible and easily extensible.
   // It allows requests that don't consume events (like [hasNext]) or
   // potentially a request that takes either five or zero events, determined
   // by the content of the fifth event.
 
   /// Source of events.
-  final ForkableStream _sourceStream;
+  final Stream _sourceStream;
 
   /// Subscription on [_sourceStream] while listening for events.
   ///
   /// Set to subscription when listening, and set to `null` when the
   /// subscription is done (and [_isDone] is set to true).
   StreamSubscription _subscription;
 
   /// Whether we have listened on [_sourceStream] and the subscription is done.
   bool _isDone = false;
 
   /// Whether a closing operation has been performed on the stream queue.
   ///
   /// Closing operations are [cancel] and [rest].
   bool _isClosed = false;
 
   /// Queue of events not used by a request yet.
   final Queue<Result> _eventQueue = new Queue();
 
   /// Queue of pending requests.
   ///
   /// Access through methods below to ensure consistency.
   final Queue<_EventRequest> _requestQueue = new Queue();
 
   /// Create a `StreamQueue` of the events of [source].
   StreamQueue(Stream source)
-      : _sourceStream = source is ForkableStream
-          ? source
-          : new ForkableStream(source);
+      : _sourceStream = source;
 
   /// Asks if the stream has any more events.
   ///
   /// Returns a future that completes with `true` if the stream has any
   /// more events, whether data or error.
   /// If the stream closes without producing any more events, the returned
   /// future completes with `false`.
   ///
   /// Can be used before using [next] to avoid getting an error in the
   /// future returned by `next` in the case where there are no more events.
@@ skipping 102 matching lines @@
   ///
   /// The fork is subscribed to the same underlying stream as this queue, but
   /// it's otherwise wholly independent. If requests are made on one, they don't
   /// move the other forward; if one is closed, the other is still open.
   ///
   /// The underlying stream will only be paused when all forks have no
   /// outstanding requests, and only canceled when all forks are canceled.
   StreamQueue<T> fork() {
     if (_isClosed) throw _failClosed();
 
+    _ensureListening();
+    if (_subscription is! ForkableSubscription) {
+      _subscription = new ForkableSubscription<T>(_subscription);
+      _subscription.onData(_onData);
+      _subscription.onError(_onError);
+      _subscription.onDone(_onDone);
+    }
     var request = new _ForkRequest<T>(this);
     _addRequest(request);
     return request.queue;
   }
 
   /// Cancels the underlying stream subscription.
   ///
   /// If [immediate] is `false` (the default), the cancel operation waits until
   /// all previously requested events have been processed, then it cancels the
   /// subscription providing the events.
@@ skipping 429 matching lines @@
 
   void close(Queue<Result> events) {
     _completeStream(events);
   }
 
   void _completeStream(Queue<Result> events) {
     if (events.isEmpty) {
       if (_streamQueue._isDone) {
         _completer.setEmpty();
       } else {
-        _completer.setSourceStream(_streamQueue._sourceStream.fork());
+        assert(_streamQueue._subscription is ForkableSubscription);
+        var stream = new SubscriptionStream<T>(_streamQueue._subscription.fork());
+        _completer.setSourceStream(stream);
       }
     } else {
       // There are prefetched events which need to be added before the
       // remaining stream.
       var controller = new StreamController<T>();
       for (var event in events) {
         event.addTo(controller);
       }
 
-      var fork = _streamQueue._sourceStream.fork();
+      var fork = new SubscriptionStream(_streamQueue._subscription.fork());
       controller.addStream(fork, cancelOnError: false)
           .whenComplete(controller.close);
       _completer.setSourceStream(controller.stream);
     }
   }
 }