Allow for thread interruption

[mattrjacobs]

This PR addresses #354, but I would like feedback before I consider merging. Especially from @benjchristensen / @abersnaze.

All of the unit tests that are added (Thanks @aadeon!) now pass, but I would love to see a better way to accomplish this goal.

It's hacky in a few ways:

1. I believe that the Observable.subscribeOn method has a bug where unsubscription is not propagated to the work which is submitted to the new Scheduler. A Subscription is generated in OperatorSubscribeOn in the schedule() method and then forgotten about. I believe that's the precise Subscription we need to unsubscribe from.
2. Upon unsubscription, we need to optionally interrupt the thread. At the moment, Subscriptions.from(Future<?> f) wraps a Future and upon unsubscription, cancels this future. It hardcodes the parameter to cancel, however. This is the value we need to have drawn from the HystrixCommandProperties. So I believe that we should consider adding this to RxJava proper, rather than the hack I've done here. That hack is to not call the createWorker() in the Scheduler interface, but cast the Scheduler to a HystrixContextScheduler and call its (new) createWorker(boolean shouldInterrupt) method. Since this is really only valid for thread pools, I'm not sure how much of this belongs in the Scheduler interface.

[cloudbees-pull-request-builder]

NetflixOSS » Hystrix » Hystrix-pull-requests #15 FAILURE
Looks like there's a problem with this pull request

[cloudbees-pull-request-builder]

Hystrix-pull-requests #254 FAILURE
Looks like there's a problem with this pull request

[mattrjacobs]

Also note that Javanica tests are failing as a result. It wasn't obvious to me what went wrong there, so I might need help from Javanica experts after we sort out the code in hystrix-core

[cloudbees-pull-request-builder]

Hystrix-pull-requests #255 FAILURE
Looks like there's a problem with this pull request

[cloudbees-pull-request-builder]

NetflixOSS » Hystrix » Hystrix-pull-requests #16 FAILURE
Looks like there's a problem with this pull request

[mattrjacobs]

@abersnaze already got 1 of these merged into RxJava master: ReactiveX/RxJava#2575. This addresses the issue where unsubscription was not propagated properly in OperatorSubscribeOn

[akarnokd]

@mattrjacobs I had to revert that PR because it was a no-op in RxJava at best because RxJava's workers track all scheduled tasks and thus unsubscribing the worker while having a task is equivalent of unsubscribing the task. See this test which schedules a bunch of tasks and cancels them all at once with a single unsubscription; the test then shows that no task was retained.

Looking at HystrixCommandScheduler the task-scheduling logic is quite old and has several races (is it even used?). The HystrixContextSchedulerWorker.unsubscribe doesn't unsubscribe the actual worker but only a boolean subscription, thus the RxJava practice of cancelling the entire worker won't affect any scheduled tasks.

Edit: I've checked the documentation of Scheduler.Worker and it states the call to unsubscibe will/has to cancel all outstanding tasks.

[mattrjacobs]

Thanks for taking a look @akarnokd , I appreciate it. Yes, the Scheduler being used is the HystrixContextScheduler.

I am now looking through the examples of concrete Schedulers in RxJava for inspiration. I hope you don't mind if I ask for a review once I get my head wrapped around those and apply those learnings to the HystrixContextScheduler.

[mattrjacobs]

Closing this in favor of #647