Add an AssetStream class to Barback.

pkg/barback/lib/src/utils.dart

 // 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;
 
 import 'dart:async';
 
 /// A pair of values.
 class Pair<E, F> {
   E first;
   F last;
 
   Pair(this.first, this.last);
 
   String toString() => '($first, $last)';
 
   bool operator==(other) {
     if (other is! Pair) return false;
     return other.first == first && other.last == last;
   }
 
   int get hashCode => first.hashCode ^ last.hashCode;
 }
 
+/// A class that represents one and only one of two types of values.
+class Either<E, F> {
+  /// Whether this is a value of type `E`.
+  final bool isFirst;
+
+  /// Whether this is a value of type `F`.
+  bool get isSecond => !isFirst;
+
+  /// The value, either of type `E` or `F`.
+  final _value;
+
+  /// The value of type `E`.
+  ///
+  /// It's an error to access this is this is of type `F`.
+  E get first {
+    assert(isFirst);
+    return _value;
+  }
+
+  /// The value of type `F`.
+  ///
+  /// It's an error to access this is this is of type `E`.
+  F get second {
+    assert(isSecond);
+    return _value;
+  }
+
+  /// Creates an [Either] with type `E`.
+  Either.withFirst(this._value)
+      : isFirst = true;
+
+  /// Creates an [Either] with type `F`.
+  Either.withSecond(this._value)
+      : isFirst = false;
+
+  /// Runs [whenFirst] or [whenSecond] depending on the type of [this].
+  ///
+  /// Returns the result of whichvever function was run.
+  match(whenFirst(E value), whenSecond(F value)) {
+    if (isFirst) return whenFirst(first);
+    return whenSecond(second);
+  }
+
+  String toString() => "$_value (${isFirst? 'first' : 'second'})";
+}
+
 /// Converts a number in the range [0-255] to a two digit hex string.
 ///
 /// For example, given `255`, returns `ff`.
 String byteToHex(int byte) {
   assert(byte >= 0 && byte <= 255);
 
   const DIGITS = "0123456789abcdef";
   return DIGITS[(byte ~/ 16) % 16] + DIGITS[byte % 16];
 }
 
@@ skipping 34 matching lines @@
 /// Passes each key/value pair in [map] to [fn] and returns a new [Map] whose
 /// values are the return values of [fn].
 Map mapMapValues(Map map, fn(key, value)) =>
   new Map.fromIterable(map.keys, value: (key) => fn(key, map[key]));
 
 /// Returns whether [set1] has exactly the same elements as [set2].
 bool setEquals(Set set1, Set set2) =>
   set1.length == set2.length && set1.containsAll(set2);
 
 /// Merges [streams] into a single stream that emits events from all sources.
-Stream mergeStreams(Iterable<Stream> streams) {
+///
+/// If [broadcast] is true, this will return a broadcast stream; otherwise, it
+/// will return a buffered stream.
+Stream mergeStreams(Iterable<Stream> streams, {bool broadcast: false}) {
   streams = streams.toList();
   var doneCount = 0;
   // Use a sync stream to preserve the synchrony behavior of the input streams.
   // If the inputs are sync, then this will be sync as well; if the inputs are
   // async, then the events we receive will also be async, and forwarding them
   // sync won't change that.
-  var controller = new StreamController(sync: true);
+  var controller = broadcast ? new StreamController.broadcast(sync: true)
+      : new StreamController(sync: true);
 
   for (var stream in streams) {
     stream.listen((value) {
       controller.add(value);
     }, onError: (error) {
       controller.addError(error);
     }, onDone: () {
       doneCount++;
       if (doneCount == streams.length) controller.close();
     });
@@ skipping 24 matching lines @@
   // We use a delayed future to allow runAsync events to finish. The
   // Future.value or Future() constructors use runAsync themselves and would
   // therefore not wait for runAsync callbacks that are scheduled after invoking
   // this method.
   return new Future.delayed(Duration.ZERO, () => pumpEventQueue(times - 1));
 }
 
 /// Like [new Future], but avoids issue 11911 by using [new Future.value] under
 /// the covers.
 Future newFuture(callback()) => new Future.value().then((_) => callback());