BehaviorSubject subscribe on another thread

user_642680 0
Viewed 1074

I'm new to RxJava and I chose to use it because I thought it would be well suited to my use case.

I have some Integer values I want to observe over an infinite period of time. Whenever one of these values changes (i.e. an event) I want all of its observers to be called on another thread.

Because of the long observation time requirement I thought I needed to use the BehaviorSubject class (although initially I thought Observable was all that I needed .. seeing as I just need to 'observe'), and I could use the subscribeOn() method to set a scheduler and hence achieve calling the subscribers on a background thread:

private BehaviorSubject<Integer> rotationPositionSubject = BehaviorSubject.createDefault(getRotorPosition());
rotationPositionSubject.subscribeOn(scheduler);

And I have a rotate() method I used to update the rotationPositionSubject which will be called from the main thread:

@Override
public synchronized int rotate()
{
    final int newRotorPosition = super.rotate();
    rotationPositionSubject.onNext(newRotorPosition);

    return newRotorPosition;
}

However with the above code I found that the subscribers are called on the 'main' thread. Examining the docs for subscribeOn():

Returns:

the source ObservableSource modified so that its subscriptions happen on the specified Scheduler

So my above code won't work as I am not using the returned ObservableSource, but the return object is an Observable which is of no use for my application?

The question is then, how do I observe long-term any object and call subscribers on a background thread with RxJava, or is RxJava the wrong choice?

rx-java2 reactive-programming java
1 Answers

After some experimentation it looks as though one needs some care when using BehaviorSubject objects and their use isn't quite as obvious as I inferred from the names of the various interfaces.

As a test method to demonstrate what I am currently doing:

@Test
public void test()
{
    System.out.println("Executing test on thread ID: " + Thread.currentThread().getId());
    final BehaviorSubject<Integer> rotorBehaviour = BehaviorSubject.create();
    rotorBehaviour.subscribeOn(Schedulers.single());
    rotorBehaviour.subscribe(new Observer<Integer>()
    {
        @Override
        public void onSubscribe(final Disposable d)
        {
            System.out.println("onSubscribe() called on thread ID: " + Thread.currentThread().getId());
        }

        @Override
        public void onNext(final Integer integer)
        {
            System.out.println("onNext() called on thread ID: " + Thread.currentThread().getId());
        }

        @Override
        public void onError(final Throwable e)
        {
            System.out.println("onError() called on thread ID: " + Thread.currentThread().getId());
        }

        @Override
        public void onComplete()
        {
            System.out.println("onComplete() called on thread ID: " + Thread.currentThread().getId());
        }
    });


    rotorBehaviour.onNext(1);
    rotorBehaviour.onNext(2);

}

This results in the undesired result:

Executing test on thread ID: 1
onSubscribe() called on thread ID: 1
onNext() called on thread ID: 1
onNext() called on thread ID: 1

Process finished with exit code 0

(Undesired because onNext() is called on the main thread)

Modifying the code to use the Observable returned from the call to subscribeOn gives the same undesired result:

@Test
public void test()
{
    System.out.println("Executing test on thread ID: " + Thread.currentThread().getId());
    final BehaviorSubject<Integer> rotorBehaviour = BehaviorSubject.create();
    Observable<Integer> rotorObservable = rotorBehaviour.subscribeOn(Schedulers.single());

    rotorObservable.subscribe(new Observer<Integer>()
    {
        @Override
        public void onSubscribe(final Disposable d)
        {
            System.out.println("onSubscribe() called on thread ID: " + Thread.currentThread().getId());
        }

        @Override
        public void onNext(final Integer integer)
        {
            System.out.println("onNext() called on thread ID: " + Thread.currentThread().getId());
        }

        @Override
        public void onError(final Throwable e)
        {
            System.out.println("onError() called on thread ID: " + Thread.currentThread().getId());
        }

        @Override
        public void onComplete()
        {
            System.out.println("onComplete() called on thread ID: " + Thread.currentThread().getId());
        }
    });
    rotorBehaviour.onNext(1);
    rotorBehaviour.onNext(2);
}

Result:

Executing test on thread ID: 1
onSubscribe() called on thread ID: 1
onNext() called on thread ID: 1
onNext() called on thread ID: 1

Process finished with exit code 0

However using the observeOn() method does give the desired result:

@Test
public void test()
{
    System.out.println("Executing test on thread ID: " + Thread.currentThread().getId());
    final BehaviorSubject<Integer> rotorBehaviour = BehaviorSubject.create();
    Observable<Integer>rotorObservable = rotorBehaviour.observeOn(Schedulers.single());

    rotorObservable.subscribe(new Observer<Integer>()
    {
        @Override
        public void onSubscribe(final Disposable d)
        {
            System.out.println("onSubscribe() called on thread ID: " + Thread.currentThread().getId());
        }

        @Override
        public void onNext(final Integer integer)
        {
            System.out.println("onNext() called on thread ID: " + Thread.currentThread().getId());
        }

        @Override
        public void onError(final Throwable e)
        {
            System.out.println("onError() called on thread ID: " + Thread.currentThread().getId());
        }

        @Override
        public void onComplete()
        {
            System.out.println("onComplete() called on thread ID: " + Thread.currentThread().getId());
        }
    });
    rotorBehaviour.onNext(1);
    rotorBehaviour.onNext(2);
}

Executing test on thread ID: 1
onSubscribe() called on thread ID: 1
onNext() called on thread ID: 13
onNext() called on thread ID: 13

Process finished with exit code 0

Also in all of the examples I am still using the BehaviorSubject object to initiate an event and I only found out by accident that this will give the desired result.

What concerns me is that I may be using Observable and BehaviorSubject in an incorrect way that just 'happens' to give me the correct result i.e. subscribers are called on the background thread. Unless I've missed it somewhere in the documentation it doesn't appear obvious how to use these objects to get the desired result.

user_6426800
Related