RxSwift: chaining several actions

Question

Lets imagine we have an array of AnObject instances and need to have following sequence of actions to execute:

• send objects to backend via separate calls
• after step 1 finishes store this array to DB in batch
• after step 2 finishes do additional processing for each item

and we'd want to receive the signal only after all those steps were executed (or there was an error). What is the correct way to achieve this via RxSwift and is it actually possible?

Please find my prototype functions below. Unfortunately I didn't come up with a valid code sample for chaining, so there's nothing to demo.

func makeAPIRequest(object: AnObject) -> Observable<Void> {
    ...
}

func storeData(data: [AnObject]) -> Observable<Void> {
    ...
}

func additionalProcessing(object: AnObject) -> Observable<Void> { 
    ...
} 

func submitData()
{
   let data: [AnObject] = ...;

   let apiOperations = data.map{ makeAPIRequest($0) };
   let storageOperation = storeData(data);
   let processingOperations = data.map{ additionalProcessing($0) };

   ... // some code to chain steps 1-3
   .subscribe { (event) -> Void in
       // should be called when operations from step 3 finished  
   }.addDisposableTo(disposeBag);
}

Answer 1

Let's assume that both makeAPIRequest and additionalProcessing return an Observable<SomeNotVoidType>, and storeData takes an array as its argument and returns an Observable<Array>. This way, you can do the following:

First, create an array of Observables representing sending individual objects to backend. Then use toObservable method, so the resulting signals can be transformed later on:

let apiOperations = data.map{ makeAPIRequest($0) }.toObservable()

then use merge operator which will make an Observable, that completes only when all of the API calls complete. You can also use toArray operator, which will put the API call results into one array:

let resultsArray = apiOperations.merge().toArray()

This will get you an Observable<Array<ApiResult>>, which will send one Next event when all API operations complete successfully. Now you can store the results in the database:

let storedResults = resultsArray.flatMap { storeDatabase($0) }

Then again you want to make Observables for each array element, so they'll represent additional processing. Note that you need to use flatMap and flatMapLates, otherwise you'll end up with nested observables like Observable<Observable<SomeType>>.

let additionalProcessingResults = storedResults.flatMap {
      return $0.map(additionalProcessing).toObservable()
    }.flatMapLatest { return $0 }

Then, you can subscribe for successful completion of the additional processing (or you can do something with its individual results):

additionalProcessingResults.subscribe { (event) -> Void in
       // should be called when operations from step 3 finished  
   }.addDisposableTo(disposeBag);

Note that you don't need all the intermediate variables, I just left them to describe all the steps.