How to combine following Observables into one in Kotlin?

Question

I have these two Observables in Kotlin where is just act as a timer and another one is HTTP network call response Observer.

timerDisposable = Observable.timer(daleyABCControlResetSeconds, TimeUnit.SECONDS, AndroidSchedulers.mainThread())
        .doOnNext {
            if (getABCUpdate() != null) {
                Log.d("ABC", "Media status reset after 3 seconds: ")
                updateABCResponse(getABCUpdate())
            }
        }.subscribe()

disposable = audioApi.setABCUpdate(abcUpdate)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe({
            timerDisposable.dispose()
            updateABCResponse(it)
            Log.d("ABC", "Media Status updated:")
        }, {
            Log.d("ABC", "Error updating Media Status: " + it.message)
            isABCControlChangeRequested = false
        })

I am not satisfied with this approach, can anyone please direct me right direction to use the rx's full potential. Thanks in advance.

EDIT

  Observable.combineLatest(Observable.timer(daleyABCControlResetSeconds, TimeUnit.SECONDS, AndroidSchedulers.mainThread())
            .doOnNext {
                if (getABCUpdate() != null) {
                    Log.d("ABC", "Media status reset after 3 seconds: ")
                    updateABCResponse(getABCUpdate())
                }
            },

            audioApi.setABCUpdate(abcUpdate)
            .subscribeOn(Schedulers.io())
            .observeOn(AndroidSchedulers.mainThread()),
            BiFunction<Long, ABCStatusUpdate, ABCStatusUpdate> { _, abcStatusUpdate ->
                abcStatusUpdate
            })
            .subscribe({
                timerDisposable.dispose()
                updateABCResponse(abcStatusUpdate)
                Log.d("ABC", "Media Status updated:")
            }, {
                Log.d("ABC", "Error updating Media Status: " + abcStatusUpdate.vol)
                isABCControlChangeRequested = false
            })

Answer 1

You can use combinelatest, zip or merge for combinig. I think in your case combinelatest is suitable

Observable.combineLatest(
            Observable.timer(daleyABCControlResetSeconds, TimeUnit.SECONDS, AndroidSchedulers.mainThread())
                .doOnNext {
                    if (getABCUpdate() != null) {
                        Log.d("ABC", "Media status reset after 3 seconds: ")
                        updateABCResponse(getABCUpdate())
                    }
                },
            audioApi.setABCUpdate(abcUpdate)
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread()),
            BiFunction<Long, YourApiResponseType, YourApiResponseType> { _, response ->
                response})
        .subscribe({
            timerDisposable.dispose()
            updateABCResponse(it)
            Log.d("ABC", "Media Status updated:")
        }, {
            Log.d("ABC", "Error updating Media Status: " + it.message)
            isABCControlChangeRequested = false
        })

UPD: You can change your code like this:

Observable.timer(5, TimeUnit.SECONDS, AndroidSchedulers.mainThread()).startWith(-1L)
        .doOnNext {
            if (it == -1L) return@doOnNext
            //your condition
        }