Scheduler Simplification

[benjchristensen]

API redesign (simplification) as per discussion at #997

The Scheduler API in this pull request is:

class Scheduler {
    public abstract Inner Inner(); 
    public int degreeOfParallelism();
    public long now();

    public abstract static class Inner implements Subscription {
        public abstract Subscription schedule(Action0 action, long delayTime, TimeUnit unit);
        public abstract Subscription schedule(Action0 action);
        public final void schedulePeriodically(Action0 action, long initialDelay, long period, TimeUnit unit);
        public long now();
    }
}

While migrating to this I found a 4th use case in addition to the previously discussed 3 that required the schedule methods inside Inner to also return Subscription for canceling the individual tasks as opposed to the entire Inner.

Here are the use cases:

1) Single Action

final Inner is = scheduler.inner();
is.schedule(new Action0() {

    @Override
    public void call() {
        // do work here
    }
})

is.unsubscribe()

with lambda

final Inner is = scheduler.inner);
is.schedule(() -> {
        // do work here
})

is.unsubscribe()

2) Inner Recursion

final Inner is = scheduler.inner();
is.schedule(new Action0() {

    @Override
    public void call() {
        // do work here then recursively reschedule
    is.schedule(this); // this will NOT work with lambdas, only anonymous inner classes
    }
})

is.unsubscribe()

3) Outer Recursion

final Inner is = scheduler.inner();

public void onNext(T t) {
    is.schedule(new Action0() {

        @Override
        public void call() {
            // do work here
        }
    })
}

is.unsubscribe()

4) Outer Recursion with Task Cancellation

This is used for things like debounce and throttleLast where tasks are being scheduled and cancelled as onNext notifications are received.

final Inner is = scheduler.inner();
SerialSubscription serial = new SerialSubscription();

public void onNext(T t) {
    // schedule a new task and cancel previous if not yet executed
    serial.set(is.schedule(new Action0() {

        @Override
        public void call() {
            // do work here
        }
    }));
}

is.unsubscribe()

[cloudbees-pull-request-builder]

RxJava-pull-requests #962 SUCCESS
This pull request looks good

[akarnokd]

In ExecutorScheduler, the innerSubscription is replaced if schedule is called multiple times, for example, with different delay values; unsubscribing the inner cancels only the latest schedule. Is this intentional?

[benjchristensen]

No it's not intentional. Thanks for seeing that. The ExecutorScheduler really needs to be completely deleted though. It breaks the contract by allowing concurrent execution. I know of no good way to make it work without building event loops on top of the thread pools which defeats the purpose.

I'm leaning heavily towards deleting ExecutorScheduler completely.

See #713 and #711

I will replace the computation() scheduler with a pool of event loops.

[benjchristensen]

Merging this ... don't want to conflate this PR with the one where I eliminate ExecutorScheduler.