title: Make Reactive Extensions Buffer wait for asynchronous operation to complete
body: <p>I am using Reactive Extensions (Rx) to buffer some data. I'm having an issue though in that I then need to do something asynchronous with this data, yet I don't want the buffer to pass the next group through until the asynchronous operation is complete. </p>
  
  <p>I've tried to structure the code two ways (contrived example):</p>
  
  <pre><code>public async Task processFiles&lt;File&gt;(IEnumerable&lt;File&gt; files)
  {
      await files.ToObservable()
          .Buffer(10)
          .SelectMany(fi =&gt; fi.Select(f =&gt; upload(f)) //Now have an IObservable&lt;Task&gt;
          .Select(t =&gt; t.ToObservable())
          .Merge()
          .LastAsync();
  }
  
  public Task upload(File item)
  {
      return Task.Run(() =&gt; { //Stuff });
  }
  </code></pre>
  
  <p>or</p>
  
  <pre><code>public async Task processFiles&lt;File&gt;(IEnumerable&lt;File&gt; files)
  {
      var buffered = files.ToObservable()
          .Buffer(10);
  
      buffered.Subscribe(async files =&gt; await Task.WhenAll(files.Select(f =&gt; upload(f)));
  
      await buffered.LastAsync();
  }
  
  public Task upload(File item)
  {
      return Task.Run(() =&gt; { //Stuff });
  }
  </code></pre>
  
  <p>Unfortunately, neither of these methods have worked as the buffer pushes the next group before the async operations complete. The intent is to have each buffered group executed asynchronously and only when that operation is complete, continue with the next buffered group. </p>
  
  <p>Any help is greatly appreciated.</p>