Or&lt;E&gt; newValue;
      try {
        newValue = convert(event);
      } catch (e, s) {
        controller.addError(e, s);
        return;
      }
      if (newValue is Future&lt;E&gt;) {
        subscription.pause();
        newValue
            .then(add, onError: addError)
            .whenComplete(subscription.resume);
      } else {
        controller.add(newValue);
      }
    }, onError: addError, onDone: controller.close);
  }

  if (this.isBroadcast) {
    controller = new StreamController&lt;E