="language-dart"><code class="language-dart">Stream&lt;E&gt; asyncMap&lt;E&gt;(FutureOr&lt;E&gt; convert(T event)) {
  StreamController&lt;E&gt; controller;
  StreamSubscription&lt;T&gt; subscription;

  void onListen() {
    final add = controller.add;
    assert(controller is _StreamController ||
        controller is _BroadcastStreamController);
    final _EventSink&lt;E&gt; eventSink = controller as Object;
    final addError = eventSink._addError;
    subscription = this.listen((T event) {
      Future