Add StreamSubscription.asFuture.

sdk/lib/async/stream_pipe.dart

 // Copyright (c) 2012, 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.
 
 part of dart.async;
 
 /** Utility function to create an [AsyncError] if [error] isn't one already. */
 AsyncError _asyncError(Object error, Object stackTrace, [AsyncError cause]) {
   if (error is AsyncError) return error;
   if (cause == null) return new AsyncError(error, stackTrace);
@@ skipping 113 matching lines @@
   void onDone(void handleDone()) {
     if (handleDone == null) handleDone = _nullDoneHandler;
     _onDone = handleDone;
   }
 
   void pause([Future resumeSignal]);
 
   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;
+  }
 }
 
 
 /**
  * Abstract superclass for subscriptions that forward to other subscriptions.
  */
 class _ForwardingStreamSubscription<S, T>
     extends _BaseStreamSubscription<T> implements _EventOutputSink<T> {
   final _ForwardingStream<S, T> _stream;
   final bool _unsubscribeOnError;
@@ skipping 371 matching lines @@
 
   void handleError(AsyncError error, EventSink<T> sink) {
     _handleError(error, sink);
   }
 
   void handleDone(EventSink<T> sink) {
     _handleDone(sink);
   }
 }