Unboxing Packages: async Part 1

Writing asynchronous code is hard. Even with Dart’s lovely async/await syntax, asynchrony inherently involves code running in a nondeterministic order, intermingled in a way that’s often difficult to understand and debug. Because there are so many possible ways to execute a program, and which one happens is so contingent on little details of timing, it’s not even possible to test asynchronous code in the same exhaustive way you might test something synchronous.

This is why it’s so important to have abstractions and utilities that are simple and robust and can be used as building blocks for more complex programs. The core libraries provide the most fundamental abstractions: Future for an asynchronous value, and Stream for an asynchronous collection or signal. But it’s up to packages to build the next layer on top of those fundamental primitives.

The async package is where the next-most-basic abstractions live. It’s a core library expansion pack that contains APIs that are just a little more advanced than would fit in dart:async itself. The classes it exposes are well-tested and as straightforward as you can hope from asynchrony, and using them properly can make your code vastly clearer and more reliable.

The async package is just chock-full of cool stuff. So full, in fact, that it won’t all fit in a single blog post—I have to split it up. Today I’ll mostly talk about APIs that deal with individual values. I’ll save Streams for the next post, and I may even need a third to cover the whole package.

AsyncMemoizer

If you’re writing synchronous code and you want to provide access to a value in a class, you just define it as a field. If that value should only be computed when it’s accessed, you can memoize it by writing a getter that sets a field the first time it’s accessed.

String get contents {
  if (_contents == null) _contents = new File(this.path).readAsStringSync();
  return _contents;
}
String _contents;

But what if the value can only be computed asynchronously? You still want to compute it lazily, but you don’t want to have every access recompute it from scratch. Once the computation has started, future accesses should return the same future. You can implement this manually, but it’s a pain; AsyncMemoizer makes it easy:

Future<String> get contents => _contentsMemo.runOnce(() {
  return new File(this.path).readAsString();
});
final _contentsMemo = new AsyncMemoizer<String>();

The first time you call runOnce() on a given memoizer, it invokes the callback and returns its return value as a Future. After that, all calls to runOnce() don’t use the callback at all; they just return the same value. Make sure you always pass the same callback—otherwise you may not be able to tell which one will run!

Some readers may be wondering why the callback isn’t passed to the constructor. After all, it’s only invoked once—future calls to runOnce() just throw it away! The answer is that we want code using AsyncMemoizer to be able to look like the snippet above, with the code the getter executes right there in the body of the getter. We also want the memoizer itself to be usable as a final variable, which means that its constructor couldn’t refer to other fields in the class (like this.path).

Another common use of AsyncMemoizer doesn’t even use the return value of the callback. It just uses the memoizer to ensure that a method’s body is only executed once, and that it always returns the same future. Which, it turns out, is exactly how close() methods work for a lot of classes.

bool get isClosed => _closeMemo.hasRun;

Future close() => _closeMemo.runOnce(() async {
  await _engine.close();
  await _loader.close();
});
final _closeMemo = new AsyncMemoizer();

Notice the isClosed getter in that example. AsyncMemoizer exposes a hasRun property specifically to make that sort of getter possible: hasRun returns true if runOnce() has been called, regardless of whether the callback has completed. It also has a future property, which returns the same future as runOnce() without actually running the memoizer.

Result

When working asynchronously, values and errors are often treated as two sides of the same coin. A future completes with either a value or an error, and a stream emits value events and error events. But in the synchronous world, errors are completely different than values, and that can cause friction when moving between synchronous and asynchronous code.

That’s what the Result class is for. Each Result is either a value or an error, and whichever it is is accessible synchronously. It has two subclasses, one for each state: ValueResult has a value getter, and ErrorResult has a error and stackTrace getters. If you have a Result, you can use isValue and isError to easily check its type, followed by asValue and asError to easily cast it to the proper type.

You can create Results manually using new Result.value() or new Result.error(), but there are utility functions to convert from asynchronous objects: Result.capture() turns a Future into a Future<Result>, and Result.captureStream() turns a Stream into a Stream<Result>. Errors that would have been emitted using the normal future or stream error channels are turned into ErrorResults instead.

You can also reverse this process using Result.release() and Result.releaseStream(). These take a Future<Result> and a Stream<Result>, respectively, and convert ErrorResults to normal error events.

Result has some instance methods for moving back to the async world too. The asFuture getter returns a future that completes to the Result’s value or error, and complete() completes a Completer so that its future does the same. For streams, you can use addTo() to add the value or the error to an EventSink.

