anchor.diff:                .then(add, onError: addError)
                 .whenComplete(subscription.resume);
           } else {
  -          if (newValue is! E) throw "unreachable"; // TODO(lrn): Remove when type promotion works.
  +          if (newValue is! E) {
  +            throw "unreachable"; // TODO(lrn): Remove when type promotion works.
  +          }
             controller.add(newValue);
           }
         });
  
positive.code:    * and the subscriptions' timers can be paused individually.
     */
    Stream<T> timeout(Duration timeLimit, {void onTimeout(EventSink<T> sink)?}) {
      _StreamControllerBase<T> controller = isBroadcast
          ? new _SyncBroadcastStreamController<T>(null, null)
          : new _SyncStreamController<T>(null, null, null, null);
  
      Zone zone = Zone.current;
      // Register callback immediately.
  
negative.code:    * The returned stream is a broadcast stream if this stream is.
     */
    Stream<E> asyncExpand<E>(Stream<E> convert(T event)) {
      _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,
  