How to use RxJava Interval Operator

问题:

I'm learning about RxJava operator, and I found these code below did not print anything:

public static void main(String[] args) {

    Observable
    .interval(1, TimeUnit.SECONDS)
    .subscribe(new Subscriber<Long>() {
        @Override
        public void onCompleted() {
            System.out.println("onCompleted");
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("onError -> " + e.getMessage());
        }

        @Override
        public void onNext(Long l) {
            System.out.println("onNext -> " + l);
        }
    });
}

As ReactiveX, interval

create an Observable that emits a sequence of integers spaced by a particular time interval

Did I make a mistake or forget about something?

回答1:

You have to block until the observable is consumed:

public static void main(String[] args) throws Exception {

    CountDownLatch latch = new CountDownLatch(1);

    Observable
    .interval(1, TimeUnit.SECONDS)
    .subscribe(new Subscriber<Long>() {
        @Override
        public void onCompleted() {
            System.out.println("onCompleted");
            // make sure to complete only when observable is done
            latch.countDown();
        }

        @Override
        public void onError(Throwable e) {
            System.out.println("onError -> " + e.getMessage());
        }

        @Override
        public void onNext(Long l) {
            System.out.println("onNext -> " + l);
        }
    });

    // wait for observable to complete (never in this case...)
    latch.await();
}

You can add .take(10) for example to see the observable complete.

回答2:

Put Thread.sleep(1000000) after the subscribe and you will see it working. Observable.interval operates by default on Schedulers.computation() so your stream is being run on a thread other than the main thread.

回答3:

As they tell you already interval works asynchronous, so you have to wait for all the events to finish.

You can get the Subscription once you subscribe and then use TestSubcriber which is part of the reactiveX platform, and which will give you the feature to wait for all events terminates.

       @Test
public void testObservableInterval() throws InterruptedException {
    Subscription subscription = Observable.interval(1, TimeUnit.SECONDS)
              .map(time-> "item emitted")
              .subscribe(System.out::print,
                         item -> System.out.print("final:" + item));
    new TestSubscriber((Observer) subscription)
            .awaitTerminalEvent(100, TimeUnit.MILLISECONDS);
}

I have in my github more examples if you need https://github.com/politrons/reactive/blob/master/src/test/java/rx/observables/scheduler/ObservableAsynchronous.java