Chromium Code Reviews
chromiumcodereview-hr@appspot.gserviceaccount.com (chromiumcodereview-hr) | Please choose your nickname with Settings | Help | Chromium Project | Gerrit Changes | Sign out
(2114)

Unified Diff: mojo/public/dart/third_party/barback/lib/src/utils/stream_pool.dart

Issue 1346773002: Stop running pub get at gclient sync time and fix build bugs (Closed) Base URL: git@github.com:domokit/mojo.git@master
Patch Set: Created 5 years, 3 months ago
Use n/p to move between diff chunks; N/P to move between comments. Draft comments are only viewable by you.
Jump to:
View side-by-side diff with in-line comments
Download patch
Index: mojo/public/dart/third_party/barback/lib/src/utils/stream_pool.dart
diff --git a/mojo/public/dart/third_party/barback/lib/src/utils/stream_pool.dart b/mojo/public/dart/third_party/barback/lib/src/utils/stream_pool.dart
new file mode 100644
index 0000000000000000000000000000000000000000..679672d2547e00db179138076efe7db2669345a2
--- /dev/null
+++ b/mojo/public/dart/third_party/barback/lib/src/utils/stream_pool.dart
@@ -0,0 +1,66 @@
+// 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.utils.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 add(Stream<T> stream) {
+ if (_subscriptions.containsKey(stream)) return;
+ _subscriptions[stream] = stream.listen(_controller.add,
+ onError: _controller.addError,
+ onDone: () => remove(stream));
+ }
+
+ /// Removes [stream] as a member of this pool.
+ void remove(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();
+ }
+ _subscriptions.clear();
+ _controller.close();
+ }
+}

Powered by Google App Engine
This is Rietveld 408576698