RxJava наблюдать за / подписаться работает только в тестовой среде

Question

У меня есть AsyncSubject

val asyncSubject = AsyncSubject.create<Int>()
observable.observeOn(AndroidSchedulers.mainThread())
          .subscribeOn(Schedulers.newThread())
          .subscribe(asyncSubject)

Если я сделаю что-то подобное в тестовой среде (папка androidTest)

asyncSubject.subscribe{
    // enter only in test environment
}

он входит только во время тестирования, но я выполнил тот же код в операции, в которую он не входит.

Единственный способ заставить его работать как в тестовой, так и в обычной среде - это удалить:

.observeOn(Schedulers.newThread())
.subscribeOn(AndroidSchedulers.mainThread())

Что я делаю не так?

Это мои зависимости:

implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
implementation 'io.reactivex.rxjava2:rxjava:2.2.8'

EDIT:

похоже, что это не сработает, только если я попытаюсь изменить внешнюю переменную. Пример:

val observable = Observable.just(1, 2, 3, 4, 5)
val asyncSubject = AsyncSubject.create<Int>()
observable.subscribeOn(Schedulers.newThread())
          .observeOn(AndroidSchedulers.mainThread())
          .subscribe(asyncSubject)
val signal = CountDownLatch(5)
var i = 0
asyncSubject.subscribe{
    i = it
    signal.countDown()
}
signal.await()
// After that `i` is always 0 

Answer 1

Вы должны sbscribe на asyncSubject до звонка подписаться на observable. Кроме того, Observable.just(..) не учитывает планировщики, поэтому ваш subscribeOn/observeOn не будет иметь здесь эффекта.

Answer 2

Когда вы запускаете модульные тесты Java, вы не можете выполнить AndroidSchedulers.mainThread()

Что вы можете сделать, это создать тестовое правило, чтобы иметь возможность выполнять модульные тесты синхронно.

В этом ответе вы найдете подробное объяснение.

Вы можете взять правило из здесь .