Add a CancelableFuture class.

lib/src/cancelable_operation.dart

Index: lib/src/cancelable_operation.dart
diff --git a/lib/src/cancelable_operation.dart b/lib/src/cancelable_operation.dart
new file mode 100644
index 0000000000000000000000000000000000000000..8095ca22e3ad5a064ed34b2215b465400244a595
--- /dev/null
+++ b/lib/src/cancelable_operation.dart
@@ -0,0 +1,148 @@
+// Copyright (c) 2015, 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.
+
+library async.cancelable_operation;
+
+import 'dart:async';
+
+import 'package:async/async.dart';
+
+/// An asynchronuos operation that can be cancelled.

[Lasse Reichstein Nielsen, 2015/08/27 10:20:54]

So basically, you have the result of an async operation (a future) and a
function that you can call to cancel the operation.

The goal of this class is to wrap that up and ensure some state management - if
the result completes after cancel is called, the result is ignored, and the
cancel function is called only once.
Then there is the asStream operation which is just added on top of the rest.

This does not require the cancelable completer class at all (it should be fairly
simple), something like:

class CancelableOperation<T> {
  final Completer<T> _completer = new Completer<T>.sync();
  final Function _cancel;
  bool _cancelled = false;
  CancelableOperation(Future<T> result, Future cancel()) : _cancel = cancel {
    result.then(_completeValue, onError: _completeError);
  }

  Future<T> get result => _completer.future;

  void _completeError(e, s) {
    if (!_cancelled) {
      _completer.completeError(e, s);
    }
  }

  void _completeValue(T v) {
    if (!_cancelled) {
      _completer.complete(v);
    } 
  }

  Future cancel() {
    if (_completer.isCompleted || _cancelled) {
      return new Future.value();  // never return null.
    }
    _cancelled = true;
    return new Future.sync(_cancel);
  }

  Stream asStream() {...}
}

So, the completer class is there as a convenience to someone who wants to create
a cancelable operation.

[nweiz, 2015/08/28 00:34:56]

All this is true, although CancelableCompleter is intentionally forefronted
because I think it's the most natural and common way to interact with a
CancelableFuture.

+///
+/// The value of this operation is exposed as [value]. When this operation is
+/// cancelled, [value] won't complete either successfully or with an error. If
+/// [value] has already completed, cancelling the operation does nothing.
+class CancelableOperation<T> {
+  /// The completer that produced this operation.
+  ///
+  /// This is canceled when [cancel] is called.
+  final CancelableCompleter<T> _completer;
+
+  Future<T> get value => _completer._inner.future;

[Lasse Reichstein Nielsen, 2015/08/27 10:20:53]

I'd call this "result" since it may be an error. We usually distinguish between
a future completing with a value or with an error.

[Lasse Reichstein Nielsen, 2015/08/27 10:20:54]

Move getter below constructor.

[nweiz, 2015/08/28 00:34:56]

"result" seems confusing when the package has a "Result" type that's unrelated.

[nweiz, 2015/08/28 00:34:56]

Done, although this is contrary to the style most packages follow, which places
getters near fields because they work the same way from an API standpoint.

[Lasse Reichstein Nielsen, 2015/09/03 13:06:17]

The order in the file is independent of how the API is presented in
documenatation. I use it find the structure of the class, which is why I prefer
to have the fields collected in one predictable place.

I thought the style guide said something about the order of members, but I can't
find it now.

+
+  /// Whether this operation has been canceled.
+  bool _canceled = false;
+
+  CancelableOperation._(this._completer);

[Lasse Reichstein Nielsen, 2015/08/27 10:57:41]

I think one thing that makes it hard for me to wrap my head around these classes
is the double dependency: operation's constructor creates the operation using a
completer, completer creates an operation.

It wasn't until I noticed this constructor that I realized why the recursion
terminated :)

It's possible to write them so that Operation stands by itself and Completer
just creates one using its own future and cancel function.

It's also slightly suspicious to me that there are two completers for each
operation. One should be enough.

[Lasse Reichstein Nielsen, 2015/08/27 10:59:47]

And it is, the other one was the cancelable completer which I misread, so ignore
this.

[nweiz, 2015/08/28 00:34:55]

That's true, but I expect exposing operations created by completers to be much
more common than those created via [fromFuture]. The current version optimizes
for that, both in terms of efficiency and in terms of the flow of the code. The
Completer has all the real functionality; the Operation is just a shell on top
to expose a specific API to the consumer.

