Reactive Cocoa - How to wait for multiple signals to complete

Recently, the Functional Reactive Programming (FPR) paradigm has been gaining momentum, but getting started can be hard as its necessary to change the way you’re thinking about the datafow in your program.

For some time now I’ve been answering questions on StackOverflow regarding ReactiveCocoa.

From time to time there are reasonable and seemingly simple questions (with code samples) where there’s a simple answer to the questions and then theres a more elaborated answer of the kind “If you restructure your code completely it could be much simpler”.

I want to take a look at some of these questions in more detail.

How to wait for multiple signals to complete

The question was posted in ObjC using ReactiveObjC. I’ll adapt the code here in Swift using ReactiveSwift and SignalProducer for RACSignal as the signals in the question have cold semantics (See here if you are not familiar with hot vs. cold semantic).

The literal answer

At first sight, this questions has a very simple answer. Given you have some producers, use the merge operator and the resulting producer will complete will only complete when all merged producers have completed.

  let producerA: SignalProducer<Any, NoError>
  let producerB: SignalProducer<Any, NoError>

  let both = SignalProducer.merge([producerA, producerB])

Here, both producers will be started immediately and if their values are produced after some delay, the values might be interleaved. If it is necessary for the to be started in order, use concat instead of merge. (In ReactiveSwift, you’ll have to do let both = SignalProducer([producerA, producerB]).flatten(.concat)).

The actual problem

Looking closer at the code sample in the question, you’ll find that the inner producers are not independent from each other. There is an implicit dependency via the contactModelMutableArray array:

var contactModel = []
let selectMessages = SignalProducer {
  Database.loadMessages { modelArray in
    // Populate `contactModel` with contacts from database
    for(model in modelArray) {
      if(model.uid > 0) {
        contactModel.append(model)
      }
    }
  }
}
let selectInfo = SignalProducer {
  // Iterate through `contactModel` and update elements
  for model in contactModel {
    // Request information for model and update the model
    // Note: self.loadInformation returns a Producer
    self.loadInformation(for: model).start { data in
      model.data = data
    }
  }
}
let replace = SignalProducer {
  // Iterate through `contactModel` and update/replace database row for element
  for model in contactModel {
    Database.update(model)
  }
}

This approach is actually not that surprising if the asking person is not yet completely comfortable with FPR and is still mixing in imperative patterns. Some other issues are:

• Database.loadMessages already returned a SignalProducer, it not necessary to wrap this in another SignalProducer
• Checking the uid can be done with filter
• self.loadInformation might be asynchronous, in that case there’s no guarantee that the model is already updated by the time it is written back to the database

Data flow

The most important issue is the implicit data flow via the shared contactModel array. On a high level, the data flow is pretty straight forward

1. Load all models (+ potential filtering)
2. Update each model
3. Write each model back to the database

So lets implement this dataflow explicitely, with each step handing its data on to the next step.

For the first step, we’ll update selectMessages:

let selectMessages = Database.loadMessages
  .flatMap { SignalProducer(values: $0) }
  .filter { $0.uid > 0 }

Database.loadMessages sends an array of models as value (SignalProducer<[Model]>). Because we want to process each single model in the following steps, it would be easier if we had a producer that sent each of the models as a separate event (SignalProducer<Model>). We can achieve this by using flatMap and the convenience initializer SignalProducer.init(values:) which takes a Sequence and sends each element of the sequence as value and the completes. We need to use flatMap, because with only map we would get a SignalProducer<SignalProducer<Model>> and the flatMap basically “flattens” that to SignalProducer<Model>. Now, we can use filter to check the uids.

For the second step, we want to process each Model that is sent and load information for that model:

let selectInfo = selectMessages
  .flatMap { self.loadInformation(for: $0) }

Since we’re sending the models as values on the signal, this is now very simple. The only thing to keep in mind is that we have to use flatMap here again to “flatten” the stream just like before.

The last step is basically the same as the second one:

let replace = selectInfo
  .flatMap { Database.update($0) }

Take all these together and the solution looks like this:

let selectMessages = Database.loadMessages
  .flatMap { SignalProducer(values: $0) }
  .filter { $0.uid > 0 }
  .flatMap { self.loadInformation(for: $0) }
  .flatMap { Database.update($0) }

Now that the data “flows” through these operators and the dependencies between them is explicit, there suddenly are not 3 (seemingly) independent signals anymore but only one. And the initial Question “how to wait until all signals complete” does not even arise anymore. Oh, and furthermore, this is a lot less code than before.

Playground

I have prepared a Playground on github with a working sample of the proposed solution.