Schedulers (merge of pull #199)

[benjchristensen]

Manual merge of #199 by @mairbek plus the following changes:

• made some classes non-public so they don't become part of the published API (if we find they have value in the public API we can make them so but once published it's hard to remove them so I'd rather keep them implementation details until then)
• consolidated ExecutorScheduler and ScheduledExecutorScheduler
• made ExecutorScheduler use a system-wide ScheduledExecutorScheduler for handling delayed events when only an Executor is available
• made the IOThreadPool use a non-bounded cached thread-pool

Next step from here is to review all operator implementations and add the Scheduler overloads.

[cloudbees-pull-request-builder]

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

[cloudbees-pull-request-builder]

RxJava-pull-requests #78 FAILURE
Looks like there's a problem with this pull request

[benjchristensen]

The unit test is poorly written (non-deterministic). I'll fix in the morning.

[jmhofer]

Awesome work! - With this, it should be easy to implement operators like "sample" etc.

[benjchristensen]

Yes it should open the gates to many operators we haven't been able to pursue.

[benjchristensen]

Committed a few tweaks and fixes.

Open questions for me are:

• we're not using the Scheduler.now value anywhere, should we be? or is that only for the Virtual scheduler used for testing?
• not quite sure how the overloads should work on merge etc ... I've implemented them but the behavior might not yet match C#

[benjchristensen]

Can someone with an Rx.Net environment setup implement a test similar to this from Java and tell me the output?

@Test
    public void testMixedSchedulers() throws InterruptedException {
        final String mainThreadName = Thread.currentThread().getName();

        Observable<String> o = Observable.<String> create(new Func1<Observer<String>, Subscription>() {

            @Override
            public Subscription call(Observer<String> observer) {

                System.out.println("Origin observable is running on: " + Thread.currentThread().getName());

                assertFalse(Thread.currentThread().getName().equals(mainThreadName));
                assertTrue("Actually: " + Thread.currentThread().getName(), Thread.currentThread().getName().startsWith("RxIOThreadPool"));

                observer.onNext("one");
                observer.onNext("two");
                observer.onNext("three");
                observer.onCompleted();

                return Subscriptions.empty();
            }
        }).subscribeOn(Schedulers.threadPoolForIO()); // subscribe to the source on the IO thread pool

        // now merge on the CPU threadpool
        o = Observable.<String> merge(o, Observable.<String> from("four", "five"))
                .subscribeOn(Schedulers.threadPoolForComputation())
                .map(new Func1<String, String>() {

                    @Override
                    public String call(String v) {
                        // opportunity to see what thread the merge is running on
                        System.out.println("Merge is running on: " + Thread.currentThread().getName());
                        return v;
                    }

                });

        final CountDownLatch latch = new CountDownLatch(1);

        final AtomicReference<RuntimeException> onError = new AtomicReference<RuntimeException>();

        // subscribe on a new thread
        o.subscribe(new Observer<String>() {

            @Override
            public void onCompleted() {
                System.out.println("==> received onCompleted");
                latch.countDown();
            }

            @Override
            public void onError(Exception e) {
                System.out.println("==> received onError: " + e.getMessage());
                onError.set((RuntimeException) e);
                latch.countDown();
            }

            @Override
            public void onNext(String v) {
                System.out.println("==> Final subscribe is running on: " + Thread.currentThread().getName());
                System.out.println("==> onNext: " + v);

            }
        }, Schedulers.newThread());

        // wait for the above to finish or blow up if it's blocked
        latch.await(5, TimeUnit.SECONDS);
    }

I'm trying to understand how a sequence should work when multiple subscribeOn operators are applied at different steps of a sequence.

Of course Rx.Net doesn't have the IO and CPU thread pools ... those are just helper methods to Executors which would be 2 separate threadpools for different work types so you'll need to adjust that.

[cloudbees-pull-request-builder]

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

[benjchristensen]

Is there any reason to use Scheduler.schedule on an internal operator implementation instead of just leveraging subscribeOn (other than when we need the delay arguments of course)?

For example, on a method overload of merge is this okay?

        return merge(source).subscribeOn(scheduler);

or is there some reason to inside the OperationMerge do this:

                return scheduler.schedule(new Func0<Subscription>() {

                    @Override
                    public Subscription call() {
                        return new MergeObservable<T>(o).call(observer);
                    }
                });

They seem to accomplish the same thing but would like to know if there's a reason to prefer one over the other.

I prefer just reusing subscribeOn.

I can't tell from reading C# code what it does as I can't find the extensions that implement the override methods!

[joshgord]

Add param for dueTime

[benjchristensen]

I'm pulling the trigger and merging this into master so people can start playing with it and providing feedback.

The public API changes are fairly limited still so most changes (which I fully expect) will be implementation details.

[cloudbees-pull-request-builder]

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