Add StreamSubscription.asFuture.

sdk/lib/async/stream_pipe.dart

Index: sdk/lib/async/stream_pipe.dart
diff --git a/sdk/lib/async/stream_pipe.dart b/sdk/lib/async/stream_pipe.dart
index aa3da2a664aebfb6506a985d95b7aa040c2c234c..1ec11194156813be9c4d0df26bb4127cee03ae74 100644
--- a/sdk/lib/async/stream_pipe.dart
+++ b/sdk/lib/async/stream_pipe.dart
@@ -131,6 +131,19 @@ abstract class _BaseStreamSubscription<T> implements StreamSubscription<T> {
   void resume();
 
   void cancel();
+
+  Future asFuture([var futureValue]) {
+    _FutureImpl<T> result = new _FutureImpl<T>();
+
+    // Overwrite the onDone and onError handlers.
+    onDone(() { result._setValue(futureValue); });
+    onError((AsyncError error) {
+      cancel();
+      result._setError(error);
+    });
+
+    return result;
+  }
 }