Program using interval and take does not terminate

[samuelgruetter]

When I run the following snippet:

static Action1<Long> onNextFunc(final String who) {
    return new Action1<Long>() {
        public void call(Long x) {
            System.out.println(who + " got " + x);
        }            
    };
}
static Action1<Throwable> onErrorFunc(final String who) {
    return new Action1<Throwable>() {
        public void call(Throwable t) {
            t.printStackTrace();
        }            
    };
}    
static Action0 onCompleteFunc(final String who) {
    return new Action0() {
        public void call() {
            System.out.println(who + " complete");
        }            
    };
}
public static void main(String[] args) {
    Observable<Long> oneNumberPerSecond = Observable.interval(1, TimeUnit.SECONDS).take(5);
    oneNumberPerSecond.subscribe(onNextFunc("subscriber 1"), onErrorFunc("subscriber 1"), onCompleteFunc("subscriber 1"));
}

then I get (as expected) the following output:

subscriber 1 got 0
subscriber 1 got 1
subscriber 1 got 2
subscriber 1 got 3
subscriber 1 got 4
subscriber 1 complete

However, the program does not terminate, and that's unexpected.

I ran this test using rxjava-core-0.13.2-SNAPSHOT.jar built from commit 00d7c3b (Sat Sep 14 08:37:09 2013 -0700).

Due to this problem, some tests in the RxScalaDemo that I'm working on do not terminate, which is annoying.

[benjchristensen]

Just replicated it ... taking a look to see if I can find the cause.

[benjchristensen]

The OperationInterval class was creating a new ExecutorService on every invocation, and that is never shut down, and launches non-daemon threads.

Schedulers.executor(Executors.newSingleThreadScheduledExecutor())

Changing that, the demo code now doesn't even run as the app quits immediately as it's all async (as it should).

I revised the example to use BlockingObservable for the demo (this and unit tests are where I generally use BlockingObservable) and it runs, then exits nicely.

    public static void main(String[] args) {
        Observable<Long> oneNumberPerSecond = Observable.interval(1, TimeUnit.SECONDS).take(5);
        oneNumberPerSecond.toBlockingObservable().forEach(onNextFunc("subscriber 1"));
    }

I'll submit a fix shortly.

[samuelgruetter]

Thanks, now it behaves as I'd expect, but the example I'd like to get to work still doesn't because of problems with groupBy #289 (comment) ...