Wait for RabbitTemplate Correlated confirms

Question

I have this function in RabbitMQ class that is used to publish messages:

    @Retryable(value = { AmqpIOException.class, AmqpTimeoutException.class, AmqpConnectException.class, AmqpReplyTimeoutException.class },
            maxAttempts = 3,
            backoff = @Backoff(delay = 1000))
    public void publish(Message msg, SuccessCallback<? super CorrelationData.Confirm> successCallback, FailureCallback failureCallback, String customRoutingKey) {
        CorrelationData data = new CorrelationData();
        data.getFuture().addCallback(successCallback, failureCallback);
        rabbitTemplate.convertAndSend(exchange.getName(), customRoutingKey, msg, data);
    }

And here is how I use this function above:

public void sendMessageToRMQ(MyEntity msgObj, String routingKey) throws ServiceUnavailableException {

        try {
            rmqPublisher.publish(
                    MessageBuilder
                            .withBody(RMQPayloadBuilder.buildPayload(msgObj))
                            .build(),
                    confirmation -> {
                        if (confirmation != null && confirmation.isAck()) {
                            msgObj.setSentDate(OffsetDateTime.now(ZoneOffset.UTC));

                            repository.save(msgObj);
                        } else {
                            StructuredLogger.logErrorMessage("Message un-ack", logValues);
                        }
                    },
                    failure -> StructuredLogger.logErrorMessage("un able to send", logValues),
                    routingKey
            );
        } catch (AmqpException amqpException) {
            StructuredLogger.logErrorMessage("Unable to publish message", logValues);
        }
    }

The the above function is used in a for loop to publish messages for huge number of msgObjects.

What I am trying to achieve is something like this:

// msgObjArray is a chunk of the whole msgObjArray

for (MyEntity msgObj : msgObjArrayChunk) {
            try {
                
                this.service.sendMessageToRMQ(msgObj, this.routingKey);


            } catch (ServiceUnavailableException e) {

                StructuredLogger.logErrorMessage("Failed to send ");
            }
        }

// What Should I write here to wait for the above chunk of msgObj to be acked and finished processing

Answer 1

You need to return the CorrelationData from publish and collect them into a list; then, after the sends are complete, iterate over them and use data.getFuture().get( <some timeout> ) to wait for each confirm.