reactor / .github Goto Github PK

Motivation

Hi, I am encountering a somewhat weird behavior when using bufferUntilChanged with a hot publisher.
When an element comes in, it is compared to the previous element and if the keyComparator returns false, the previous element is added to the buffer and the buffer is emitted.
A new buffer is created containing the current element and it is left hanging until a new element arrives or the publisher completes.
If no new element arrives, this new buffer is not emitted.

More specifically, I am trying to implement a session window as defined by Kafka Streams and Flink, so I am actually comparing the event times of the elements.
This window is defined based on the time between two consecutive messages.
If the time between two consecutive messages is less than the specified session gap, then the messages are considered to belong to the same session.
If the gap is larger than the session gap, the window is emitted and a new window is started.

I created a test that shows this behavior, the last element is emitted in its own buffer only when the publisher is completed.

   void testBufferUntilChanged() {
        var testPublisher = TestPublisher.<Pair<Integer, Long>>create();
        var flux = testPublisher.flux().bufferUntilChanged(Function.identity(), (pair1, pair2) -> {
            var diff = pair2.getRight() - pair1.getRight();
            System.out.println("Diff " + diff);
            return diff < 500L;
        });
        
        StepVerifier
                .create(flux)
                .then(() -> testPublisher.next(Pair.of(1, System.currentTimeMillis())))
                .thenAwait(Duration.ofMillis(400L))
                .then(() -> testPublisher.next(Pair.of(2, System.currentTimeMillis())))
                .thenAwait(Duration.ofMillis(400L))
                .then(() -> testPublisher.next(Pair.of(3, System.currentTimeMillis())))
                .thenAwait(Duration.ofMillis(600L))
                .then(() -> testPublisher.next(Pair.of(4, System.currentTimeMillis())))
                .assertNext(buffer -> assertThat(buffer.stream().map(Pair::getKey).collect(Collectors.toList()), containsInAnyOrder(1, 2, 3)))
                .thenAwait(Duration.ofMillis(1000L))
                .then(testPublisher::complete)
                .assertNext(buffer -> assertThat(buffer.stream().map(Pair::getKey).collect(Collectors.toList()), containsInAnyOrder(4)))
                .verifyComplete();
    } 

Desired solution

As an enhancement to the bufferUntilChanged methods, do you think it would be useful to create a new signature with a new Duration parameter?
It would be similar to other buffer methods that have Duration parameters.
If a new element does not arrive in the specified Duration interval, the buffer will be emitted no matter what.

Considered alternatives

As a workaround, I am thinking of adding some kind of scheduled heartbeat elements to the publisher.
I am using Flux.create, so I can emit items from a scheduled service.
But it is not ideal, I believe it will complicate the processing pipeline unnecessarily.
Thanks.

Additional context