Handle the case of Stream.take(0) better.

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;
 
 /** Runs user code and takes actions depending on success or failure. */
 _runUserCode(userCode(),
              onSuccess(value),
              onError(error, StackTrace stackTrace)) {
@@ skipping 308 matching lines @@
     // This test is done early to avoid handling an async error
     // in the _handleData method.
     if (count is! int) throw new ArgumentError(count);
   }
 
   StreamSubscription<T> _createSubscription(
       void onData(T data),
       Function onError,
       void onDone(),
       bool cancelOnError) {
+    if (_count == 0) {
+      return new _DoneSubscription<T>(onDone);
+    }
     return new _StateStreamSubscription<T>(
         this, onData, onError, onDone, cancelOnError, _count);
   }
 
   void _handleData(T inputEvent, _EventSink<T> sink) {
     _StateStreamSubscription<T> subscription = sink;
     int count = subscription._count;
     if (count > 0) {
       sink._add(inputEvent);
       count -= 1;
@@ skipping 147 matching lines @@
         _addErrorWithReplacement(sink, e, s);
         return null;
       }
       if (!isEqual) {
         sink._add(inputEvent);
         _previous = inputEvent;
       }
     }
   }
 }