Drain applies mappers without scheduling.

[AlecKazakova]

Here is the test case I've created

val testThread = Thread.currentThread();

rx.Observable.fromCallable { 10 }
        .toV2()
        .switchMap { original -> BehaviorSubject.createDefault(2)
            .observeOn(Schedulers.io())
            .map { multiplier ->
              if (Thread.currentThread() == testThread) throw AssertionError()
              original * multiplier
            }
        }
        .subscribe { throw IllegalStateException("nice") }

Thread.sleep(1000)

This hits the assertion error as SwitchMapObserver polls the inner query inside its drain method, running the poll method on MapObserver which ends up running on the thread the switchmap was subscribed to on (i think?).

I dont think the interop part is necessary to reproduce this bug, but was the consistent way I was able to get the drain method to poll the inner query as fromCallable emits a completion after the relay has emitted but before the mapper has received the event.

[akarnokd]

Thanks for reporting. This is a bug in the fusion code of switchMap: it should report itself as a boundary just like flatMap to prevent this type of execution. I'll post a fix shortly. Until then you can apply hide() after map to prevent the fusion.