Java Rx - filter timed events

Question

Let assume my observable emit integers. I want my observer fire integer x if in last 30 seconds x was not generated by observable.

The behavior is similar to denounce but inverted.

Answer 1

Maybe this is not the more elegant and concise solution but i think you can use a filter like one below

This solution is not perfect because : is triggered by emitter so to work as expected you have to emit a fake event after TIME to flush the last

private static class TimedFilter<T> implements   Func1<T, Collection<T>>{

    private NavigableMap<Long,T> treeMap = new TreeMap<>();
    private HashMap<T,Long> indexes = new HashMap<>();
    private final long delta_millis;

    public TimedFilter(long delta_millis) {
        this.delta_millis = delta_millis;
    }

    @Override
    public Collection<T> call(T x) {

        long now_millis = System.currentTimeMillis();

        Long oldIndex = indexes.put(x, now_millis);

        if(oldIndex!=null)
            treeMap.remove(oldIndex); // throws NPE - if the specified key is null and this map uses natural ordering

        treeMap.put(now_millis,x);

        long then_millis = now_millis - delta_millis;
        Collection<T> values = treeMap.headMap(then_millis).values();
        for (T v : values){
            indexes.remove(v);
        }

        treeMap = treeMap.tailMap(then_millis, true);
        return values;
    }
}

test

void testFunction(){
    getEmitter()
            .map(new TimedFilter<>(TIME))
            .flatMap(Observable::from)
            .subscribe(System.out::print)
    ;
}

Update: as @tilois suggested System.nanoTime() is more reliable, if you run un Android can also use SystemClock.elapsedRealtime() that return millis. ;