How to filter values that change over delta threshold

user_3606820 0
Viewed 373

I have an Observable that emits numbers (integers, doubles, BigDecimal doen'st matter actually). And I have a threshold value defined at observable composition time. Threshold is not going to change during observable life time, in other words.

What I need is to filter out values that differ from the last passed one within the threshold. Or rephrase it: pass value downstream only if it differs from the last passed one more than the threshold.

Example would explain it better:

Source observable values: 
[5, 4, 7, 9, 15, 14, 13, 12, 11, 7, 3, 2, 1]

Filetered values with threshold = 3 : 
[5, 9, 15, 11, 7, 3]

To illustrate it better, lets assume that it's temperature readings, for example. I want to filter only those values where temperature changes more than 3 degree. The very first value always passes and serves as an initial value to compare to. If for example first temperature reading was 21 Celsius and within an hour each subsequent reading was in [18..24] then none of those values should pass downstream. But as soon as it gets over those bounds it should be passed and create new bounds to compare to.

The question is: how to do this with RxJava operators only? Or there's no other way besides storing state outside rx pipeline (in volatile or atomic reference or any other synchronized state, details don't matter) ?

rx-java2 rx-java
1 Answers

You could use a combination of scan and distinctUntilChanged to achieve this with just RX operators.

scan allows you to access the previous and next values, allowing you to compare against your given threshold.

If the threshold isn't met then you could just emit the previous value.

distinctUntilChanged could then be used to drop duplicate emissions.

in Kotlin this would look like the following:

val threshold = 3
val list = listOf(5, 4, 7, 9, 15, 14, 13, 12, 11, 7, 3, 2, 1)

return Observable.fromIterable<Int>(list)
        .scan { previous: Int, next: Int ->
            val difference = abs(next - previous)
            when {
                difference > threshold -> next
                else -> previous
            }
        }.distinctUntilChanged()
user_95492990
Related