title: A way to push buffered events in even intervals
body: <p>What I'm trying to achieve is to buffer incoming events from some IObservable ( they come in bursts) and release them further, but one by one, in even intervals.
  Like this:</p>
  
  <pre><code>-oo-ooo-oo------------------oooo-oo-o--------------&gt;
  
  -o--o--o--o--o--o--o--------o--o--o--o--o--o--o----&gt;
  </code></pre>
  
  <p>Since I'm quite new to <strong>Rx</strong>, I'm not sure if there already is a Subject or an operator that does just this. Maybe It can be done by composition?</p>
  
  <p><strong>update:</strong></p>
  
  <p>Thanks to
  <a href="https://stackoverflow.com/users/3603/richard-szalay">Richard Szalay</a> for pointing out the <strong>Drain</strong> operator, I found another <a href="http://social.msdn.microsoft.com/Forums/en-US/rx/thread/749771d5-544b-4619-adff-6e79b2906152" rel="nofollow noreferrer">example by James Miles</a> of Drain operator usage. Here's how I managed to get it to work in a WPF app:</p>
  
  <pre><code>    .Drain(x =&gt; {
          Process(x);
          return Observable.Return(new Unit())
              .Delay(TimeSpan.FromSeconds(1), Scheduler.Dispatcher );
      }).Subscribe();
  </code></pre>
  
  <p>I had some fun, because omitting the scheduler parameter causes the app to crash in debug mode without any exception showing up ( I need to learn how to deal with exceptions in Rx).
  The Process method modifies the UI state directly, but I guess it's quite simple to make an IObservable out of it (using a ISubject?).</p>
  
  <p><strong>update:</strong></p>
  
  <p>In the meantime I've been experimenting with ISubject, the class below does what I wanted - it lets out buffered Ts in a timely manner:</p>
  
  <pre><code>public class StepSubject&lt;T&gt; : ISubject&lt;T&gt;
  {
      IObserver&lt;T&gt; subscriber;
      Queue&lt;T&gt; queue = new Queue&lt;T&gt;();
      MutableDisposable cancel = new MutableDisposable();
      TimeSpan interval;
      IScheduler scheduler;
      bool idle = true;
  
      public StepSubject(TimeSpan interval, IScheduler scheduler)
      {
          this.interval = interval;
          this.scheduler = scheduler;
      }
  
      void Step()
      {
          T next;
          lock (queue)
          {
              idle = queue.Count == 0;
              if (!idle)
                  next = queue.Dequeue();
          }
  
          if (!idle)
          {
              cancel.Disposable = scheduler.Schedule(Step, interval);
              subscriber.OnNext(next);
          }
      }
  
      public void OnNext(T value)
      {
          lock (queue)
              queue.Enqueue(value);
  
          if (idle)
              cancel.Disposable = scheduler.Schedule(Step);
      }
  
      public IDisposable Subscribe(IObserver&lt;T&gt; observer)
      {
          subscriber = observer;
          return cancel;
      }
  }
  </code></pre>
  
  <p>This naive implementation is stripped from OnCompleted and OnError for clarity, also only single subscription allowed.</p>