title: Processing a batch of request with Reactive Extension
body: <p>I'm learning Reactive Extensions, and I've been trying to find out if it's a match for a task like this. </p>
  
  <p>I have a Process() method that processes a batch of requests as a unit of work, and invoking a callback when all requests have completed. </p>
  
  <p>The important thing here is that each request will call the callback either synchronous or asynchronous depending on it's implementation, and the batch processor must be able to handle both.</p>
  
  <p>But no threads are started from the batch processor, any new threads (or other async execution) will be initiated from inside the request handlers if necessary. I don't know if this match the use cases of rx.</p>
  
  <p>My current working code looks (almost) like this:</p>
  
  <pre><code>public void Process(ICollection&lt;IRequest&gt; requests, Action&lt;List&lt;IResponse&gt;&gt; onCompleted)
  {
      IUnitOfWork uow = null;
      try
      {
          uow = unitOfWorkFactory.Create();
  
          var responses = new List&lt;IResponse&gt;();
          var outstandingRequests = requests.Count;
          foreach (var request in requests)
          {
              var correlationId = request.CorrelationId;
              Action&lt;IResponse&gt; requestCallback = response =&gt;
              {
                  response.CorrelationId = correlationId;
                  responses.Add(response);
                  outstandingRequests--;
                  if (outstandingRequests != 0)
                      return;
  
                  uow.Commit();
                  onCompleted(responses);
              };
  
              requestProcessor.Process(request, requestCallback);
          }
      }
      catch(Exception)
      {
          if (uow != null) 
              uow.Rollback();
      }
  
      if (uow != null) 
          uow.Commit();
  }        
  </code></pre>
  
  <p>How would you implement this using rx? Is it reasonable?</p>
  
  <p>Note, that the unit of work is to be committed synchronously even if there are async requests that have not yet returned.</p>