Как объединить следующие Observables в один в Котлине? - PullRequest
Новая
       12

Как объединить следующие Observables в один в Котлине?

0 голосов
/ 27 сентября 2018

У меня есть эти два Observable в Котлине, где он просто действует как таймер, а другой - ответ сетевого вызова HTTP 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
        })

Я не удовлетворен этим подходом, может кто-нибудь, пожалуйста, направьте меня в правильном направлении, чтобы использовать полный потенциал rx.Заранее спасибо.

РЕДАКТИРОВАТЬ 

  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
            })

1 Ответ

0 голосов
/ 27 сентября 2018

Вы можете использовать comblatest, zip или merge для объединения.Я думаю, что в вашем случае comblatest подходит 

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: Вы можете изменить свой код следующим образом: 

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