#7: Backpressure with WebFlux – the hookOnSubscribe and hookOnNext methods

🕐 Reading time: 4 minutes

In the world of reactive programming, one of the essential concepts for efficient data handling is backpressure. This mechanism lets you regulate the flow between producer (publisher) and consumer (subscriber), preventing the system from being overloaded. In practice, it allows the consumer to signal when it can't handle any more data – a crucial aspect when the rate at which data is generated outpaces the rate at which it's processed.

In WebFlux, two key methods for managing this dynamic within reactive streams are hookOnSubscribe and hookOnNext. Both belong to the BaseSubscriber class and let you configure the subscription logic and the logic for receiving the stream's elements.

📌 `hookOnSubscribe(Subscription subscription)`

Invoked when the BaseSubscriber subscribes to a publisher, such as a Flux or Mono, it receives a Subscription object as a parameter (representing the relationship between producer and consumer) and lets you configure the initial request of elements to receive via request(n), thus preventing overloads right from the start.

📌 `hookOnNext(T value)`

It's invoked every time a new element (value) is emitted by the publisher, to handle the consumer's specific logic: it takes care of processing the data and includes a new request for elements only when the consumer is ready. This approach guarantees granular control and lets you avoid overload, requesting new elements only after you've finished processing the previous ones.

📑 Example: requests in batches of 5 elements

Let's look at an implementation that shows how to use backpressure in a reactive stream with WebFlux. The test generates a stream of 40 elements and, through a custom BaseSubscriber, the hookOnSubscribe method initially requests a batch of 5 elements, starting the stream in a controlled way. The hookOnNext method is called for each element received and, every time 5 elements have been processed, it requests a new batch of data, handling load peaks without running into inefficiencies or slowdowns.

private static final int BUFFER_SIZE = 5; // number of elements requested in each batch
private static final int LIMIT = 40;
private int countRequests = 0;

@Test
void example() {
    Flux.range(1, LIMIT)
        .subscribe(new BaseSubscriber<>() {
            private int count = 0; //counter for processed elements

            @Override
            protected void hookOnSubscribe(final Subscription subscription) {
                System.out.println("--> hookOnSubscribe");
                System.out.println("Requested the first " + BUFFER_SIZE + " elements");
                request(BUFFER_SIZE);
                countRequests++;
            }

            @Override
            protected void hookOnNext(final Integer value) {
                //data processing — in our example, just printing the value
                System.out.println("Received: " + value);
                count++;
                //once BUFFER_SIZE elements have been processed, request another batch
                if (count % BUFFER_SIZE == 0 && count != LIMIT) {
                    System.out.println("--> hookOnNext");
                    System.out.println("Requested " + BUFFER_SIZE + " elements");
                    request(BUFFER_SIZE);
                    countRequests++;
                }
            }
        });

    assertEquals(LIMIT / BUFFER_SIZE, countRequests);
}

See you at the next pill! ☕

Java Pills