Return futures on Stream.cancel when possible.

sdk/lib/async/stream_transformers.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;
 
 /**
  * Wraps an [_EventSink] so it exposes only the [EventSink] interface.
  */
 class _EventSinkWrapper<T> implements EventSink<T> {
@@ skipping 15 matching lines @@
  * the transformation. The returned sink is the transformation's input.
  */
 class _SinkTransformerStreamSubscription<S, T>
     extends _BufferingStreamSubscription<T> {
   /// The transformer's input sink.
   EventSink<S> _transformerSink;
 
   /// The subscription to the input stream.
   StreamSubscription<S> _subscription;
 
+  /// The cancelFuture (if any).
+  Future _cancelFuture;

[Lasse Reichstein Nielsen, 2016/08/01 15:23:26]

Why a field? I only see it used inside one method?

[floitsch, 2016/08/01 21:00:37]

Acknowledged.

+
   _SinkTransformerStreamSubscription(Stream<S> source,
                                      _SinkMapper<S, T> mapper,
                                      void onData(T data),
                                      Function onError,
                                      void onDone(),
                                      bool cancelOnError)
       // We set the adapter's target only when the user is allowed to send data.
       : super(onData, onError, onDone, cancelOnError) {
     _EventSinkWrapper<T> eventSink = new _EventSinkWrapper<T>(this);
     _transformerSink = mapper(eventSink);
@@ skipping 56 matching lines @@
   }
 
   void _onResume() {
     if (_isSubscribed) _subscription.resume();
   }
 
   Future _onCancel() {
     if (_isSubscribed) {
       StreamSubscription subscription = _subscription;
       _subscription = null;
-      subscription.cancel();
+      _cancelFuture = subscription.cancel();

[Lasse Reichstein Nielsen, 2016/08/01 15:23:26]

That is, why is _cancelFuture not a local variable here.

[floitsch, 2016/08/01 15:55:55]

In case someone calls `cancel` multiple times.

[Lasse Reichstein Nielsen, 2016/08/01 16:49:38]

Possibly a good point.
I'm not sure _onCancel will be called more than once, though.
In the default _BufferingStreamSubscription implementation (stream_impl.dart,
line 191) we check whether we are already canceled, and only call _cancel (which
calls _onCancel) if we arent'. We also call _cancel in _sendDone and _sendError
(if cancelOnError), which can also only happen once. The implementation buffers
the cancelFuture there, so you don't need to do it here.

[floitsch, 2016/08/01 21:00:37]

You are right.
Changed a bit more...

     }
-    return null;
+    return _cancelFuture ?? new Future.value(null);
   }
 
   void _handleData(S data) {
     try {
       _transformerSink.add(data);
     } catch (e, s) {
       _addError(e, s);
     }
   }
 
@@ skipping 179 matching lines @@
                                  void onDone(),
                                  bool cancelOnError }) {
     cancelOnError = identical(true, cancelOnError);
     StreamSubscription<T> result = _transformer(_stream, cancelOnError);
     result.onData(onData);
     result.onError(onError);
     result.onDone(onDone);
     return result;
   }
 }