Add new features to package:async.

lib/src/delegating_stream_subscription.dart

Index: lib/src/delegating_stream_subscription.dart
diff --git a/lib/src/delegating_stream_subscription.dart b/lib/src/delegating_stream_subscription.dart
new file mode 100644
index 0000000000000000000000000000000000000000..89f210c3a7e4e9bf89730f004a8abcced78ccdfd
--- /dev/null
+++ b/lib/src/delegating_stream_subscription.dart
@@ -0,0 +1,88 @@
+// 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.delegating_stream_subscription;
+
+import 'dart:async';
+
+/// Base class for delegating stream subscriptions.
+abstract class _DelegatingSubscriptionBase<T>
+         implements StreamSubscription<T> {
+  StreamSubscription get _source;
+
+  void onData(void handleData(T data)) {
+    _source.onData(handleData);
+  }
+
+  void onError(Function handleError) {
+    _source.onError(handleError);
+  }
+
+  void onDone(void handleDone()) {
+    _source.onDone(handleDone);
+  }
+
+  void pause([Future resumeFuture]) {
+    _source.pause(resumeFuture);
+  }
+
+  void resume() {
+    _source.resume();
+  }
+
+  Future cancel() => _source.cancel();
+
+  Future asFuture([futureValue]) => _source.asFuture(futureValue);
+
+  bool get isPaused => _source.isPaused;
+}
+
+/// Simple delegating wrapper around a [StreamSubscription].
+///
+/// Subclasses can override individual methods.
+class DelegatingStreamSubscription<T> extends _DelegatingSubscriptionBase<T> {
+  final StreamSubscription _source;
+
+  DelegatingStreamSubscription(StreamSubscription subscription)
+      : _source = subscription;
+}
+
+/// A delegating wrapper around a [StreamSubscription], which can be changed.
+///
+/// Like a [DelegatingStreamSubscription] except that the subscription
+/// being delegated to can be changed after the wrapper has been created
+/// using the static [setDelegate] methods.
+///
+/// This class can be used if a stream subscription needs to act as different
+/// other subscriptions at different times, but act as a single subscription
+/// outwards.
+class _MutableDelegatingSubscription<T> extends _DelegatingSubscriptionBase<T> {
+  /// The subscription being delegated to.
+  ///
+  /// Can be changed by a [MutableDelegatingStreamSubscriptionController].
+  StreamSubscription _source;
+
+  _MutableDelegatingSubscription(this._source);
+}
+
+// A delegating stream subscription and a way to update its source subscription.
+class MutableDelegatingStreamSubscriptionController<T> {

[nweiz, 2015/06/16 01:05:23]

I think it's quite unlikely that people will want to use this outside this
package. It would be nice to keep it package-private so we don't commit to
maintaining it as part of the public API.

[Lasse Reichstein Nielsen, 2015/06/18 12:10:12]

I'm not using it any more, so I'll just remove it.

+  /// A delegating stream subscription that delegates to a stream subscription.
+  ///
+  /// When this controller is created, the subscription delegates to the
+  /// subscription provided in the constructor.
+  /// After that it can be changed using the [setSourceSubscription] function.
+  final StreamSubscription<T> subscription;
+
+  /// Create a controller with a [subscription] delegating to [source].
+  MutableDelegatingStreamSubscriptionController(StreamSubscription source)
+      : subscription = new _MutableDelegatingSubscription<T>(source);
+
+  /// The current subscription being delegated to.
+  StreamSubscription get sourceSubscription => subscription._source;
+
+  void set sourceSubscription(StreamSubscription source) {
+    subscription._source = source;
+  }
+}