Remove Signal class and use Future/.whenComplete instead.

sdk/lib/async/stream_impl.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;
 
 // States shared by single/multi stream implementations.
 
 /// Initial and default state where the stream can receive and send events.
 const int _STREAM_OPEN = 0;
@@ skipping 198 matching lines @@
 
   /**
    * Record that a listener wants a pause from events.
    *
    * This methods is called from [_StreamListener.pause()].
    * Subclasses can override this method, along with [isPaused] and
    * [createSubscription], if they want to do a different handling of paused
    * subscriptions, e.g., a filtering stream pausing its own source if all its
    * subscribers are paused.
    */
-  void _pause(_StreamListener<T> listener, Signal resumeSignal) {
+  void _pause(_StreamListener<T> listener, Future resumeSignal) {
     assert(identical(listener._source, this));
     if (!listener._isSubscribed) {
       throw new StateError("Subscription has been canceled.");
     }
     assert(!_isComplete);  // There can be no subscribers when complete.
     bool wasPaused = _isPaused;
     _incrementPauseCount(listener);
     if (resumeSignal != null) {
-      resumeSignal.then(() { this._resume(listener, true); });
+      resumeSignal.whenComplete(() { this._resume(listener, true); });
     }
     if (!wasPaused) {
       _onPauseStateChange();
     }
   }
 
   /** Stops pausing due to one request from the given listener. */
   void _resume(_StreamListener<T> listener, bool fromEvent) {
     if (!listener.isPaused) return;
     assert(listener._isSubscribed);
@@ skipping 438 matching lines @@
   }
 
   void _sendDone() {
     _onDone();
   }
 
   void cancel() {
     _source._cancel(this);
   }
 
-  void pause([Signal resumeSignal]) {
+  void pause([Future resumeSignal]) {
     _source._pause(this, resumeSignal);
   }
 
   void resume() {
     if (!isPaused) {
       throw new StateError("Resuming unpaused subscription");
     }
     _source._resume(this, false);
   }
 }
@@ skipping 278 matching lines @@
   }
 
   bool get _isComplete => _timer == null && _pauseCount == 0;
 
   void onData(void handleAction(T value)) {}
   void onError(void handleError(StateError error)) {}
   void onDone(void handleDone(T value)) {
     _handler = handleDone;
   }
 
-  void pause([Signal signal]) {
+  void pause([Future signal]) {
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
     }
     if (_timer != null) _timer.cancel();
     _pauseCount++;
   }
 
   void resume() {
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
@@ skipping 11 matching lines @@
     if (_isComplete) {
       throw new StateError("Subscription has been canceled.");
     }
     if (_timer != null) {
       _timer.cancel();
       _timer = null;
     }
     _pauseCount = 0;
   }
 }