+
+  /// Creates a [CancelableOperation] wrapping [inner].
+  ///
+  /// When this operation is canceled, [onCancel] will be called and any value
+  /// or error produced by [inner] will be discarded. The callback may return a
+  /// Future to indicate that asynchronous work has to be done to cancel the
+  /// future; this Future will be returned by [cancel].
+  factory CancelableOperation.fromFuture(Future<T> inner, {onCancel()}) {

[Lasse Reichstein Nielsen, 2015/08/27 10:20:54]

Why is onCancel optional? If it's null, I don't see how this would do anything.

Should be named "cancel". It's not something that is triggered by a cancel, but
something which does a cancel.

[nweiz, 2015/08/28 00:34:56]

From the user's perspective, canceling an operation (like canceling a stream
subscription) primarily means that the future won't fire. It may or may not make
things more efficient down the line, but that's not their primary concern. The
callback is optional because a user may want an operation that can be told not
to complete even if there's no way to take advantage of that to avoid doing
work.

Along the same lines, the class uses "canceling" to mean "telling the operation
not to complete the future". Taken that way, the callback's job isn't to cancel
the operation; the class already does that. The callback is just there so that
the code that produces the CancelableOperation can react to the cancellation if
they have some useful way to do so.

It's basically the same principle as StreamController's onCancel argument.

[Lasse Reichstein Nielsen, 2015/09/03 13:06:17]

Ack, makes sense. Cancel only cancels the "operation" object, the source can
optiohally be notified if it wants to.

+    var completer = new CancelableCompleter<T>(onCancel: onCancel);
+    completer.complete(inner);
+    return completer.operation;

[Lasse Reichstein Nielsen, 2015/08/27 10:20:54]

You are passing a callback here, which is not going to be called in response to
an async event.
For that, I'd rather provide a future that completes when you call cancel, so
you can manually add onCancel as a listener on that.

That is more general than passing one onCancel function since you can add
multiple listeners to such a future.

[nweiz, 2015/08/28 00:34:56]

I don't think it's very likely that multiple disparate pieces of code will want
to register callbacks for handling cancellation, and the way it works now makes
it possible to pipe the result of [onCancel] to the return value of [cancel].

It's also entirely parallel to StreamController, which is familiar to users and
has proven itself quite usable.

[Lasse Reichstein Nielsen, 2015/09/03 13:06:17]

True. It has to be documented when the callback will be called (and be carefully
chosen so as to not cause any reentrancy problems).

[nweiz, 2015/09/08 21:14:45]

Done.

