Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

How to use RxJava Interval Operator

Tags:

rx-java

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?

like image 449
HuangDong.Li Avatar asked Jun 06 '16 08:06

HuangDong.Li


People also ask

What is observable interval?

Interval Method (TimeSpan) Returns an observable sequence that produces a value after each period.

What is onNext in RxJava?

onNext(): This method is called when a new item is emitted from the Observable. onError(): This method is called when an error occurs and the emission of data is not successfully completed. onComplete(): This method is called when the Observable has successfully completed emitting all items.

What is Completable RxJava?

Single and Completable are new types introduced exclusively at RxJava that represent reduced types of Observable , that have more concise API. Single represent Observable that emit single value or error. Completable represent Observable that emits no value, but only terminal events, either onError or onCompleted.


1 Answers

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.

like image 55
Reut Sharabani Avatar answered Jan 04 '23 03:01

Reut Sharabani