RxJava - emit an observable every second

Question

I am making a timer in Android using RxJava. I need to make a timer in RxJava to emit an observable every second. I have tried the following but with no luck. Any thoughts on what I am doing wrong?

Observable.interval(1000L, TimeUnit.MILLISECONDS)
          .timeInterval()
          .observeOn(AndroidSchedulers.mainThread())
          .subscribe({Log.d(LOG_TAG, "&&&& on timer") })

Solution 1

Your code seems not to be called. Check whether it is executed and when. As of working with Observable, it is completely correct.

For example, I put your snippet inside onCreate(...) of my MainActivity:

override fun onCreate(savedInstanceState: Bundle?) {
    super.onCreate(savedInstanceState)
    Observable.interval(1000L, TimeUnit.MILLISECONDS)
            .timeInterval()
            .observeOn(AndroidSchedulers.mainThread())
            .subscribe { Log.d("tag", "&&&& on timer") }
    // ...
}

And it works:

Also, probably you don't need .timeInterval() because Observable.interval(...) itself emits sequential numbers within the specified rate, and .timeInterval() just transforms it to emit the time intervals elapsed between the emissions.

Solution 2

In Kotlin & RxJava 2.0.2 simply define an Observable Interval with an initialDelay 0 and period 1 which will emit list item each second.

val list = IntRange(0, 9).toList()
val disposable = Observable
    .interval(0,1, TimeUnit.SECONDS)
    .map { i -> list[i.toInt()] }
    .take(list.size.toLong())
    .subscribe {
       println("item: $it")
    }
Thread.sleep(11000)

Solution 3

In your subscribe() you don't consume the longTimeInterval object that's returned by the timeInterval() operator.

Correct version:

.subscribe(longTimeInterval -> {
     Log.d(LOG_TAG, "&&&& on timer"); 
}

Also I think you don't need the timeInterval() operator at all. Observable.interval() will emit an observable every second in your case, which I guess is what you want. timeInterval() transforms that to an observable that holds the exact time difference between two events occur, I doubt you'll need that.

Comments

[hotkey]

Kotlin doesn't require the lamda parameter declaration in case of single parameter, so that's not the problem. .subscribe { Log.d(...) } is correct RxJava usage in Kotlin.

[fergdev]

I do not need the .timeInterval() :) I removed it but still does not work :/

[Vesko]

Good one, thanks for pointing it out! Got to look at Kotlin soon :)

[Karan Sharma]

When I use this, I get a lint error: Missing Disposable handling

[kukroid]

@KaranSharma , you can wrap it with disposable meaning you can write something like disposable = Observable.interval().... and then in onDestroy of Activity use disposable.dispose()