Add events in barback to bubble up the dirty bit.

pkg/barback/lib/src/stream_pool.dart

Index: pkg/barback/lib/src/stream_pool.dart
diff --git a/pkg/barback/lib/src/stream_pool.dart b/pkg/barback/lib/src/stream_pool.dart
new file mode 100644
index 0000000000000000000000000000000000000000..0e8a83d3033e866773331ae20c0f78f8ba3c64a3
--- /dev/null
+++ b/pkg/barback/lib/src/stream_pool.dart
@@ -0,0 +1,65 @@
+// Copyright (c) 2013, 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 barback.stream_pool;
+
+import 'dart:async';
+
+/// A pool of streams whose events are unified and emitted through a central
+/// stream.
+class StreamPool<T> {
+  /// The stream through which all events from streams in the pool are emitted.
+  Stream<T> get stream => _controller.stream;
+  final StreamController<T> _controller;
+
+  /// Subscriptions to the streams that make up the pool.
+  final _subscriptions = new Map<Stream<T>, StreamSubscription<T>>();
+
+  /// Creates a new stream pool that only supports a single subscriber.
+  ///
+  /// Any events from broadcast streams in the pool will be buffered until a
+  /// listener is subscribed.
+  StreamPool()
+      // Create the controller as sync so that any sync input streams will be
+      // forwarded synchronously. Async input streams will have their asynchrony
+      // preserved, since _controller.add will be called asynchronously.
+      : _controller = new StreamController<T>(sync: true);
+
+  /// Creates a new stream pool where [stream] can be listened to more than
+  /// once.
+  ///
+  /// Any events from buffered streams in the pool will be emitted immediately,
+  /// regardless of whether [stream] has any subscribers.
+  StreamPool.broadcast()
+      // Create the controller as sync so that any sync input streams will be
+      // forwarded synchronously. Async input streams will have their asynchrony
+      // preserved, since _controller.add will be called asynchronously.
+      : _controller = new StreamController<T>.broadcast(sync: true);
+
+  /// Adds [stream] as a member of this pool.
+  ///
+  /// Any events from [stream] will be emitted through [this.stream]. If
+  /// [stream] is sync, they'll be emitted synchronously; if [stream] is async,
+  /// they'll be emitted asynchronously.
+  void addStream(Stream<T> stream) {

[Bob Nystrom, 2013/08/01 18:10:33]

How about just "add()" (and "remove()" below)?

[nweiz, 2013/08/01 20:30:52]

Done.

+    if (_subscriptions.containsKey(stream)) return;
+    _subscriptions[stream] = stream.listen(_controller.add,
+        onError: _controller.addError,
+        onDone: () => removeStream(stream));
+  }
+
+  /// Removes [stream] as a member of this pool.
+  void removeStream(Stream<T> stream) {
+    var subscription = _subscriptions.remove(stream);
+    if (subscription != null) subscription.cancel();
+  }
+
+  /// Removes all streams from this pool and closes [stream].
+  void close() {
+    for (var subscription in _subscriptions.values) {
+      subscription.cancel();
+    }

[Bob Nystrom, 2013/08/01 18:10:33]

_subscriptions.clear();

[nweiz, 2013/08/01 20:30:52]

Done.

+    _controller.close();
+  }
+}