Android LiveData - how to throttle results

user_835883 0
Viewed 2988

I have a liveData set up that emits some integers. but i would like to throttle the emissions such that it only happens every 5 seconds. but after the first 5 seconds have passed then the throttling can be removed.

here i am observing a count like this:

count1.observe(this, new Observer() {
      @Override public void onChanged(@Nullable Integer i) {
        //Do something with "integer"
      }
    });

but i want onChanged() callback to be throttled such that its just called maximum of 2 times within the first 5 seconds. After the first 5 seconds have passed i dont need the throttling anymore at all. Im not sure how to do this as i dont see any options for livedata to handle streams. i tried Transformation but it only has map and switchMap. how can this be done ? enter code here

android-livedata android
2 Answers

Here's MediatorLiveData implementation that throttles using a Handler (in Kotlin since Java has a lot of boilerplate).

If source livedata updates too often result emission will be delayed:

import android.os.Handler
import android.os.Looper
import androidx.lifecycle.LiveData
import androidx.lifecycle.MediatorLiveData

/**
 * LiveData throttling value emissions so they don't happen more often than [delayMs].
 */
class ThrottledLiveData<T>(source: LiveData<T>, delayMs: Long) : MediatorLiveData<T>() {
    val handler = Handler(Looper.getMainLooper())
    var delayMs = delayMs
        private set

    private var isValueDelayed = false
    private var delayedValue: T? = null
    private var delayRunnable: Runnable? = null
        set(value) {
            field?.let { handler.removeCallbacks(it) }
            value?.let { handler.postDelayed(it, delayMs) }
            field = value
        }
    private val objDelayRunnable = Runnable { if (consumeDelayedValue()) startDelay() }

    init {
        addSource(source) { newValue ->
            if (delayRunnable == null) {
                value = newValue
                startDelay()
            } else {
                isValueDelayed = true
                delayedValue = newValue
            }
        }
    }

    /** Start throttling or modify the delay. If [newDelay] is `0` (default) reuse previous delay value. */
    fun startThrottling(newDelay: Long = 0L) {
        require(newDelay >= 0L)
        when {
            newDelay > 0 -> delayMs = newDelay
            delayMs < 0 -> delayMs *= -1
            delayMs > 0 -> return
            else -> throw IllegalArgumentException("newDelay cannot be zero if old delayMs is zero")
        }
    }

    /** Stop throttling, if [immediate] emit any pending value now. */
    fun stopThrottling(immediate: Boolean = false) {
        if (delayMs <= 0) return
        delayMs *= -1
        if (immediate) consumeDelayedValue()
    }

    override fun onInactive() {
        super.onInactive()
        consumeDelayedValue()
    }

    // start counting the delay or clear it if conditions are not met
    private fun startDelay() {
        delayRunnable = if (delayMs > 0 && hasActiveObservers()) objDelayRunnable else null
    }

    private fun consumeDelayedValue(): Boolean {
        delayRunnable = null
        return if (isValueDelayed) {
            value = delayedValue
            delayedValue = null
            isValueDelayed = false
            true
        } else false
    }
}

Use it by passing your source live data as first argument and observing it instead:

val throttledCount = ThrottledLiveData(count1, 2500L)  // maximum of one update per 2.5 sec
throttledCount.observe(this, Observer { i: Int ->
    //Do something with "integer"
})

If you want to stop throttling after 5 seconds just post a delayed runnable that will disable it:

val disableThrottle = Runnable { throttledCount.stopThrottling() }
throttledCount.handler.postDelayed(disableThrottle, 5000L)
user_92419780

In your case you should use Interval operator of rxjava.

Interval Operator create an Observable that emits a sequence of integers spaced by a given time interval. It is used when we want to do a task again and again after some interval.

example :

val disposable =
Observable.interval(0, 2, TimeUnit.SECONDS)
    .flatMap {
        return@flatMap Observable.create<String> { emitter ->
            Log.d("IntervalExample", "Create")
            emitter.onNext("MindOrks")
            emitter.onComplete()
        }
    }
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe {
        Log.d("IntervalExample", it)
    }
compositeDisposable.add(disposable)

Here the task will be done again and again after 2 seconds of interval.

One thing to notice: It will keep going on forever.

How to stop that?

There are two ways to stop. The following are the two ways to stop it.

  1. Using the compositeDisposable.dispose()

  2. Using take(n) operator like below

    Observable.interval(0, 2, TimeUnit.SECONDS).take(5).flatMap {
return@flatMap Observable.create<String> { emitter ->
    Log.d("IntervalExample", "Create")
    emitter.onNext("MindOrks")
    emitter.onComplete()
    }
}
.observeOn(AndroidSchedulers.mainThread())
.subscribe {
    Log.d("IntervalExample", it)
}

As we have passed 5 as a parameter in the take(5), the task will only be done 5 times with an interval of 2 seconds.

user_80577020
Related