How to emit cumulative sum only from a reactive stream?

Question

I've a use case where the stream should only emit when the cumulative "sum" equals or exceeds a given value, n. Let's take the example of six integers with n = 5.

+---+------+---------+
| i | Emit |   Sum   |
+---+------+---------+
| 1 |    - | 1       |
| 2 |    - | 3       |
| 3 |    5 | 1       |
| 4 |    5 | 0       |
| 5 |    5 | 0       |
| 2 |    2 | 0 (end) |
+---+------+---------+

As you can see, nothing is emitted unless the sum equals or exceeds 5, except for the last element, which is emitted anyway.

Once an item is emitted, the sum gets reduced by that value (n). In reality, I'm reading data from a network call, and subsequently sending them to a downstream consumer who only accepts fixed size chunks, except for the last one, of course (upstream completed).

I'm using project Reactor Flux as the Publisher; I couldn't find any method on it that allows me do what is shown above. scan comes closest, but it also emits intermediate elements that need to be filtered out.

Answer 1

In reality, I'm reading data from a network call, and subsequently sending them to a downstream consumer who only accepts fixed size chunks, except for the last one, of course (upstream completed).

It occurred to me that trying to split the response Flux myself is probably little late and quite difficult; instead, I could use something like Netty FixedLengthFrameDecoder, which does exactly what I'm looking for.

That led me to reactor-netty source code, and after extensive digging, I found exactly what I needed.

fun get(url: String, maxChunkSize: Int): List<ByteArray> {
    return HttpClient.create()
        .httpResponseDecoder { it.maxChunkSize(maxChunkSize) }
        .get()
        .uri(url)
        .responseContent()
        .asByteArray()
        .collectList()
        .block()!!
}

The crucial part is httpResponseDecoder { it.maxChunkSize(maxChunkSize) }; a unit test proves this to be working:

@Test

fun testHonorsMaxChunkSize() {
    val maxChunkSize = 4096
    val chunks = FixedLengthResponseFrameClient.get(
        "http://doesnotexist.nowhere/binary", maxChunkSize
    )

    assertThat(chunks.subList(0, chunks.size - 1))
        .allMatch { it.size ==  maxChunkSize}
    assertThat(chunks.last().size).isLessThanOrEqualTo(maxChunkSize)
}

WebClient can be configured with a custom HttpClient (configured with httpResponseDecoder) as shown below:

WebClient
  .builder()
  .clientConnector(ReactorClientHttpConnector(httpClient))
  .build()
  .get()
  .uri("uri")
  .exchange()
  .flatMapMany { it.body(BodyExtractors.toDataBuffers()) }
  ...

The size of these buffers would be what's set in the HttpClient.httpResponseDecoder (8192 Kb by default).

Answer 2

This is not possible to do directly on Flux object, but you might achieve solution if you have access to resource from which the Flux object is created. Since inside stream (Flux) you are not able access previous element's you can create Flux over indices to you resource and access this resource (since its read only operation) from that Flux of indices directly. For example something like this:

List<Integer> list = List.of(1, 2, 3, 4, 5, 2);
AtomicReference<Integer> atomicSum = new AtomicReference<>(0);
return Flux.fromStream(IntStream.range(0, list.size() - 1).boxed())
        .flatMap(i -> {
            int sum = atomicSum.updateAndGet((integer -> integer + list.get(i)));
            if (sum >= 5) {
                atomicSum.updateAndGet(integer -> integer - 5);
                return Flux.just(5);
            }

            return (i.equals(list.size() -1))
                    ? Flux.just(list.get(i)) // emit last element even if sum was not 5
                    : Flux.empty();
        }); // emitted element's

Note that this is not good practice to do and i don't advice such solution. Flux objects processing might skip between thread's, so if you modify object outside of Flux you should do it in synchronized way (therefore usage of AtomicReference). List is used only for read-only operation's and therefore it's OK. Also i don't know if that part of the code will actually work but i wanted to show you how you might find solution if you have access to resource over which your Flux object is created.

Edit: even such solution would not work. I have mistaken myslef, Flux object don't skip between threads but might be processed by multiple threads leading single atomic reference to invalid state. This cloud still be solved with some synchronizing mechanism like lock's instead of Atomic reference but is far beyond average developer experience. Are you sure you cannot use scan() function since you can provided your own accumulator function as argument?

Answer 3

If you need to keep a running total or otherwise maintain state from which your flux is derived, a way to go about it is to create a new flux that subscribes to the first flux, and maintains the state with the subscription, e.g

Flux<Long> flux = Flux.just(1L, 2L, 3L, 4L, 5L);

Sinks.Many<Long> runningTotalSink = Sinks.many().unicast().onBackpressureBuffer();
Flux<Long> runningTotalFlux = runningTotalSink.asFlux()
        .doOnSubscribe(subscription -> {
            AtomicLong runningTotal = new AtomicLong();
            flux
                    .doOnCancel(subscription::cancel)
                    .doOnError(runningTotalSink::tryEmitError)
                    .doOnComplete(runningTotalSink::tryEmitComplete)
                    .subscribe(i -> {
                        runningTotalSink.tryEmitNext(runningTotal.accumulateAndGet(i, Long::sum));
                    });
        });

runningTotalFlux.toStream().forEach(i -> {
    System.out.println(i);
});