Add an AssetStream class to Barback.

pkg/barback/lib/src/asset.dart

Index: pkg/barback/lib/src/asset.dart
diff --git a/pkg/barback/lib/src/asset.dart b/pkg/barback/lib/src/asset.dart
index db7a5ea2ca8c0788511ee1eed80f8c504faaab72..a60296e2e72d03f4001d2919d89977ab3f17116f 100644
--- a/pkg/barback/lib/src/asset.dart
+++ b/pkg/barback/lib/src/asset.dart
@@ -10,6 +10,7 @@ import 'dart:io';
 import 'dart:utf';
 
 import 'asset_id.dart';
+import 'stream_replayer.dart';
 import 'utils.dart';
 
 /// A blob of content.
@@ -34,6 +35,12 @@ abstract class Asset {
   factory Asset.fromPath(AssetId id, String path) =>
       new _FileAsset(id, new File(path));
 
+  /// Creates an asset from a stream.
+  ///
+  /// This immediately starts draining [stream].
+  factory Asset.fromStream(AssetId id, Stream<List<int>> stream) =>
+      new _StreamAsset(id, stream);
+
   /// Returns the contents of the asset as a string.
   ///
   /// If the asset was created from a [String] the original string is always
@@ -140,3 +147,22 @@ class _StringAsset extends Asset {
         .replaceAll("\t", r"\t");
   }
 }
+
+/// An asset whose data is available from a stream.
+class _StreamAsset extends Asset {
+  /// A stream replayer that records and replays the contents of the input
+  /// stream.
+  final StreamReplayer<List<int>> _replayer;
+
+  _StreamAsset(AssetId id, Stream<List<int>> stream)
+      : _replayer = new StreamReplayer(stream),
+        super(id);
+
+  Future<String> readAsString({Encoding encoding}) {
+    if (encoding == null) encoding = Encoding.UTF_8;
+    return _replayer.getReplay().toList()
+        .then((chunks) => encoding.decode(flatten(chunks)));
+  }
+
+  Stream<List<int>> read() => _replayer.getReplay();
+}