ResultFuture

Sometimes you want limited synchronous access to a future. Maybe you want to do something with its value if it exists, but not wait for it if it doesn’t. The ResultFuture class makes this possible by exposing a result getter. Before the future has completed, this is just null; afterwards, it’s a Result. Otherwise, the ResultFuture is just a normal future that works like futures work.

// A Shelf handler that forwards requests to a `Future<Handler>`.
class AsyncHandler {
  final ResultFuture<shelf.Handler> _future;

  AsyncHandler(Future<shelf.Handler> future) : _future = new ResultFuture(future);

  call(shelf.Request request) {
    if (_future.result == null) {
      return _future.then((handler) => handler(request));
    }

    // Because [_future]'s a [Future], we can return it to throw error.
    if (_future.result.isError) return _future;

    return _future.result.asValue.value(request);
  }
}

CancelableOperation

One cool feature of streams in Dart is that their subscriptions can be cancelled. In addition to stopping any more events callbacks from firing, this indicates to the stream producer that it can stop generating events at all. Unfortunately, there’s no similar facility for futures, which is where CancelableOperation comes in.

A CancelableOperation represents an asynchronous operation that will ultimately produce a single value which is exposed as a future (and which may complete to null). It can also be canceled, which causes the value future never to complete and lets the code that created it know to stop work on the operation.

Normally when a CancelableOperation is canceled, it just doesn’t complete—which is analogous to a stream subscription not emitting any events once it’s canceled. But sometimes, especially when using async/await, this isn’t what you want. In that case, you can call valueOrCancellation(), which returns a future that completes even if the operation was canceled. By default it completes to null, but you can pass in a custom value if you want.

There are two ways to create a CancelableOperation. If you already have a value future and you just want to wrap it, you can call new CancelableOperation.fromFuture(). This also takes an onCancel callback that is called if the operation is canceled.

CancelableOperation runSuite(String path) {
  var suite;
  var canceled = false;
  return new CancelableOperation(() async {
    suite = await loadSuite(path);
    if (canceled) return null;

    return suite.run();
  }(), onCancel: () {
    canceled = true;
    return suite?.close();
  });
}

If the onCancel callback returns a Future, it will be forwarded to the return value for the call to cancel(). This is just like how cancelling a StreamSubscription works, except that for consistency CancelableOperation.cancel() never returns null.

You can also create a CancelableOperation using a CancelableCompleter. This works a lot like a Completer for a future: it has complete() and completeError() methods, and it exposes the operation it controls through the operation getter. But its constructor takes an onCancel callback that’s called if the operation is canceled, and it has an isCanceled getter.

/// Like [Stream.first], but cancelable.
CancelableOperation cancelableFirst(Stream stream) {
  var subscription;
  var completer = new CancelableCompleter(
    onCancel: () => subscription.cancel());

  subscription = stream.listen((value) {
    completer.complete(value);
    subscription.cancel();
  }, onError: (error, stackTrace) {
    completer.completeError(error, stackTrace);
    subscription.cancel();
  });

  return completer.operation;
}

FutureGroup

Fun fact: at least three different versions of the FutureGroup class existed across the Dart world before a canonical implementation finally ended up in the async package. That’s a pretty good indication that it’s a broadly-applicable abstraction!

A FutureGroup collects a bunch of input futures and exposes a single output future that completes when all the inputs have completed. Inputs are added using add(), and the output is called future.

Once you’ve added all the futures you want to the group, call close() to tell it that no more are coming. Some astute students of the core libraries will recognize the pattern of add() followed by close() as the hallmark of a sink—and indeed, FutureGroup implements Sink.

// An engine for running tests.
class Engine {
  final _group = new FutureGroup();

  // Completes when all tests are done.
  Future get onDone => _group.future;

  void addTest(Test test) {
    _group.add(test.run());
  }

  void noMoreTests() {
    _group.close();
  }
}

If all the futures that have been added to a FutureGroup have completed but it hasn’t been closed, we say that the group is idle. You can tell whether a group is idle using the isIdle getter. You can also use the onIdle stream to get an event whenever the group becomes idle—that is, whenever the last running future completes.

!isCompleted

I hope I’ve whet your appetite for asynchrony, because there’s plenty more to come. Working with individual values is useful, but the bulk of the package—and, in my opinion, some of the coolest stuff it contains—has to do with streams. Check back in two weeks, when I tell you all about it!