.do(onSubscribed:) and .do(onSubscribe:) don't respect .observe(on:)

[c128128]

Short description of the issue:

1 Case

Looks like .do(onSubscribed:) and .do(onSubscribe:) don't respect .observe(on:).

example:

  Observable<String>.just("1")
      .observe(on: SerialDispatchQueueScheduler(internalSerialQueueName: "observe"))
      .do(onSubscribed: {
          print("do, isMain ==> \(Thread.isMainThread), DispatchQueue ==> \(DispatchQueue.name())")
      })
      .subscribe(onNext: { _ in
          print("subscribe, isMain ==> \(Thread.isMainThread), DispatchQueue ==> \(DispatchQueue.name())")
      })

result:

do, isMain ==> true, DispatchQueue ==> main-thread
subscribe, isMain ==> false, DispatchQueue ==> observe

expected:

do, isMain ==> false, DispatchQueue ==> observe
subscribe, isMain ==> false, DispatchQueue ==> observe

2 Case

Also might be related:

Looks like .do(onSubscribed:) and .do(onSubscribe:) don't respect .subscribe(on:) if it is above do statement.

example:

Observable<String>.just("1")
    .subscribe(on: SerialDispatchQueueScheduler(internalSerialQueueName: "subscribe"))
    .do(onSubscribed: {
        print("do, isMain ==> \(Thread.isMainThread), DispatchQueue ==> \(DispatchQueue.name())")
    })
    .subscribe(onNext: { _ in
        print("subscribe, isMain ==> \(Thread.isMainThread), DispatchQueue ==> \(DispatchQueue.name())")
    })

result:

subscribe, isMain ==> false, DispatchQueue ==> subscribe
do, isMain ==> true, DispatchQueue ==> main-thread

expected:

subscribe, isMain ==> false, DispatchQueue ==> subscribe
do, isMain ==> false, DispatchQueue ==> subscribe

[danielt1263]

In both cases you are subscribing on the main thread. It seems to me that the result you are seeing is correct. Note that if you put the do(onSubscribed:) before the subscribe(on:) then you will see isMain equal to false. Because everything in the stream above the subscribe(on:) is subscribed on the Scheduler provided.

[c128128]

I was expecting all operators to behave the same. For example, if I use .map() instead of .do(), it would behave as expected (do, isMain ==> false, DispatchQueue ==> observe).

[danielt1263]

The do(onNext:) (which is the closest to the map) does receive on the same thread as the subscribe(onNext:)... Subscription is not the same as event emission so they won't necessarily happen on the same thread.

[c128128]

Got it. Then https://github.com/ReactiveX/RxSwift/blob/main/Documentation/Schedulers.md is wrong.

From Schedulers.md:

If you want to perform work on a different scheduler just use observeOn(scheduler) operator.

[danielt1263]

What are you talking about? The example is exactly correct. It seems like you are confused as to the difference between subscription and event propagation. Those are different processes.

Use do(onNext:) instead of do(onSubscribed:)...