How to Reactively read a Queue with Dynamic Delay using Mutiny in Java?

[allanh1193cr]

Currently I have a service which read events from a queue in SQS and process the found messages. I have a concurrency with 2 threads reading at the same time the queue to process the messages. The iteration is one second to validate if there are messages in the queue. This has increased the AWS costs a lot. Looking for a solution, Now I want to have a dynamic delay or something similar when the service read the messages from the queue. These are the steps I want to implement in my method

If there are not messages in the queue increase the delay to read again the queue in one second. For example, the service read the queue and there are not messages, the delay will be increased from 1 second to 2 seconds. It will be working in the same way until the delay was 60 seconds. In case of found a message the delay will be reset to 1 second and the validation will start again from 1 second to 60.

Basically I want to know if there are or not messages in the queue and inside of that validation increase the delay to read the queue.

This is the method I have created to read the messages from the queue.

private void createPollingStream() {
        Multi.createBy().repeating()
                .supplier(
                        () -> sqsMessagePoller.pollUDMUsages() //read messages from the queue. In case of not messages I want to increase the delay
                                .runSubscriptionOn(Infrastructure.getDefaultExecutor())
                                .onItem().transformToUniAndMerge(this::processMessagesQueue) //If the poll has messages we call to another method to process them.
                                .onFailure().invoke(failure ->
                                        Log.errorf("Error processing the messages: %s", failure))
                                .subscribe()
                                .with(succ -> Log.info("Current iteration of processing message complete"),
                                        failure -> Log.error("Failed to process message in flow", failure)
                                )
                ).withDelay(Duration.ofMillis(delay))//Initial delay of read messages from the queue. I guess it should be dynamic here.
                .indefinitely()
                .subscribe().with(x -> Log.info("Current iteration of processing message complete"),
                        failure -> Log.error("Failed to poll messages in flow", failure));
    }