Return futures on Stream.cancel when possible.

sdk/lib/async/stream.dart

 // Copyright (c) 2013, 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;
 
 // -------------------------------------------------------------------
 // Core Stream types
 // -------------------------------------------------------------------
 
@@ skipping 214 matching lines @@
           watch.start();
           timer = new Timer(period - elapsed, () {
             timer = null;
             startPeriodicTimer();
             sendEvent();
           });
         },
         onCancel: () {
           if (timer != null) timer.cancel();
           timer = null;
+          return Future._nullFuture;
         });
     return controller.stream;
   }
 
   /**
    * Creates a stream where all events of an existing stream are piped through
    * a sink-transformation.
    *
    * The given [mapSink] closure is invoked when the returned stream is
    * listened to. All events from the [source] are added into the event sink
@@ skipping 189 matching lines @@
       controller = new StreamController/*<E>*/.broadcast(
         onListen: onListen,
         onCancel: () { subscription.cancel(); },
         sync: true
       );
     } else {
       controller = new StreamController/*<E>*/(
         onListen: onListen,
         onPause: () { subscription.pause(); },
         onResume: () { subscription.resume(); },
-        onCancel: () { subscription.cancel(); },
+        onCancel: () => subscription.cancel(),
         sync: true
       );
     }
     return controller.stream;
   }
 
   /**
    * Creates a new stream with the events of a stream per original event.
    *
    * This acts like [expand], except that [convert] returns a [Stream]
@@ skipping 37 matching lines @@
       controller = new StreamController/*<E>*/.broadcast(
         onListen: onListen,
         onCancel: () { subscription.cancel(); },
         sync: true
       );
     } else {
       controller = new StreamController/*<E>*/(
         onListen: onListen,
         onPause: () { subscription.pause(); },
         onResume: () { subscription.resume(); },
-        onCancel: () { subscription.cancel(); },
+        onCancel: () => subscription.cancel(),
         sync: true
       );
     }
     return controller.stream;
   }
 
   /**
    * Creates a wrapper Stream that intercepts some errors from this stream.
    *
    * If this stream sends an error that matches [test], then it is intercepted
@@ skipping 887 matching lines @@
 abstract class StreamSubscription<T> {
   /**
    * Cancels this subscription.
    *
    * After this call, the subscription no longer receives events.
    *
    * The stream may need to shut down the source of events and clean up after
    * the subscription is canceled.
    *
    * Returns a future that is completed once the stream has finished
-   * its cleanup. May also return `null` if no cleanup was necessary.
+   * its cleanup.
+   *
+   * For historical reasons, may also return `null` if no cleanup was necessary.
+   * Returning `null` is deprecated and should be avoided.
    *
    * Typically, futures are returned when the stream needs to release resources.
    * For example, a stream might need to close an open file (as an asynchronous
    * operation). If the listener wants to delete the file after having
    * canceled the subscription, it must wait for the cleanup future to complete.
    *
    * A returned future completes with a `null` value.
    * If the cleanup throws, which it really shouldn't, the returned future
    * completes with that error.
    */
@@ skipping 283 matching lines @@
    *                   // Duplicate the data.
    *                   controller.add(data);
    *                   controller.add(data);
    *                 },
    *                 onError: controller.addError,
    *                 onDone: controller.close,
    *                 cancelOnError: cancelOnError);
    *             },
    *             onPause: () { subscription.pause(); },
    *             onResume: () { subscription.resume(); },
-   *             onCancel: () { subscription.cancel(); },
+   *             onCancel: () => subscription.cancel(),
    *             sync: true);
    *           return controller.stream.listen(null);
    *         });
    */
   const factory StreamTransformer(
       StreamSubscription<T> transformer(Stream<S> stream, bool cancelOnError))
       = _StreamSubscriptionTransformer<S, T>;
 
   /**
    * Creates a [StreamTransformer] that delegates events to the given functions.
@@ skipping 97 matching lines @@
 class _ControllerEventSinkWrapper<T> implements EventSink<T> {
   EventSink _sink;
   _ControllerEventSinkWrapper(this._sink);
 
   void add(T data) { _sink.add(data); }
   void addError(error, [StackTrace stackTrace]) {
     _sink.addError(error, stackTrace);
   }
   void close() { _sink.close(); }
 }