通用 AMQP 1.0 支持

BlockingQueue<Delivery> receivedDeliveries = new LinkedBlockingQueue<>();

@Bean
AmqpMessageListenerContainer protonDeliveryListenerContainer(AmqpConnectionFactory connectionFactory) {
    var amqpMessageListenerContainer = new AmqpMessageListenerContainer(connectionFactory);
    amqpMessageListenerContainer.setQueueNames(TEST_QUEUE_FOR_NATIVE_PROTON);
    amqpMessageListenerContainer.setAutoAccept(false);
    amqpMessageListenerContainer.setupMessageListener((ProtonDeliveryListener) this.receivedDeliveries::add);
    return amqpMessageListenerContainer;
}

@Bean
AmqpMessageListenerContainerFactory customContainerFactory(AmqpConnectionFactory connectionFactory) {
	var containerFactory = new MethodAmqpMessageListenerContainerFactory(connectionFactory);
	containerFactory.setAutoStartup(false);
	containerFactory.setAutoAccept(true);
	containerFactory.setConcurrency(5);
	containerFactory.setReceiveTimeout(Duration.ofSeconds(5));
	containerFactory.setGracefulShutdownPeriod(Duration.ofSeconds(36));
	containerFactory.setInitialCredits(55);
	containerFactory.setTaskExecutor(this.taskExecutor);
	containerFactory.setAdviceChain(new DebugInterceptor());
	containerFactory.setListenerErrorHandler(mock());
	containerFactory.setDefaultRequeueRejected(false);
	containerFactory.setHeaderMapper(this.headerMapper);
	containerFactory.setMessageConverter(this.messageConverter);
	return containerFactory;
}

@Bean
TaskExecutor taskExecutor() {
	ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
	threadPoolTaskExecutor.setCorePoolSize(10);
	return threadPoolTaskExecutor;
}

@Bean
AmqpHeaderMapper headerMapper() {
	return new SimpleAmqpHeaderMapper();
}

@Bean
DebugInterceptor debugInterceptor() {
	return new DebugInterceptor();
}

@AmqpListener(
        id = "monoListener",
        addresses = TEST_QUEUE2 + "," + TEST_QUEUE3,
        concurrency = "2",
        autoStartup = "#{true}",
        autoAccept = "true",
        containerFactory = "customContainerFactory",
        receiveTimeout = "${receive.timeout:22000}",
        gracefulShutdownPeriod = "pt2m",
        initialCredits = "101",
        defaultRequeueRejected = "true",
        executor = "#{taskExecutor}",
        messageConverter = "jsonMessageConverter",
        headerMapper = "headerMapper",
        adviceChain = "debugInterceptor")
Mono<DataOut> requestReplyWithJsonAndOtherAnnotationAttributes(DataIn dataIn) {
       DataOut dataOut = new DataOut(dataIn.data + "_out");
       return Mono.just(dataOut);
}

@Bean
fun jsonMessageConverter() = JacksonJsonMessageConverter()

@Bean
fun amqpConnectionFactory(protonClient: Client) =
    SingleAmqpConnectionFactory(protonClient)
        .setPort(amqpPort())

@Bean
fun amqpClient(connectionFactory: AmqpConnectionFactory, jsonMessageConverter: MessageConverter) =
    AmqpClient.builder(connectionFactory)
        .messageConverter(jsonMessageConverter)
        .build()

@Bean(AmqpDefaultConfiguration.DEFAULT_AMQP_LISTENER_CONTAINER_FACTORY_BEAN_NAME)
fun containerFactory(connectionFactory: AmqpConnectionFactory, jsonMessageConverter: MessageConverter) =
    MethodAmqpMessageListenerContainerFactory(connectionFactory)
        .also { it.setMessageConverter(jsonMessageConverter) }

@AmqpListener(addresses = ["TEST_QUEUE"])
@SendTo("TEST_REPLY_TO")
suspend fun requestReplyAsyncListener(payload: DataIn) = DataOut(payload.data + "_out")

data class DataIn(val data: String)

data class DataOut(val data: String)