Array of observables where the output of each is the input of the next

Question

I'm trying to use RxSwift to execute actions on multiple data sources. However, I have no idea how to accomplish the following.

I have an array of observabless where the output of each, should be the input of the next. So, I want to do something like, get the first observable, wait for the result and pass it to the next, all the way to the end of the array and return one final value.

Is that possible? Thanks in advance.

*** Update: Ok, I'll be more specific as requested.

The 'observables' I'm using in the array, are custom. I use a function that returns Observable.create { ... }. Inside the closure, I run an asynchronous operation that transforms the value and then send the result to the observer before completing. That resulting value, must pass to the next observable, and so on to the last observable in the array to get a final value.

The observables may send multiple values, but they must pass from one observable to the next like an assembly line.

Can you be more specific? What do you mean by "wait for the result and pass it to the next"? Observables produce 0..n "results" by definition, so is it every result or one specific result? That said, what is "one final value"? — Maxim Volgin
It would be so nice if you had code to post in your question. Right now your description doesn't make sense - if you have an array of observables you cannot pass the output of one to the "input" of the next - observables don't have "inputs". Observers do, but they don't have outputs. Subjects have both, but that's not what you've described. — Enigmativity
"resulting value, must pass to the next observable" - ok, this is called "Continuation Passing Style (CPS)" and describes control flow in functional programming in general, and in Rx in particular. Transforming values emitted by one observable into other observables is achieved by .flatMap() operator. There are plenty of other operators though, it is better to learn them upfront to know your choices - see overview at rxmarbles.com — Maxim Volgin

Jon Willis Jon Willis · Accepted Answer · 2017-11-01T16:36:02

It is difficult to know exactly what you are asking for, since Observables do not exactly have inputs but I think this is a common problem.

You may be looking for a combination of the concat or reduce operators, which allow you to accumulate data from the values emitted from an Observable. See ReactiveX's documentation for Mathematical and Aggregate Operators.

Hopefully this can get you started:

// "I have an array of observables..."
let one = Observable.deferred { Observable.just(1) }
let two = Observable.deferred { Observable.just(2) }
let observables = [one, two]

// "the output of each, should be the input of the next"
// this is problematic, because observables do not strictly have inputs.
let resultsFromEach = Observable.concat(observables)

resultsFromEach
    .reduce(0) { result, next in
        result + 1
    }
    .debug("result")
    .subscribe()

Array of observables where the output of each is the input of the next

1 Answers