Concatenating two observable sequences that both have subscribeOn. How do I ensure my observable runs on a thread?

Question

When it comes to enforcing that a certain piece of Observable.create code runs in a specific thread (i.e. background thread), i worry that using the subscribeOn operator might not work because there are times that I might chain this observable sequence to another observable sequence that runs on a main thread (using observeOn).

Example

The situation is that I have an Observable sequence running on the main thread (i.e. an alert box asking the user for input, as to whether perform the network call or not).

Would it be better to ensure that this Observable.create code runs in the background thread by having something like:

Observable<String>.empty()
   .observeOn(ConcurrentMainScheduler(queue: background.queue))
   .concat(myObservableNetworkCall)

Why not just use subscribeOn?

The problem is if I had used subscribeOn (second) and the previous observable (the alert controller) was set to run on the background thread using subscribeOn (first), then the second subscribeOn operator would not work since the first call is closer to the source observable:

If you specify multiple subscribeOn() operators, the one closes to the source (the left-most), will be the one used.

• Thomas Nield on RxJava's subscribeOn and observeOn operators (February 2016)

That may be the behavior for RxJava, but I am not sure for Swift. Reactivex.io simply says that we should not call subscribeOn multiple times.

I tend to wrap operations into Observable<Void>s and they need to be run on different threads... That is why I am asking for how to ensure an Observable code run in the thread I specified it to. subscribeOn wouldn't work because I can concatenate the observable.

I want the thread it should run in to be encapsulated in my Observable definition, not higher up in the chain.

Is the best practice to do the following:

1. Start with an Observable.empty using the data type I wish to use.
2. Use observeOn to force the thread that I want it to run in.
3. Concatenate it with the actual Observable that I want to use.

Edit

1. I have read the subscribeOn and observeOn documentation on reactivex.io.

2. I'm familiar with how to switch between threads using subscribeOn and observeOn.

3. What I'm specifically concerned about is the complication of using subscribeOn when concatenating or combining observable sequences.

4. The problem is, the observables need to run specifically on one thread, AND they don't know where and who they'll be concatenated with. Since I know exactly which thread they should run on, I'd prefer to encapsulate the scheduler definition within the observable's definition instead of when I'm chaining a sequence.

Answer 1

In the function declaration it is better not to specify on which thread the function is to be called. For instance:

func myObservableNetworkCall() -> Observable<String> {
    return Observable<String>.create { observer in
        // your network code here

        return Disposables.create {
            // Your dispose
        }
    }
}

func otherObservableNetworkCall(s: String) -> Observable<String> {
    return Observable<String>.create { observer in
        // your network code here

        return Disposables.create {
            // Your dispose
        }
    }
}

And then switch between Schedulers:

myObservableNetworkCall()
    .observeOn(ConcurrentMainScheduler(queue: background.queue)) // .background thread, network request, mapping, etc...
    .flatMap { string in
        otherObservableNetworkCall(s: string)
    }
    .observeOn(MainScheduler.instance) // switch to MainScheduler, UI updates
    .subscribe(onNext:{ string in
        // do something 
    })