am;
      try {
        newStream = convert(event);
      } catch (e, s) {
        controller.addError(e, s);
        return;
      }
      if (newStream != null) {
        subscription.pause();
        controller.addStream(newStream).whenComplete(subscription.resume);
      }
    },
        onError: eventSink._addError, &#47;&#47; Avoid Zone error replacement.
        onDone: controller.close);
  }

  if (this.isBroadcast) {
    controller = new StreamController&lt;E&gt;.broadcast(
        onListen: onList