Skip to content

TestScheduler: Fix race condition in queue#3832

Closed
nerd2 wants to merge 1 commit into
ReactiveX:1.xfrom
nerd2:1.x
Closed

TestScheduler: Fix race condition in queue#3832
nerd2 wants to merge 1 commit into
ReactiveX:1.xfrom
nerd2:1.x

Conversation

@nerd2

@nerd2 nerd2 commented Apr 5, 2016

Copy link
Copy Markdown

No description provided.

@akarnokd

akarnokd commented Apr 5, 2016

Copy link
Copy Markdown
Member

TestScheduler is a synchronous Scheduler that has to run on a single thread and its main purpose is to force async operators to execute synchronously as well.

Is there a reason you wanted this change?

@nerd2

nerd2 commented Apr 5, 2016

Copy link
Copy Markdown
Author

Humm. This is the crash I was trying to fix, looked like a race but perhaps I'm using the testscheduler incorrectly somehow.

java.lang.NullPointerException
at rx.schedulers.TestScheduler$TimedAction.access$200(TestScheduler.java:37)
at rx.schedulers.TestScheduler.triggerActions(TestScheduler.java:111)
at rx.schedulers.TestScheduler.advanceTimeTo(TestScheduler.java:97)
at rx.schedulers.TestScheduler.advanceTimeBy(TestScheduler.java:84)

@nerd2

nerd2 commented Apr 5, 2016

Copy link
Copy Markdown
Author

My class under test is subscribing to the scheduler asynchronously, I guess you're saying TestScheduler wasn't designed to operate list this?

Observable.interval(mPollTimeMinutes, TimeUnit.MINUTES, mScheduler)
.subscribeOn(Schedulers.newThread()).subscribe(v -> {
getMyUpgrade();
}, v -> {
});

@akarnokd

akarnokd commented Apr 5, 2016

Copy link
Copy Markdown
Member

It won't work with regular schedulers, for example:

Observable.interval(1, TimeUnit.MILLISECONDS) // <-- note: it uses the default computation() scheduler
.observeOn(testScheduler)
.subscribe();

is prone to concurrency issues.

If you use TestScheduler, it has to be applied everywhere.

@nerd2

nerd2 commented Apr 5, 2016

Copy link
Copy Markdown
Author

Thanks for helping - is this how it's intended to be used?

Observable.interval(1, TimeUnit.MILLISECONDS, test ? testScheduler : Schedulers.computation())
.observeOn(test ? testScheduler : Schedulers.newThread())
.subscribe();

@akarnokd

akarnokd commented Apr 5, 2016

Copy link
Copy Markdown
Member

Yes.

@nerd2

nerd2 commented Apr 5, 2016

Copy link
Copy Markdown
Author

Awesome, thanks.

@nerd2 nerd2 closed this Apr 5, 2016
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants