Rx Buffer with timeout since first new group element

Question

pretty new to Rx world, and I need to implement the following behavior:

I need the observable to gather values and emit them as a list once I have at least N items, or if a T amount of time passed since the first item of the group was emitted.

I read the docs again and again, pretty sure it'll use

buffer(timespan, unit, count[, scheduler])

But the issue is that the timespan here is depending on the last group of items.

And if possible, I would also need to be able to flush (force the emission) the current buffer, some items needs to be processed immediately. Am I correct to assume that for such case I would need a second observable, with a treatment to execute before each item and merge both ?

Any idea?

Ps: I work in Java, but I don't require Java code, an explanation will be enough.

Thanks!

Answer 1

The buffering aspect of this question can be achieved via multicast-trickery but I find it much easier to write an operator for it so the data and context are in the same accessible place:

public final class OperatorBufferFirst<T> implements Operator<List<T>, T> {
    final Scheduler scheduler;
    final long timeout;
    final TimeUnit unit;
    final int maxSize;
    public OperatorBufferFirst(
            long timeout, TimeUnit unit, 
            Scheduler scheduler, int maxSize) {
        this.timeout = timeout;
        this.unit = unit;
        this.scheduler = scheduler;
        this.maxSize = maxSize;
    }

    @Override
    public Subscriber<? super T> call(
            Subscriber<? super List<T>> t) {
        BufferSubscriber<T> parent = new BufferSubscriber<>(
                new SerializedSubscriber<>(t), 
                timeout, unit, 
                scheduler.createWorker(), maxSize);
        t.add(parent);
        return parent;
    }

    static final class BufferSubscriber<T> 
    extends Subscriber<T> {
        final Subscriber<? super List<T>> actual;
        final Scheduler.Worker w;
        final long timeout;
        final TimeUnit unit;
        final int maxSize;
        final SerialSubscription timer;

        List<T> buffer;
        long index;


        public BufferSubscriber(
                Subscriber<? super List<T>> actual, 
                long timeout, 
                TimeUnit unit, 
                Scheduler.Worker w, 
                int maxSize) {
            this.actual = actual;
            this.timeout = timeout;
            this.unit = unit;
            this.w = w;
            this.maxSize = maxSize;
            this.timer = new SerialSubscription();
            this.buffer = new ArrayList<>();
            this.add(timer);
            this.add(w);
        }

        @Override
        public void onNext(T t) {
            List<T> b;
            boolean startTimer = false;
            boolean emit = false;
            long idx;
            synchronized (this) {
                b = buffer;
                b.add(t);
                idx = index;
                int n = b.size();
                if (n == 1) {
                    startTimer = true;
                } else
                if (n < maxSize) {
                    return;
                } else {
                    buffer = new ArrayList<>();
                    index = ++idx;
                    emit = true;
                }
            }

            if (startTimer) {
                final long fidx = idx;
                timer.set(w.schedule(() -> timeout(fidx), timeout, unit));
            }
            if (emit) {
                timer.set(Subscriptions.unsubscribed());
                actual.onNext(b);
            }
        }

        @Override
        public void onError(Throwable e) {
            actual.onError(e);
        }

        @Override
        public void onCompleted() {
            timer.unsubscribe();
            List<T> b;
            synchronized (this) {
                b = buffer;
                buffer = null;
                index++;
            }
            if (!b.isEmpty()) {
                actual.onNext(b);
            }
            actual.onCompleted();
        }

        public void timeout(long idx) {
            List<T> b;
            synchronized (this) {
                b = buffer;
                if (idx != index) {
                    return;
                }
                buffer = new ArrayList<>();
                index = idx + 1;
            }

            actual.onNext(b);
        }
    }

    public static void main(String[] args) {
        TestScheduler s = Schedulers.test();

        PublishSubject<Integer> source = PublishSubject.create();

        source.lift(new OperatorBufferFirst<>(1, TimeUnit.SECONDS, s, 3))
        .subscribe(System.out::println, Throwable::printStackTrace, 
                () -> System.out.println("Done"));

        source.onNext(1);
        source.onNext(2);
        source.onNext(3);

        source.onNext(4);
        s.advanceTimeBy(1, TimeUnit.SECONDS);

        source.onNext(5);
        source.onNext(6);

        s.advanceTimeBy(1, TimeUnit.SECONDS);
        s.advanceTimeBy(1, TimeUnit.SECONDS);

        source.onNext(7);
        source.onCompleted();
    }
}

It accumulates values into a list and starts a timed task for the very first element or emits the buffer if it is full.

As for the flush, that can't be simply done generally and you have to establish a protocol with the operator, let's say flush if the incoming T value is some special kind. For example, you have a FLUSH constant of type T somewhere and whenever the operator encounters this, it should emit the current buffer:

synchronized (this) {
    b = buffer;
    idx = index;
    if (t != FLUSH) {
        b.add(t);
        int n = b.size();
        if (n == 1) {
            startTimer = true;
        } else
        if (n < maxSize) {
            return;
        } else {
            buffer = new ArrayList<>();
            index = ++idx;
            emit = true;
        }
    } else {
        buffer = new ArrayList<>();
        index = ++idx;
        emit = true;
    }
}