RX Java - непрерывно выдавать значения в течение определенного периода

Question

У меня есть датчик ЧСС, который периодически выдает значение (где-то между 500-3000 милли). Когда датчик ЧСС издает недетерминированный c. С RX Java я хотел бы, чтобы константа выдавала значение частоты пульса «последний раз видели», а константа выдает значение в течение 10 секунд, пока она не пометит его как слишком устаревшее и вместо этого отправит NULL. NULL означает, что датчик сердечного ритма больше не выдает показания датчика.

У меня есть следующий (kotlin) код:

    val heartRateObservable: Observable<BLEDataValue> = observable
            .flatMap { it.setupNotification(characteristic.uniqueIdentifier) }
            .flatMap { it }
            .map { BTDataPacket(characteristic.uniqueIdentifier, BleParseable(it)).btValue() }.onErrorReturn { BLEDataValueHeartRate(null) }
    return Observable.combineLatest(Observable.interval(1000, TimeUnit.MILLISECONDS), heartRateObservable, BiFunction { _, t2 -> t2 })

Вопрос: Можно ли ввести способ воспроизвести последнее наблюдаемое значение частоты пульса до тех пор, пока последнее значение не станет устаревшим (то есть после отсутствия показаний пульса в течение 10 секунд) .. когда отображается значение пульса, оно воспроизводится до тех пор, пока не появится новое значение частоты пульса ИЛИ истекло время ожидания в 10 секунд, поскольку последнее значение слишком устарело?

Answer 1

Вы можете использовать операторы takeWhile или takeUntil для завершения наблюдаемой скорости.

подождите 10 выбросов:

Observable heartbeat = Observable.interval(1000, TimeUnit.MILLISECONDS)
.takeWhile(e -> e < 10)

или используя timer witj takeUntil

Observable heartbeat = Observable.interval(1000, TimeUnit.MILLISECONDS)
.takeUntil(Observable.timer(10000, TimeUnit.MILLISECONDS))

После этого вы сказали:

... и передали это в течение 10 000 миллисекунд от последнего полученного значения датчика после ...

Для этого вы можете использовать switchMap (Если я понял ваш вопрос)

heartRateObservable
.switchMap(data -> heartbeat.map(data))

Следовательно, сердцебиение начинает выдавать значения после каждых последних переданных данных.

val heartRateObservable: Observable<BLEDataValue> = observable
            .flatMap { it.setupNotification(characteristic.uniqueIdentifier) }
            .flatMap { it }
            .map { BTDataPacket(characteristic.uniqueIdentifier, BleParseable(it)).btValue() }.onErrorReturn { BLEDataValueHeartRate(null) }
return heartRateObservable
       .switchMap { data -> Observable.interval(1000, TimeUnit.MILLISECONDS)
                     .takeWhile(e -> e < 10) 
                     .map(l -> data)
                  }