title: Composing a command bus using RX
body: <p>I'm getting familiar with RX and as my experimental project I'm trying to create a simple command bus conceptually similar to this:</p>
  
  <pre><code>class Bus
  {
      Subject&lt;Command&gt; commands;
      IObservable&lt;Invocation&gt; invocations;
  
      public Bus()
      {
          this.commands = new Subject&lt;Command&gt;();
          this.invocations = commands.Select(x =&gt; new Invocation { Command = x }).Publish();
      }
  
      public IObserver&lt;Command&gt; Commands
      {
          get { return this.commands; }
      }
  
      public IObservable&lt;Invocation&gt; Invocations
      {
          get { return this.invocations; }
      }
  }
  
  class Invocation
  {
      public Command Command { get; set; }
      public bool Handled { get; set; }
  }
  </code></pre>
  
  <p>The idea is that modules can install command handlers at startup using the <code>Invocations</code> property and can apply any filtering they wish to their subscription. Clients on the other hand can trigger command execution by calling <code>Commands.OnNext(command)</code>.</p>
  
  <p>However, I would like the Bus to provide a guarantee that each command submitted will be handled by exactly one handler. That is, <code>OnNext</code> processing should ideally terminate as soon as the first handler sets Invocation.Handled to true AND should throw an exception if, at the conclusion of <code>OnNext()</code>, <code>Invocation.Handled</code> is still false.</p>
  
  <p>I played around creating my own ISubject, IObservable and IObserver implementations but this feels "dirty and cheap" ;)</p>
  
  <p>I'm endeavouring to get my head around the compositional power that RX provides. In a  compositional way, how can I provide the "exactly once" guarantee? </p>
  
  <p>Thanks for any insights you can provide.</p>