Preventing rapid clicks on a button and making a request using rxjava

Bascially, what I am looking for if the user rapidly clicks on the button within 300 milliseconds duration it should accept the last click in that duration

to me sounds more like that debounce operator behaviour. From the documentation

Debounce — only emit an item from an Observable if a particular timespan has passed without it emitting another item

you can see the marble diagram here

private val subject = PublishSubject.create<String>()

init {
    processClick()
}

fun onClick(name: String) {
    subject.onNext(name)
}

private fun processClick() {
    subject
        .debounce(300, TimeUnit.MILLISECONDS)
        .switchMap { getPokemonDetailByName(it) }
        .subscribe(
            { pokemonDetailLiveData.value = it },
            { Timber.e(TAG, it.localizedMessage) }
        )
}

private fun getPokemonDetailByName(name: String): Observable<Pokemon> =   
     pokemonDetailInteractor
        .getPokemonDetailByName(name)
        .subscribeOn(pokemonSchedulers.background())
        .observeOn(pokemonSchedulers.ui())
        .toObservable()

In your case, the getPokemonDetailByName creates a new subscription every time. Instead, send the click events to a Subject, create a single subscription to that stream and apply debounce.

I think you should handle click events as an observable so you can call debounce on clicks themselves. One way to achieve what you look for is to have a class creating an observable from the view being clicked:

public class RxClickObservable {

    public static Observable<String> fromView(View view, String pokemonName) {

        final PublishSubject<String> subject = PublishSubject.create();

        view.setOnClickListener(new View.OnClickListener() {
            @Override
            public void onClick(View v) {
                subject.onNext(pokemonName);
            }
        });

        return subject;
    }

}

and in activity/fragment:

RxClickObservable.fromView(binding.button, pokemonName)
        .subscribeOn(pokemonSchedulers.background())
        .observeOn(pokemonSchedulers.ui())
        .debounce(300, TimeUnit.MILLISECONDS)
        .switchMap(pokemonName ->  pokemonDetailInteractor.getPokemonDetailByName(pokemonName))
        .subscribe(... );

Update: Thanks to Amit Shekhar for this article: Implement Search Using RxJava Operators

There is another interesting approach to achieve this i.e. WatchDog. The concept comes from electronics and hardware design. (to see more information about it, refer to wikipedia)

The main point of a WatchDog is that the delegated job will be done if the WatchDog wouldn't reset until due time.

However, we can implement this concept as the following:

TimerWatchDog.kt

import java.util.*

/**
 * @author aminography
 */
class TimerWatchDog(private val timeout: Long) {

    private var timer: Timer? = null

    fun refresh(job: () -> Unit) {
        timer?.cancel()
        timer = Timer().also {
            it.schedule(object : TimerTask() {
                override fun run() = job.invoke()
            }, timeout)
        }
    }

    fun cancel() = timer?.cancel()

}

Usage:

class MyFragment : Fragment {

    private val watchDog = TimerWatchDog(300)

    override fun onViewCreated(view: View, savedInstanceState: Bundle?) {
        super.onViewCreated(view, savedInstanceState)
        
        button.setOnClickListener {
            watchDog.refresh {
                getPokemonDetailByName(name)
            }
        }
    }
}

In this way, if the user clicks on the button relentlessly with an interval shorter than 300ms, the getPokemonDetailByName(name) wouldn't call. So, only the last click causes calling the function.

It's highly beneficial also where we have a search box that runs a query, based on the user input text. (i.g. adding a TextWatcher on the EditText) It leads to fewer api calls while the user is typing, consequently optimizing resource consumption.

The getPokemonDetailByName() subscribes to a new stream every time it is called.

Instead of subscribing to a new stream every time, just provide a subject to send the data to and map it directly to a LiveData with LiveDataReactiveStreams.fromPublisher().

private val nameSubject = PublishSubject.create<String>()

val pokemonDetailLiveData = nameSubject.distinctUntilChanged()
                .observeOn(pokemonSchedulers.background())
                .switchMap(pokemonDetailInteractor::getPokemonDetailByName)
                .doOnError { Timber.e(TAG, it.localizedMessage) }
                .onErrorResumeNext(Observable.empty())
                .toFlowable(BackpressureStrategy.LATEST)
                .to(LiveDataReactiveStreams::fromPublisher)

fun getPokemonDetailByName(name: String) {
    nameSubject.onNext(name)
}

The observeOn(pokemonSchedulers.background()) operator is needed as subjects handle subscriptions differently. The onErrorResumeNext(Observable.empty()) ensures that only valid objects end up in the LiveData.

Like this only a single stream is subscribed to once the pokemonDetailLiveData is being observed. The PublishSubject ensures that only a user click triggers the update from the API and only a single API call is active at the same time.

Everyone has complicated ways to create an observable out of a button click. Rxjs seems to have a built-in way to do this from the very first page:

import { fromEvent } from 'rxjs';

fromEvent(document, 'click').subscribe(() => console.log('Clicked!'));

So take your subscription chain that includes the throttle, but start with the fromEvent method. It turns an event into an observable for you. (I'm not sure where you'd create this, but in C# we do it all in the class constructor.)

Thanks for all the answers.

However, I found a solution that works using RxBinding with debounce operator. I'm posting here as it could be useful to someone else.

override fun onCreateViewHolder(parent: ViewGroup, viewType: Int): PokemonViewHolder {
    binding = PokemonListItemBinding.inflate(LayoutInflater.from(parent.context), parent, false)
    val pokemonViewHolder = PokemonViewHolder(binding.root)

    pokemonViewHolder.itemView.clicks()
        .debounce(300, TimeUnit.MILLISECONDS)
        .subscribeBy(
            onNext = {
                val name = pokemonList[pokemonViewHolder.adapterPosition].name

                if(::pokemonTapped.isInitialized) {
                    pokemonTapped(name)
                }
            },
            onError = { Timber.e(it, "Failed to send pokemon request %s", it.localizedMessage) }
        ).addTo(compositeDisposable)

    return pokemonViewHolder
}

This is based on the answer by @ckunder and slightly modified and works as expected.

    // Start listening for item clicks when viewmodel created
    init {
        observeOnItemClicks()
    }

    // remove clicks that are emitted during the 200ms duration. In the onNext make the actual request  
    private fun observeOnItemClicks() {
        subject
            .debounce(300, TimeUnit.MILLISECONDS)
            .subscribeBy(
                onNext = { pokemonName ->
                    getPokemonDetailByName(pokemonName)
                },
                onError = { Timber.e(it, "Pokemon click event failed ${it.localizedMessage}")}
            )
            .addTo(compositeDisposable)
    }

// No need to change this as this will be called in the onNext of the subject's subscribeBy
fun getPokemonDetailByName(name: String) {
    shouldShowLoading.postValue(true)

    pokemonDetailInteractor.getPokemonDetailByName(name)
        .subscribeOn(pokemonSchedulers.background())
        .observeOn(pokemonSchedulers.ui())
        .subscribeBy(
            onSuccess = { pokemon ->
                shouldShowLoading.postValue(false)
                pokemonDetailLiveData.postValue(pokemon)
            },
            onError = {
                shouldShowLoading.value = false
                Timber.e(TAG, it.localizedMessage)
            }
        ).addTo(compositeDisposable)
}

// emit the name when the user clicks on an pokemon item in the list
fun onPokemonItemClicked(name: String) {
    subject.onNext(name)
}