RxJava observeOn / subscribeOn works only in test environment

Question

I have this AsyncSubject

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

If I do something like this in test environment (androidTest folder)

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

it enters only during testing, but I ran the same code in an activity it doesn't enter.

The only way to made it work both test and normal environment is to remove:

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

What I am doing wrong?

Those are my dependencies:

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

EDIT:

it appears that it doesn't work only if I try to change an outside variable. An example:

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

score 0 · Answer 1 · edited Jul 07 '19 at 11:19

0

When you run java unit tests, you can't execute AndroidSchedulers.mainThread()

What you can do is create a test rule to be able to execute the unit tests in a synchronous way.

In this answer you will find a detailed explanation.

You can take rule from here.

edited Jul 07 '19 at 11:19

Asad Ali Choudhry

4,985
4
31
36

answered Jul 07 '19 at 11:04

jpgpuyo

99
7

the problem is that it works only in testing environment. If I normally run the app, it won't work. Take a look at the edit to see what I mean. – stefano capra Jul 07 '19 at 12:07
I think that you should observeOn in main thread and subscribeOn on newThread. If you try to modify UI out of main thread, you will have a crash or nothing will happen. – jpgpuyo Jul 07 '19 at 12:48
I edited the question and try to switch. Nothing happened. The problem seems that I cannot change a variable outside the scope of the subscriber – stefano capra Jul 07 '19 at 13:03
1

Hey. I was really curious about this and I tested in a sample activity and I found the reason: asyncSubject is observing value in main thread, but you are blocking main thread in signal.await(). If you execute asyncSubject.observeOn(Schedulers.newThread()).subscribe, then it will work. Anyway, I don't know what is your intention, but I'm sure that you can find a simple solution using rxjava operators. CountDownLatch probably is not needed. – jpgpuyo Jul 09 '19 at 20:56

score 0 · Answer 2 · answered Jul 07 '19 at 13:07

0

You should sbscribe to your asyncSubject prior to calling subscribe on observable. Also, Observable.just(..) doesn't honour schedulers, so your subscribeOn/observeOn wont have an effect here.

answered Jul 07 '19 at 13:07

PPartisan

8,173
4
29
48

But if I add observeOn and subscribeOn in actual code it doesn't work, conversely to test environment. – stefano capra Jul 07 '19 at 17:21
nothing. It's like the the `subscribe` method never gets called. Only if avoid to assign or call outside variable works(so for example print something inside onNext, onComplete. In this case it works) – stefano capra Jul 07 '19 at 18:10

RxJava observeOn / subscribeOn works only in test environment

2 Answers2