anchor.diff:     * The returned stream is a broadcast stream if this stream is.
      */
     Stream<E> asyncMap<E>(FutureOr<E> convert(T event)) {
  -    _StreamControllerBase<E> controller = isBroadcast
  -        ? _SyncBroadcastStreamController<E>(null, null)
  -        : _SyncStreamController<E>(null, null, null, null);
  +    _StreamControllerBase<E> controller;
  +    if (isBroadcast) {
  +      controller = _SyncBroadcastStreamController<E>(null, null);
  +    } else {
  +      controller = _SyncStreamController<E>(null, null, null, null);
  +    }
   
       controller.onListen = () {
         StreamSubscription<T> subscription = this.listen(null,
  
positive.code:               .then(add, onError: addError)
                .whenComplete(subscription.resume);
          } else {
            if (newValue is! E) throw "unreachable"; // TODO(lrn): Remove when type promotion works.
            controller.add(newValue);
          }
        });
  
negative.code:       FutureOr<Null> add(E value) {
          controller.add(value);
        }
  
        final addError = controller._addError;
        subscription.onData((T event) {
          FutureOr<E> newValue;
  