Fix for RxJava 0.18.+ #468

chrisjenx · 2014-04-28T15:42:45Z

Needs review.

This adds a RetrofitScheduler which is loosely based on the NewThreadScheduler but alows us to provide the Executor.

I was not sure the direction the Square guys want to go with this. So apologies if it does not follow your guidelines, I'll happily change/move stuff.

I moved the RetrofitScheduler into package safe Schedulers class to hide and share the API in the RestAdapter and MockRestAdapter.

Few questions still outstanding:

Is Schedulers the right place for the RetrofitScheduler? Not sure if there would ever be more than one?
The RetrofitScheduler does a type check for ExecutorService but thats not guaranteed. Should we force ExecutorService?
MockRetrofitScheduler seems to do waiting of its own, Is there a reason we need to use the httpExecutor or could we just use rx.Schedulers.newThread()?
Should we implement the schedule method in the RetrofitScheduler, I feel that any timing based observable work should not be passed to the Retrofit ThreadPool.

I was toying with a way to allow users to pass through a Scheduler to Retrofit, but I think that would be unnecessary as we can just call retrofit service synchronously.

JakeWharton · 2014-04-28T15:48:59Z

retrofit/src/main/java/retrofit/Schedulers.java

+      @Override
+      public Subscription schedule(final Action0 action) {
+        if (innerSubscription.isUnsubscribed()) {
+          // don't schedule, we are unsubscribed


Capitalize + end with a period. Comments should be sentences (or at least punctuated like them).

JakeWharton · 2014-04-28T15:55:57Z

The RetrofitScheduler does a type check for ExecutorService but thats not guaranteed. Should we force ExecutorService?

We can't force, but we can add an API for it and deprecate the old one.

MockRetrofitScheduler seems to do waiting of its own, Is there a reason we need to use the httpExecutor or could we just use rx.Schedulers.newThread()?

I'd prefer not to do this. If you have a pool of 3 threads and make 10 requests you want to see the queuing behavior that the normal app would exhibit.

Is Schedulers the right place for the RetrofitScheduler? Not sure if there would ever be more than one?

There's an RxJavaSupport class already inside RestAdapter. We should probably promote it to top-level and have this scheduler live inside of it.

Should we implement the schedule method in the RetrofitScheduler, I feel that any timing based observable work should not be passed to the Retrofit ThreadPool.

Agree.

chrisjenx · 2014-04-28T16:06:26Z

We can't force, but we can add an API for it and deprecate the old one.

Unless you feel it's necessary for this issue? I will leave that for another PR.

There's an RxJavaSupport class already inside RestAdapter. We should probably promote it to top-level and have this scheduler live inside of it.

Yeah I started adding to RxSupport, then doubled back. I shall promote RxSupport class then and add it as an inner class.

I'd prefer not to do this. If you have a pool of 3 threads and make 10 requests you want to see the queuing behavior that the normal app would exhibit.

I thought that might be the case, I'll leave as-is then.

codefromthecrypt · 2014-04-28T16:23:12Z

@jhump @dragonsinth you might be interested in this.

chrisjenx · 2014-04-28T16:29:14Z

Made changes as per feedback.

jhump · 2014-04-28T20:26:08Z

retrofit/src/main/java/retrofit/RxSupport.java

+          //  I would potentially force an API change to make sure this is always an
+          //  ExecutorService.
+          s = Subscriptions.empty();
+          executor.execute(getActionRunnable(action, sf));


You could just use new FutureTask<Void>(getActionRunnable(action, sf), null).

…e executor.

chrisjenx · 2014-04-29T09:15:20Z

Made changes as per @jhump. Didn't Realise I could do that (guess I should of checked what submit actually did, my bad!).

@JakeWharton latest change would negate the need to move towards needing an ExecutorService to be provided.

Might be worth @benjchristensen checking it over, to see if I have missed anything.

JakeWharton · 2014-04-29T15:48:15Z

retrofit/src/main/java/retrofit/RxSupport.java

+import rx.subscriptions.Subscriptions;
+
+/**
+ * Indirection to avoid VerifyError if RxJava isn't present.


This comment is no longer useful. Either drop it altogether or change it to something like "Utilities for supporting RxJava Observables".

benjchristensen · 2014-04-29T18:44:04Z

The reason why ExecutorScheduler was removed is because it does not meet the guarantees of sequential execution expected of an Observable. See ReactiveX/RxJava#713 and ReactiveX/RxJava#711.

I considered trying to put event loops on top of an Executor but it's rather tricky, as the Executor queue can't be used, each loop (a Scheduler.Worker) must keep the enqueued work external from the Executor and only allow work to be done sequentially.

Keep this in mind while adding support for Executor here. If the responses only ever emit a single item, or it's only ever subscribeOn that gets used, then it's no big deal. If observeOn or other similar mechanism occur however where multiple onNext are scheduled, an Executor will allow concurrent emission and break the Rx contract.

If someone wants to tackle an ExecutorScheduler implementation that has Worker event loops on top of a multi-threaded Executor and maintains the contract, I'll happily accept it back into RxJava.

Hope this provides some context on this breaking change.

As for how many more breaking changes to expect ... theoretically there should not be any more significant ones as we are near the end of the roadmap to 1.0: ReactiveX/RxJava#1001

benjchristensen · 2014-04-29T18:46:14Z

retrofit/src/main/java/retrofit/RxSupport.java

+          throw new RuntimeException(e);
+        }
+      }
+    }).subscribeOn(scheduler);


If this is the only use of Scheduler within Retrofit, then a multi-threaded Executor is fine, since it will only ever schedule a single action.

Yep. We internalize scheduling for now since RxJava support was kind-of shoved in. In a future world, Retrofit will have a leaner core that an RxJava shim can be built upon where the library doesn't own scheduling at all.

chrisjenx · 2014-04-30T10:40:16Z

So based off of @benjchristensen comments; Implementing the executor into OnSubscribe makes more sense, we never want to expose the RetrofitScheduler to the outside world anyway, it was a workaround for the removal of Schedulers.executor().

This also allows the user to define the subscribeOn Scheduler without having to worry if they will lock up Retrofit Threadpool.
As we don't define our Scheduler, by default Rx uses the computation Scheduler for operators. @benjchristensen if you could confirm?
Which means we don't need to then worry about the user then running operators on our thread pool, potentially locking up requests.

We do ignore the passed in callbackExecutor via the RestAdapter.Builder but Retrofit did that before anyway, so that does not change from the current implementation.
I would expect the user to be defining their own observeOn anyway, the Rx Android plugin provides Activity/Fragment binding.

From what I have understood, I hope that makes sense?

benjchristensen · 2014-04-30T14:59:09Z

default Rx uses the computation Scheduler for operators.

When an operator requires concurrency (such as window or interval) then yes, Schedulers.computation() is the default.

The callback via onNext from your Observable would run on your thread pool though, as Rx does not inject new concurrency unless asked with a particular operator.

This should not be a problem though, as the use of Rx is supposed to be non-blocking. If someone does IO in a callback (such as in a map function) and blocks the threads, they are "doing the wrong thing" :-) That problem exists whether you use subscribeOn or your own executor inside OnSubscribe.

If you do not want the callbacks running on your thread pool, then you would use executor within OnSubscribe and then observable.observeOn(Schedulers.computation())' to move theonNextfrom your thread to thecomputation` event loops.

chrisjenx · 2014-04-30T15:17:03Z

Yeah that co-1insides with what I understood.

I'll amend this pr and add some tests so anyone looking at retrofit will
see how this is intended to work.
On 30 Apr 2014 15:59, "Ben Christensen" notifications@github.com wrote:

default Rx uses the computation Scheduler for operators.

When an operator requires concurrency (such as window or interval) then
yes, Schedulers.computation() is the default.

The callback via onNext from your Observable would run on your thread
pool though, as Rx does not inject new concurrency unless asked with a
particular operator.

This should not be a problem though, as the use of Rx is supposed to be
non-blocking. If someone does IO in a callback (such as in a mapfunction) and blocks the threads, they are "doing the wrong thing" :-) That
problem exists whether you use subscribeOn or your own executor inside
OnSubscribe.

If you do not want the callbacks running on your thread pool, then you
would use executor within OnSubscribe and then observable.observeOn(Schedulers.computation())'
to move theonNextfrom your thread to thecomputation` event loops.

—
Reply to this email directly or view it on GitHubhttps://github.com//pull/468#issuecomment-41806796
.

benjchristensen · 2014-04-30T16:08:52Z

Great. And sorry about the volatility during the 0.x versions! We are very close to stabilizing on 1.0.

chrisjenx · 2014-04-30T17:56:46Z

Done a bit of a round trip here, but I'm much happier with this implementation now.

Hope you guys agree? I might of over engineered the Observer.OnSubscribe. Do i need to add the child subscription?

Other than that, removal of the Subscriber made sense, we shouldn't need to worry about that. We only care about executing the Request and passing it off to something else as quick as possible.

benjchristensen · 2014-04-30T18:05:08Z

retrofit-mock/src/main/java/retrofit/MockRestAdapter.java

-          }
+        @Override public void call(final Subscriber<? super Object> subscriber) {
+          if (subscriber.isUnsubscribed()) return;
+          httpExecutor.execute(new Runnable() {


If you want to support cancelation, you can capture the Future from this and register it with the Subscriber.

For example:

subscriber.add(Subscriptions.from(future));

Oh, this is the MockRestAdapter so it probably doesn't matter ...

benjchristensen · 2014-04-30T18:08:22Z

The RxSupport class looks good.

codefromthecrypt · 2014-04-30T18:48:06Z

Thanks for the advice, Ben!

chrisjenx · 2014-05-02T16:04:28Z

Are you guys happy with this? Let me know if not.

JakeWharton · 2014-05-05T08:01:00Z

@benjchristensen Thanks for all the help!

@chrisjenx I'll take a closer look tomorrow morning but it's looking good.

JakeWharton · 2014-05-08T15:39:44Z

Squashed and merged to master.

chrisjenx added 5 commits April 28, 2014 14:47

added inner RetrofitScheduler

9488e86

removed unused imports

7e3bebc

checkstyle fixes

966254b

migrated Scheduler into the Schedulers class for shared usage.

a048035

checkstyle

ee29804

chrisjenx mentioned this pull request Apr 28, 2014

RxJava update breaks Retrofit #464

Closed

JakeWharton reviewed Apr 28, 2014
View reviewed changes

cleaned up comments as-per feedback.

53fee10

Comment to end-of-line style

37ef1a3

fix checkstyle

f584287

promoted RxSupport to retrofit.

a350444

jhump reviewed Apr 28, 2014
View reviewed changes

as per @jhump's suggestions simplified executing a task. passed to th…

17360ca

…e executor.

JakeWharton reviewed Apr 29, 2014
View reviewed changes

Comments cleanup

e009ea8

benjchristensen reviewed Apr 29, 2014
View reviewed changes

Moved away from Scheduler and Execute inside OnSubscribe.

2765e40

Fixed MockRestAdapter to mimic new OnSubscribe, added RxSupportTests

2c19f42

comment clean up

97db5c9

benjchristensen reviewed Apr 30, 2014
View reviewed changes

loganj mentioned this pull request May 7, 2014

Lazily construct RxSupport #474

Merged

JakeWharton closed this May 8, 2014

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix for RxJava 0.18.+ #468

Fix for RxJava 0.18.+ #468

chrisjenx commented Apr 28, 2014

JakeWharton Apr 28, 2014

JakeWharton commented Apr 28, 2014

chrisjenx commented Apr 28, 2014

codefromthecrypt commented Apr 28, 2014

chrisjenx commented Apr 28, 2014

jhump Apr 28, 2014

chrisjenx commented Apr 29, 2014

JakeWharton Apr 29, 2014

benjchristensen commented Apr 29, 2014

benjchristensen Apr 29, 2014

JakeWharton Apr 29, 2014

chrisjenx commented Apr 30, 2014

benjchristensen commented Apr 30, 2014

chrisjenx commented Apr 30, 2014

benjchristensen commented Apr 30, 2014

chrisjenx commented Apr 30, 2014

benjchristensen Apr 30, 2014

benjchristensen Apr 30, 2014

benjchristensen commented Apr 30, 2014

codefromthecrypt commented Apr 30, 2014

chrisjenx commented May 2, 2014

JakeWharton commented May 5, 2014

JakeWharton commented May 8, 2014

Fix for RxJava 0.18.+ #468

Fix for RxJava 0.18.+ #468

Conversation

chrisjenx commented Apr 28, 2014

Needs review.

JakeWharton Apr 28, 2014

Choose a reason for hiding this comment

JakeWharton commented Apr 28, 2014

chrisjenx commented Apr 28, 2014

codefromthecrypt commented Apr 28, 2014

chrisjenx commented Apr 28, 2014

jhump Apr 28, 2014

Choose a reason for hiding this comment

chrisjenx commented Apr 29, 2014

JakeWharton Apr 29, 2014

Choose a reason for hiding this comment

benjchristensen commented Apr 29, 2014

benjchristensen Apr 29, 2014

Choose a reason for hiding this comment

JakeWharton Apr 29, 2014

Choose a reason for hiding this comment

chrisjenx commented Apr 30, 2014

benjchristensen commented Apr 30, 2014

chrisjenx commented Apr 30, 2014

benjchristensen commented Apr 30, 2014

chrisjenx commented Apr 30, 2014

benjchristensen Apr 30, 2014

Choose a reason for hiding this comment

benjchristensen Apr 30, 2014

Choose a reason for hiding this comment

benjchristensen commented Apr 30, 2014

codefromthecrypt commented Apr 30, 2014

chrisjenx commented May 2, 2014

JakeWharton commented May 5, 2014

JakeWharton commented May 8, 2014