title: Reordering events with Reactive Extensions
body: <p>I'm trying to reorder events arriving unordered on different threads.</p>
  
  <p>Is it possible to create a reactive extension query that matches these marble diagrams:</p>
  
  <pre><code>s1          1   2       3   4
  
  s2          1   3   2       4
  
  result      1       2   3   4
  </code></pre>
  
  <p>and...</p>
  
  <pre><code>s1          1   2   3   4
  
  s2          4   3   2   1
  
  result                  1234
  </code></pre>
  
  <p>That is: Only publish results in version number order.</p>
  
  <p>The closest I have got is using a Join to open a window each time s1 ticks and only close it when s2 arrives with the same number. </p>
  
  <p>Like this:</p>
  
  <pre><code>var publishedEvents = events.Publish().RefCount();
  publishedEvents.Join(
          publishedEvents.Scan(0, (i, o) =&gt; i + 1),
          expectedVersion =&gt; publishedEvents.Any(@event =&gt; @event.Version == expectedVersion),
          _ =&gt; Observable.Never&lt;Unit&gt;(),
          (@event, expectedVersion) =&gt; new {@event,expectedVersion})
      .Where(x =&gt; x.expectedVersion == x.@event.Version)
      .Select(x =&gt; x.@event)
      .Subscribe(Persist);
  </code></pre>
  
  <p>But that won't work with diagram no 2. Group 2 will be completed once s2 ticks with the number 2, and thus before 1.</p>
  
  <p>Does it make sense? Can it be done with Rx? Should it?</p>
  
  <p><em>EDIT</em>: I guess it's like overlapping windows, where later windows can't close before all preceding windows have closed. And the preceding windows won't close before the window number matches the event version number.</p>
  
  <p><em>EDIT 2</em>: </p>
  
  <p>I have something like this now, but it's not really the reactive, functional, thread-safe LINQ-revelation, I hoped for (please ignore that my events are JObjects for now):</p>
  
  <pre><code>var orderedEvents = Observable.Create&lt;JObject&gt;(observer =&gt;
  {
      var nextVersionExpected = 1;
      var previousEvents = new List&lt;JObject&gt;();
      return events
          .ObserveOn(Scheduler.CurrentThread)
          .Subscribe(@event =&gt;
          {
              previousEvents.Add(@event);
  
              var version = (long) @event["Version"];
              if (version != nextVersionExpected) return;
  
              foreach (var previousEvent in previousEvents.OrderBy(x =&gt; (long) x["Version"]).ToList())
              {
                  if ((long) previousEvent["Version"] != nextVersionExpected)
                      break;
  
                  observer.OnNext(previousEvent);
                  previousEvents.Remove(previousEvent);
                  nextVersionExpected++;
              }
          });
  });
  </code></pre>