Netflix/RxJava — pull request #797 by benjchristensen (2014-02-02 05:40:27+00:00)
Title: Scheduler Outer/Inner [Preview]

Following are proposed changes to the `Scheduler` signature based on discussions between @headinthebox and I intended to simplify scheduling and make it easier to do the right thing.

This originates from three findings:

1) It was very easy to cause memory leaks or inadvertent parallel execution since the distinction between outer and inner scheduling was not obvious.

To solve this the new design explicitly has the outer `Scheduler` and then `Scheduler.Inner` for recursion.

2) The passing of state is not useful since scheduling over network boundaries with this model does not work.

In this new design all state passing signatures have been removed. This was determined while implementing a `RemoteScheduler` that attempted to use `observeOn` to transition execution from one machine to another. This does not work because of the requirement for serializing/deserializing the state of the entire execution stack. Migration of work over the network has been bound to be better suited to explicit boundaries established by Subjects. Thus, the complications within the Schedulers are unnecessary.

3) The number of overloads with different ways of doing the same things were confusing.

This new design removes all but the essential and simplest methods.

This is the new signature for `Scheduler`:

```java
public abstract class Scheduler {

// the primary entry point, it immediately schedulers work on a new thread and executes it
// the return subscription will shut down the Inner scheduler when unsubscribed
public abstract Subscription schedule(Action1 action);
public Subscription schedule(final Action1 action, final long delayTime, final TimeUnit unit);
public int degreeOfParallelism();
public long now();

public abstract class Inner implements Subscription {

public abstract void schedule(Action1 action);
public abstract void schedule(Action1 action, long delayTime, TimeUnit unit);
public void schedulePeriodically(final Action1 action, long initialDelay, long period, TimeUnit unit);
// I question having schedulePeriodically here, since recursion allow the same behavior, and with conditional logic
public void schedule(Action1 action, Date dueTime);
}
}
```

The simplest execution of a single task is:

```java
Schedulers.newThread().schedule(new Action1() {

@Override
public void call(Inner inner) {
doWork();
}

});
```

Recursion is easily invoked:

```java
Schedulers.newThread().schedule(new Action1() {

@Override
public void call(Inner inner) {
doWork();
// recurse until unsubscribed (the schedule will do nothing if unsubscribed)
inner.schedule(this);
}

});
```

The use of `Action1` on both the outer and inner levels makes it so recursion that refer to `this` and it works easily.

Similar to the new `lift`/`create` pattern with `Subscriber` the `Inner` is also a `Subscription` so it allows efficient loops with `unsubscribe` support:

```java
Schedulers.newThread().schedule(new Action1() {

@Override
public void call(Inner inner) {
while(!inner.isUnsubscribed()) {
doWork();
}
}

});
```

An action can now `unsubscribe`	the `Scheduler.Inner`:

```java
Schedulers.newThread().schedule(new Action1() {

@Override
public void call(Inner inner) {
while(!inner.isUnsubscribed()) {
int i = doOtherWork();
if(i > 100) {
// an Action can cause the Scheduler to unsubscribe and stop
inner.unsubscribe();
}
}
}

});
```

Typically just stopping is sufficient:

```java
Schedulers.newThread().schedule(new Action1() {

@Override
public void call(Inner inner) {
int i = doOtherWork();
if (i < 10) {
// recurse until done 10
inner.schedule(this);
}
}

});
```

but if other work in other tasks is being done and you want to unsubscribe conditionally you could:

```java
Schedulers.newThread().schedule(new Action1() {

@Override
public void call(Inner inner) {
int i = doOtherWork();
if (i < 10) {
// recurse until done 10
inner.schedule(this);
} else {
inner.unsubscribe();
}
}

});
```

and the recursion can be delayed:

```java
Schedulers.newThread().schedule(new Action1() {

@Override
public void call(Inner inner) {
doWork();
// recurse until unsubscribed ... but delay the recursion
inner.schedule(this, 500, TimeUnit.MILLISECONDS);
}

});
```

The methods on the `Inner` never return a `Subscription` because they are always a single thread/event-loop/actor/etc and controlled by the `Subscription` returned by the initial `Scheduler.schedule` method. This is part of clarifying the contract.

Thus an `unsubscribe` controlled from the outside would be done like this:

```java
Subscription s = Schedulers.newThread().schedule(new Action1() {

@Override
public void call(Inner inner) {
while(!inner.isUnsubscribed()) {
doWork();
}
}

});

// unsubscribe from outside
s.unsubscribe();
```

I'd appreciate feedback on this design direction.

---

NOTE: This pull request is not yet complete. I have not modified the language adaptors or other modules, and there are 3 unit tests in core failing (related to `buffer` and `window`).