title: Cancel stream onData
body: <p>I have an event bus that handles all central events to my app. I have a special case where I have a sequence of asynchronous actions that I want to execute once only (when a special event happens), so I have to launch async function one, go through out of control other functions that will trigger an event for my second action, and so on. </p>
  
  <p>So I need to launch action one, then listen to the event bus waiting for action one to trigger (undirectly) an event that will launch action two etc...</p>
  
  <p>Naturally, once each element of the sequence is executed, I want to stop listening to the event that triggered it.</p>
  
  <p>I imagined for that, a consumeOnce(event, action) function that will subscribe to the bus, wait for the expected event, execute the action when receiving the event and immediately cancel the subscription once the action launched (asynchronously) </p>
  
  <pre><code>  final StreamController&lt;Map&lt;PlaceParam, dynamic&gt;&gt; _controller =
    new StreamController&lt;Map&lt;PlaceParam, dynamic&gt;&gt;.broadcast();
  
    void consumeOnce(PlaceParam param, Function executeOnce) {
      StreamSubscription subscription = _controller.stream.listen((Map&lt;PlaceParam, dynamic&gt; params) {
        if(params.containsKey(param)) {
          executeOnce();
          subscription.cancel(); //can't access, too early: not created yet
        }
      });
    }
  </code></pre>
  
  <p>The issue is that I can't access the variable <strong>subscription</strong> in the body of my callback, since it is still not created at the time</p>
  
  <p>Since nothing garanties that listeners will execute in their subscription order, I cannot register another subscriber who will remove my subscription (and even if the execution order was guaranteed, I will anyway find my self with a subscription that I can't remove : the one responsible for removing my original subscription)...</p>
  
  <p>Any ideas please?</p>
  
  <p>This pattern could solve my problem, but I don't find it elegant:</p>
  
  <pre><code>@Injectable()
  class EventBus&lt;K, V&gt; {
    final StreamController&lt;Map&lt;PlaceParam, dynamic&gt;&gt; _controller =
    new StreamController&lt;Map&lt;PlaceParam, dynamic&gt;&gt;.broadcast();
  
    Future&lt;Null&gt; fire(Map&lt;PlaceParam, dynamic&gt; params) async {
      await _controller.add(params);
    }
  
    Stream&lt;Map&lt;PlaceParam, dynamic&gt;&gt; getBus() {
      return _controller.stream;
    }
  
  
    void consumeOnce(PlaceParam param, Function executeOnce) {
      SubscriptionRemover remover = new SubscriptionRemover(param, executeOnce);
      StreamSubscription subscription = _controller.stream.listen(remover.executeOnce);
      remover.subscription = subscription;
    }
  }
  
  class SubscriptionRemover {
    PlaceParam param;
    Function executeOnce;
    StreamSubscription subscription;
  
    SubscriptionRemover(this.param, this.executeOnce);
  
    void execute(Map&lt;PlaceParam, dynamic&gt; params) {
      if (params.containsKey(param)) {
        executeOnce();
        subscription.cancel();
      }
    }
  }
  </code></pre>
  
  <p>But I don't like it much since theoretically, the event could happen between the two calls :</p>
  
  <pre><code>    StreamSubscription subscription = _controller.stream.listen(remover.executeOnce); //event may occur now!!!
      remover.subscription = subscription;
  </code></pre>
  
  <p>I think the existence of a method: _controller.stream.remove(Function fn) would have been a lot more direct and clear.</p>
  
  <p>Am I right? or is there a way I didn't think of?</p>