+  }
+
+  /// Creates a [Stream] containing the result of this operation.
+  ///
+  /// This is like `value.asStream()`, but if a subscription to the stream is
+  /// canceled, this is as well.
+  Stream<T> asStream() {

[Lasse Reichstein Nielsen, 2015/08/27 10:20:54]

I prefer to make this a getter.

[nweiz, 2015/08/28 00:34:56]

I think it's more important for it to be consistent with [Future.asStream].
Also, a getter makes it seem like the Stream is the same "tier" of exposed value
as the Future as opposed to being derived, which is confusing in a couple ways:
it makes it less clear that this class is "cancel behavior like
Stream/StreamController but for Futures", and it makes it less clear that
calling [asStream] will stop any errors in [value] from being top-leveled.

[Lasse Reichstein Nielsen, 2015/09/03 13:06:17]

Ok, that's fair.

+    var controller = new StreamController<T>(
+        sync: true, onCancel: _completer._cancel);
+
+    value.then((value) {
+      controller.add(value);
+      controller.close();
+    }, onError: (error, stackTrace) {
+      controller.addError(error, stackTrace);
+      controller.close();
+    });
+    return controller.stream;
+  }
+
+  /// Cancels this operation.
+  ///
+  /// This returns the [Future] returned by the [CancelableCompleter]'s
+  /// `onCancel` callback. Unlike [Stream.cancel], it never returns `null`.
+  Future cancel() {
+    _canceled = true;

[Lasse Reichstein Nielsen, 2015/09/03 13:06:17]

Maybe drop the _canceled field in this class and just read
_completer._isCanceled directly.

[nweiz, 2015/09/08 21:14:45]

We don't actually use this field anyway, it turns out.

+    return _completer._cancel();
+  }
+}
+
+/// A completer for a [CancelableOperation].
+class CancelableCompleter<T> {

[Lasse Reichstein Nielsen, 2015/08/27 10:20:54]

I don't think you need this class.
Do you *want* it?

[nweiz, 2015/08/28 00:34:56]

Yeah; see above comments.

+  /// The completer for the wrapped future.
+  final Completer<T> _inner;
+
+  /// The callback to call if the future is canceled.
+  final ZoneCallback _onCancel;
+
+  CancelableOperation<T> get operation => _operation;
+  CancelableOperation<T> _operation;

[Lasse Reichstein Nielsen, 2015/08/27 10:20:53]

I find it very hard to see which fields the class have when it's interleaved
like this. I'd prefer to move the getters down below the constructor.

[nweiz, 2015/08/28 00:34:56]

Done, although see above.

+
+  /// Whether the completer has completed.
+  bool get isCompleted => _isCompleted;
+  bool _isCompleted = false;
+
+  /// Whether the completer was canceled before being completed.
+  bool get isCanceled => _isCanceled;
+  bool _isCanceled = false;
+
+  /// Whether the completer has fired.
+  ///
+  /// This is distinct from [isCompleted] when a [Future] is passed to
+  /// [complete]; this won't be `true` until that [Future] fires.
+  bool _fired = false;

[Lasse Reichstein Nielsen, 2015/09/03 13:06:17]

I think _fired is equivalent to _inner.isCompleted, so you don't need an extra
field for it.

[nweiz, 2015/09/08 21:14:45]

Done.

+
+  /// Creates a new completer for a [CancelableOperation].
+  ///
+  /// When the future operation canceled, as long as the completer hasn't yet
+  /// completed, [onCancel] is called. The callback may return a [Future]; if
+  /// so, that [Future] is returned by [CancelableOperation.cancel].
+  CancelableCompleter({onCancel()})
+      : _onCancel = onCancel,
+        _inner = new Completer<T>() {
+    _operation = new CancelableOperation<T>._(this);
+  }
+
+  /// Completes [operation] to [value].
+  ///
+  /// If [value] is a [Future], this will complete to the result of that
+  /// [Future] once it completes.
+  void complete([value]) {
+    if (_isCompleted) throw new StateError("Operation already completed");
+    _isCompleted = true;
+
+    if (_isCanceled) return;

[Lasse Reichstein Nielsen, 2015/08/27 10:20:54]

If value is a future with an error, the error is not handled.
It should probably be ignored since we have said we don't care about the
operation's result.

Maybe this line just shouldn't be here, since you check _isCanceled again below?

[nweiz, 2015/08/28 00:34:56]

Done.
I don't want to complete the inner completer for non-Future values or keep
around closures with references to this completer for longer than necessary, but
done.

+    if (value is! Future) {
+      _fired = true;
+      _inner.complete(value);
+      return;
+    }
+
+    value.then((result) {
+      if (_isCanceled) return;
+      _fired = true;
+      _inner.complete(result);
+    }, onError: (error, stackTrace) {
+      if (_isCanceled) return;
+      _fired = true;
+      _inner.completeError(error, stackTrace);
+    });
+  }
+
+  /// Completes [operation] to [error].
+  void completeError(Object error, [StackTrace stackTrace]) {
+    if (_isCompleted) throw new StateError("Operation already completed");
+    _isCompleted = true;
+
+    if (_isCanceled) return;
+    _fired = true;
+    _inner.completeError(error, stackTrace);
+  }
+
+  /// Cancel the completer.
+  Future _cancel() => _cancelMemo.runOnce(() {
+    if (_fired) return null;
+    _isCanceled = true;
+    if (_onCancel != null) return _onCancel();
+  });
+  final _cancelMemo = new AsyncMemoizer();

[Lasse Reichstein Nielsen, 2015/08/27 10:20:54]

Move field to top of class.

[nweiz, 2015/08/28 00:34:56]

I will if you insist, but I think it's a lot clearer to have memoization fields
right next to the corresponding methods/getters.

[Lasse Reichstein Nielsen, 2015/09/03 13:06:17]

I still prefer to have all the fields collected in one place. That makes it
easier to see what the state of the object is. 

We should have static variables in instance functions, like in C. Then it's
obvious that it's an instance field that's only visible to the current function.

[nweiz, 2015/09/08 21:14:45]

Done.

+}