spring amqp async nack

1// setAcknowledgeMode to MANUAL
2@Bean
3fun jsaFactory(
4    connectionFactory: ConnectionFactory,
5    configurer: SimpleRabbitListenerContainerFactoryConfigurer
6): SimpleRabbitListenerContainerFactory {
7    val factory = SimpleRabbitListenerContainerFactory()
8
9    factory.setAcknowledgeMode(AcknowledgeMode.MANUAL)
10    factory.setMessageConverter(jsonMessageConverter())
11
12    configurer.configure(factory, connectionFactory)
13
14    return factory
15}
16
17// throw AmqpRejectAndDontRequeueException to nack
18@RabbitListener(queues = ["salesforce.contact-courses.update"])
19fun update(message: CourseRequestDto): Mono<Void> {
20    message.email ?: return Mono.error<Void>(AmqpRejectAndDontRequeueException("Invalid message received"))
21
22    return courseDtoBuilder.getInstance(message)
23        .flatMap { contactRestClient.coursePatch(message.email, it) }
24        .doOnError { throw AmqpRejectAndDontRequeueException(it.message) } // NACK
25}