Как вызвать функционал после того, как попытки повторной попытки RabbitMQ закончатся? (Spring Integration - RabbitMQ Listener) - PullRequest
Новая
       59

Как вызвать функционал после того, как попытки повторной попытки RabbitMQ закончатся? (Spring Integration - RabbitMQ Listener)

0 голосов
/ 27 июня 2018

Я хочу вызвать электронное письмо после того, как повторные попытки RabbitMQ Listener закончились, и все еще, если процесс обработки не удался.

логика повтора работает с кодом ниже. Но как активировать функциональность (триггер электронной почты), когда максимальное количество попыток повторной попытки закончено. 

@Bean
public SimpleMessageListenerContainer container() {
    SimpleMessageListenerContainer container =
            new SimpleMessageListenerContainer(connectionFactory());
    container.setQueues(myQueue());
    container.setDefaultRequeueRejected(false);
    Advice[] adviceArray = new Advice[]{interceptor()};
    container.setAdviceChain(adviceArray);
    return container;
}

@Bean
public IntegrationFlow inboundFlow() {
    return IntegrationFlows.from(
            Amqp.inboundAdapter(container()))
            .log()
            .handle(listenerBeanName, listenerMethodName)
            .get();
}

@Bean
RetryOperationsInterceptor interceptor() {
    return RetryInterceptorBuilder.stateless()
            .maxAttempts(retryMaxAttempts)
            .backOffOptions(initialInterval, multiplier, maxInterval)
             //.recoverer(new RejectAndDontRequeueRecoverer()) 
            .recoverer(new CustomRejectAndRecoverer())
            .build();
}

Добавление с кодом CustomeRecover 

@Service
public class CustomRejectAndRecoverer implements MessageRecoverer {
@Autowired
private EmailGateway emailgateway;

@Override
public void recover(Message message, Throwable cause) {
    // INSERT CODE HERE.... HOW TO CALL GATEWAY
    // emailgateway.sendMail(cause);
    throw new ListenerExecutionFailedException("Retry Policy Exhausted",
            new AmqpRejectAndDontRequeueException(cause), message);
} }

1 Ответ

0 голосов
/ 27 июня 2018

Это именно то, для чего .recoverer() в этом RetryInterceptorBuilder.

Теперь вы используете RejectAndDontRequeueRecoverer, но никто не мешает вам реализовать свой собственный MessageRecoverer с делегированием на RejectAndDontRequeueRecoverer и отправкой сообщения некоторым MessageChannel с логикой отправки электронных писем.

...