Spring/spring-amqp
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-1210: Add Kotlin suspend functions support (#2460)

Browse Source 
* GH-1210: Add Kotlin suspend functions support

Fixes https://github.com/spring-projects/spring-amqp/issues/1210

Kotlin Coroutines are essentially `Future` wrapping.
Therefore, it is natural to have `suspend` support on `@RabbitListener`
methods as we do now for `CompletableFuture` and `Mono`

* Introduce some utilities since we cannot reuse existing from Spring Messaging:
they are there about Kotlin Coroutines only for reactive handlers
* Some code clean up in the `RabbitListenerAnnotationBeanPostProcessor`
for the latest Java
* Add optional dep for `kotlinx-coroutines-reactor` and document the feature

* * Remove unused import
This commit is contained in:
Artem Bilan
2023-05-25 10:58:14 -04:00
committed by GitHub
parent 6b720bbd22
commit e452af157e
9 changed files with 289 additions and 129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
build.gradle
View File

@@ -56,6 +56,7 @@ ext {
jaywayJsonPathVersion = '2.7.0'
junit4Version = '4.13.2'
junitJupiterVersion = '5.9.2'
kotlinCoroutinesVersion = '1.6.4'
log4jVersion = '2.19.0'
logbackVersion = '1.4.4'
lz4Version = '1.8.0'
@@ -436,6 +437,7 @@ project('spring-rabbit') {
}
optionalApi "com.jayway.jsonpath:json-path:$jaywayJsonPathVersion"
optionalApi "org.apache.commons:commons-pool2:$commonsPoolVersion"
optionalApi "org.jetbrains.kotlinx:kotlinx-coroutines-reactor:$kotlinCoroutinesVersion"
testApi project(':spring-rabbit-junit')
testImplementation("com.willowtreeapps.assertk:assertk-jvm:$assertkVersion")

126
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/annotation/RabbitListenerAnnotationBeanPostProcessor.java
View File

@@ -18,8 +18,6 @@ package org.springframework.amqp.rabbit.annotation;
import java.lang.reflect.AnnotatedElement;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
@@ -30,7 +28,6 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@@ -56,6 +53,7 @@ import org.springframework.amqp.rabbit.listener.MultiMethodRabbitListenerEndpoin
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistry;
import org.springframework.amqp.rabbit.listener.adapter.AmqpMessageHandlerMethodFactory;
import org.springframework.amqp.rabbit.listener.adapter.ReplyPostProcessor;
import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler;
import org.springframework.amqp.support.converter.MessageConverter;
@@ -76,7 +74,6 @@ import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.expression.StandardBeanExpressionResolver;
import org.springframework.core.MethodParameter;
import org.springframework.core.Ordered;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.core.annotation.MergedAnnotations;
@@ -88,12 +85,9 @@ import org.springframework.core.env.Environment;
import org.springframework.core.task.TaskExecutor;
import org.springframework.format.support.DefaultFormattingConversionService;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.GenericMessageConverter;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.messaging.handler.annotation.support.MessageHandlerMethodFactory;
import org.springframework.messaging.handler.annotation.support.MethodArgumentNotValidException;
import org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver;
import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
import org.springframework.util.Assert;
@@ -101,8 +95,6 @@ import org.springframework.util.ClassUtils;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.validation.BindingResult;
import org.springframework.validation.ObjectError;
import org.springframework.validation.Validator;
/**
@@ -440,14 +432,10 @@ public class RabbitListenerAnnotationBeanPostProcessor
List<Object> resolvedQueues = resolveQueues(rabbitListener, declarables);
if (!resolvedQueues.isEmpty()) {
if (resolvedQueues.get(0) instanceof String) {
endpoint.setQueueNames(resolvedQueues.stream()
.map(o -> (String) o)
.collect(Collectors.toList()).toArray(new String[0]));
endpoint.setQueueNames(resolvedQueues.stream().map(o -> (String) o).toArray(String[]::new));
}
else {
endpoint.setQueues(resolvedQueues.stream()
.map(o -> (Queue) o)
.collect(Collectors.toList()).toArray(new Queue[0]));
endpoint.setQueues(resolvedQueues.stream().map(o -> (Queue) o).toArray(Queue[]::new));
}
}
endpoint.setConcurrency(resolveExpressionAsStringOrInteger(rabbitListener.concurrency(), "concurrency"));
@@ -667,12 +655,10 @@ public class RabbitListenerAnnotationBeanPostProcessor
String[] queues = rabbitListener.queues();
QueueBinding[] bindings = rabbitListener.bindings();
org.springframework.amqp.rabbit.annotation.Queue[] queuesToDeclare = rabbitListener.queuesToDeclare();
List<String> queueNames = new ArrayList<String>();
List<Queue> queueBeans = new ArrayList<Queue>();
if (queues.length > 0) {
for (int i = 0; i < queues.length; i++) {
resolveQueues(queues[i], queueNames, queueBeans);
}
List<String> queueNames = new ArrayList<>();
List<Queue> queueBeans = new ArrayList<>();
for (String queue : queues) {
resolveQueues(queue, queueNames, queueBeans);
}
if (!queueNames.isEmpty()) {
// revert to the previous behavior of just using the name when there is mixture of String and Queue
@@ -684,8 +670,8 @@ public class RabbitListenerAnnotationBeanPostProcessor
throw new BeanInitializationException(
"@RabbitListener can have only one of 'queues', 'queuesToDeclare', or 'bindings'");
}
for (int i = 0; i < queuesToDeclare.length; i++) {
queueNames.add(declareQueue(queuesToDeclare[i], declarables));
for (org.springframework.amqp.rabbit.annotation.Queue queue : queuesToDeclare) {
queueNames.add(declareQueue(queue, declarables));
}
}
if (bindings.length > 0) {
@@ -755,7 +741,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
declareExchangeAndBinding(binding, queueName, declarables);
}
}
return queues.toArray(new String[queues.size()]);
return queues.toArray(new String[0]);
}
private String declareQueue(org.springframework.amqp.rabbit.annotation.Queue bindingQueue,
@@ -862,7 +848,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
}
private Map<String, Object> resolveArguments(Argument[] arguments) {
Map<String, Object> map = new HashMap<String, Object>();
Map<String, Object> map = new HashMap<>();
for (Argument arg : arguments) {
String key = resolveExpressionAsString(arg.name(), "@Argument.name");
if (StringUtils.hasText(key)) {
@@ -1027,7 +1013,7 @@ public class RabbitListenerAnnotationBeanPostProcessor
}
private MessageHandlerMethodFactory createDefaultMessageHandlerMethodFactory() {
DefaultMessageHandlerMethodFactory defaultFactory = new DefaultMessageHandlerMethodFactory();
DefaultMessageHandlerMethodFactory defaultFactory = new AmqpMessageHandlerMethodFactory();
Validator validator = RabbitListenerAnnotationBeanPostProcessor.this.registrar.getValidator();
if (validator != null) {
defaultFactory.setValidator(validator);
@@ -1040,74 +1026,14 @@ public class RabbitListenerAnnotationBeanPostProcessor
List<HandlerMethodArgumentResolver> customArgumentsResolver = new ArrayList<>(
RabbitListenerAnnotationBeanPostProcessor.this.registrar.getCustomMethodArgumentResolvers());
defaultFactory.setCustomArgumentResolvers(customArgumentsResolver);
GenericMessageConverter messageConverter = new GenericMessageConverter(
this.defaultFormattingConversionService);
defaultFactory.setMessageConverter(messageConverter);
// Has to be at the end - look at PayloadMethodArgumentResolver documentation
customArgumentsResolver.add(new OptionalEmptyAwarePayloadArgumentResolver(messageConverter, validator));
defaultFactory.setMessageConverter(new GenericMessageConverter(this.defaultFormattingConversionService));
defaultFactory.afterPropertiesSet();
return defaultFactory;
}
}
private static class OptionalEmptyAwarePayloadArgumentResolver extends PayloadMethodArgumentResolver {
OptionalEmptyAwarePayloadArgumentResolver(
org.springframework.messaging.converter.MessageConverter messageConverter,
@Nullable Validator validator) {
super(messageConverter, validator);
}
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception { // NOSONAR
Object resolved = null;
try {
resolved = super.resolveArgument(parameter, message);
}
catch (MethodArgumentNotValidException ex) {
Type type = parameter.getGenericParameterType();
if (isOptional(message, type)) {
BindingResult bindingResult = ex.getBindingResult();
if (bindingResult != null) {
List<ObjectError> allErrors = bindingResult.getAllErrors();
if (allErrors.size() == 1) {
String defaultMessage = allErrors.get(0).getDefaultMessage();
if ("Payload value must not be empty".equals(defaultMessage)) {
return Optional.empty();
}
}
}
}
throw ex;
}
/*
* Replace Optional.empty() list elements with null.
*/
if (resolved instanceof List) {
List<?> list = ((List<?>) resolved);
for (int i = 0; i < list.size(); i++) {
if (list.get(i).equals(Optional.empty())) {
list.set(i, null);
}
}
}
return resolved;
}
private boolean isOptional(Message<?> message, Type type) {
return (Optional.class.equals(type) || (type instanceof ParameterizedType pType
&& Optional.class.equals(pType.getRawType())))
&& message.getPayload().equals(Optional.empty());
}
@Override
protected boolean isEmptyPayload(Object payload) {
return payload == null || payload.equals(Optional.empty());
}
}
/**
* The metadata holder of the class with {@link RabbitListener}
* and {@link RabbitHandler} annotations.
@@ -1147,28 +1073,14 @@ public class RabbitListenerAnnotationBeanPostProcessor
/**
* A method annotated with {@link RabbitListener}, together with the annotations.
*
* @param method the method with annotations
* @param annotations on the method
*/
private static class ListenerMethod {
final Method method; // NOSONAR
final RabbitListener[] annotations; // NOSONAR
ListenerMethod(Method method, RabbitListener[] annotations) { // NOSONAR
this.method = method;
this.annotations = annotations; // NOSONAR
}
private record ListenerMethod(Method method, RabbitListener[] annotations) {
}
private static class BytesToStringConverter implements Converter<byte[], String> {
private final Charset charset;
BytesToStringConverter(Charset charset) {
this.charset = charset;
}
private record BytesToStringConverter(Charset charset) implements Converter<byte[], String> {
@Override
public String convert(byte[] source) {

8
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AbstractAdaptableMessageListener.java
View File

@@ -364,7 +364,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe
* response message back.
* @param resultArg the result object to handle (never <code>null</code>)
* @param request the original request message
* @param channel the Rabbit channel to operate on (may be <code>null</code>)
* @param channel the Rabbit channel to operate on (maybe <code>null</code>)
* @param source the source data for the method invocation - e.g.
* {@code o.s.messaging.Message<?>}; may be null
* @see #buildMessage
@@ -391,8 +391,8 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe
}
else if (monoPresent && MonoHandler.isMono(resultArg.getReturnValue())) {
if (!this.isManualAck) {
this.logger.warn("Container AcknowledgeMode must be MANUAL for a Mono<?> return type; "
+ "otherwise the container will ack the message immediately");
this.logger.warn("Container AcknowledgeMode must be MANUAL for a Mono<?> return type" +
"(or Kotlin suspend function); otherwise the container will ack the message immediately");
}
MonoHandler.subscribe(resultArg.getReturnValue(),
r -> asyncSuccess(resultArg, request, channel, source, r),
@@ -448,7 +448,7 @@ public abstract class AbstractAdaptableMessageListener implements ChannelAwareMe
}
private void asyncFailure(Message request, Channel channel, Throwable t) {
this.logger.error("Future or Mono was completed with an exception for " + request, t);
this.logger.error("Future, Mono, or suspend function was completed with an exception for " + request, t);
try {
channel.basicNack(request.getMessageProperties().getDeliveryTag(), false,
ContainerUtils.shouldRequeue(this.defaultRequeueRejected, t, this.logger));

143
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/AmqpMessageHandlerMethodFactory.java Normal file
View File

@@ -0,0 +1,143 @@
/*
* Copyright 2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit.listener.adapter;
import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.List;
import java.util.Optional;
import org.springframework.core.KotlinDetector;
import org.springframework.core.MethodParameter;
import org.springframework.lang.Nullable;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
import org.springframework.messaging.handler.annotation.support.MethodArgumentNotValidException;
import org.springframework.messaging.handler.annotation.support.PayloadMethodArgumentResolver;
import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolverComposite;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
import org.springframework.validation.BindingResult;
import org.springframework.validation.ObjectError;
import org.springframework.validation.Validator;
/**
* Extension of the {@link DefaultMessageHandlerMethodFactory} for Spring AMQP requirements.
*
* @author Artem Bilan
*
* @since 3.0.5
*/
public class AmqpMessageHandlerMethodFactory extends DefaultMessageHandlerMethodFactory {
private final HandlerMethodArgumentResolverComposite argumentResolvers =
new HandlerMethodArgumentResolverComposite();
private MessageConverter messageConverter;
private Validator validator;
@Override
public void setMessageConverter(MessageConverter messageConverter) {
super.setMessageConverter(messageConverter);
this.messageConverter = messageConverter;
}
@Override
public void setValidator(Validator validator) {
super.setValidator(validator);
this.validator = validator;
}
@Override
protected List<HandlerMethodArgumentResolver> initArgumentResolvers() {
List<HandlerMethodArgumentResolver> resolvers = super.initArgumentResolvers();
if (KotlinDetector.isKotlinPresent()) {
// Insert before PayloadMethodArgumentResolver
resolvers.add(resolvers.size() - 1, new ContinuationHandlerMethodArgumentResolver());
}
// Has to be at the end, but before PayloadMethodArgumentResolver
resolvers.add(resolvers.size() - 1,
new OptionalEmptyAwarePayloadArgumentResolver(this.messageConverter, this.validator));
this.argumentResolvers.addResolvers(resolvers);
return resolvers;
}
@Override
public InvocableHandlerMethod createInvocableHandlerMethod(Object bean, Method method) {
InvocableHandlerMethod handlerMethod = new KotlinAwareInvocableHandlerMethod(bean, method);
handlerMethod.setMessageMethodArgumentResolvers(this.argumentResolvers);
return handlerMethod;
}
private static class OptionalEmptyAwarePayloadArgumentResolver extends PayloadMethodArgumentResolver {
OptionalEmptyAwarePayloadArgumentResolver(MessageConverter messageConverter, @Nullable Validator validator) {
super(messageConverter, validator);
}
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message) throws Exception { // NOSONAR
Object resolved;
try {
resolved = super.resolveArgument(parameter, message);
}
catch (MethodArgumentNotValidException ex) {
Type type = parameter.getGenericParameterType();
if (isOptional(message, type)) {
BindingResult bindingResult = ex.getBindingResult();
if (bindingResult != null) {
List<ObjectError> allErrors = bindingResult.getAllErrors();
if (allErrors.size() == 1) {
String defaultMessage = allErrors.get(0).getDefaultMessage();
if ("Payload value must not be empty".equals(defaultMessage)) {
return Optional.empty();
}
}
}
}
throw ex;
}
/*
* Replace Optional.empty() list elements with null.
*/
if (resolved instanceof List<?> list) {
for (int i = 0; i < list.size(); i++) {
if (list.get(i).equals(Optional.empty())) {
list.set(i, null);
}
}
}
return resolved;
}
private boolean isOptional(Message<?> message, Type type) {
return (Optional.class.equals(type) ||
(type instanceof ParameterizedType pType && Optional.class.equals(pType.getRawType())))
&& message.getPayload().equals(Optional.empty());
}
@Override
protected boolean isEmptyPayload(Object payload) {
return payload == null || payload.equals(Optional.empty());
}
}
}

50
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/ContinuationHandlerMethodArgumentResolver.java Normal file
View File

@@ -0,0 +1,50 @@
/*
* Copyright 2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit.listener.adapter;
import org.springframework.core.MethodParameter;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver;
import reactor.core.publisher.Mono;
/**
* No-op resolver for method arguments of type {@link kotlin.coroutines.Continuation}.
* <p>
* This class is similar to
* {@link org.springframework.messaging.handler.annotation.reactive.ContinuationHandlerMethodArgumentResolver}
* but for regular {@link HandlerMethodArgumentResolver} contract.
*
* @author Artem Bilan
*
* @since 3.0.5
*
* @see org.springframework.messaging.handler.annotation.reactive.ContinuationHandlerMethodArgumentResolver
*/
public class ContinuationHandlerMethodArgumentResolver implements HandlerMethodArgumentResolver {
@Override
public boolean supportsParameter(MethodParameter parameter) {
return "kotlin.coroutines.Continuation".equals(parameter.getParameterType().getName());
}
@Override
public Object resolveArgument(MethodParameter parameter, Message<?> message) {
return Mono.empty();
}
}

49
spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/KotlinAwareInvocableHandlerMethod.java Normal file
View File

@@ -0,0 +1,49 @@
/*
* Copyright 2023 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* https://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.amqp.rabbit.listener.adapter;
import java.lang.reflect.Method;
import org.springframework.core.CoroutinesUtils;
import org.springframework.core.KotlinDetector;
import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
/**
* An {@link InvocableHandlerMethod} extension for supporting Kotlin {@code suspend} function.
*
* @author Artem Bilan
*
* @since 3.0.5
*/
public class KotlinAwareInvocableHandlerMethod extends InvocableHandlerMethod {
public KotlinAwareInvocableHandlerMethod(Object bean, Method method) {
super(bean, method);
}
@Override
protected Object doInvoke(Object... args) throws Exception {
Method method = getBridgedMethod();
if (KotlinDetector.isSuspendingFunction(method)) {
return CoroutinesUtils.invokeSuspendingFunction(method, getBean(), args);
}
else {
return super.doInvoke(args);
}
}
}

26
spring-rabbit/src/test/kotlin/org/springframework/amqp/rabbit/annotation/EnableRabbitKotlinTests.kt
View File

@@ -20,6 +20,7 @@ import assertk.assertThat
import assertk.assertions.isEqualTo
import assertk.assertions.isTrue
import org.junit.jupiter.api.Test
import org.springframework.amqp.core.AcknowledgeMode
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory
import org.springframework.amqp.rabbit.core.RabbitTemplate
@@ -56,14 +57,14 @@ class EnableRabbitKotlinTests {
private lateinit var config: Config
@Test
fun `send and wait for consume` () {
fun `send and wait for consume`() {
val template = RabbitTemplate(this.config.cf())
template.convertAndSend("kotlinQueue", "test")
assertThat(this.config.latch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
fun `send and wait for consume with EH` () {
fun `send and wait for consume with EH`() {
val template = RabbitTemplate(this.config.cf())
template.convertAndSend("kotlinQueue1", "test")
assertThat(this.config.ehLatch.await(10, TimeUnit.SECONDS)).isTrue();
@@ -78,27 +79,22 @@ class EnableRabbitKotlinTests {
val latch = CountDownLatch(1)
@RabbitListener(queues = ["kotlinQueue"])
fun handle(@Suppress("UNUSED_PARAMETER") data: String) {
suspend fun handle(@Suppress("UNUSED_PARAMETER") data: String) {
this.latch.countDown()
}
@Bean
fun rabbitListenerContainerFactory(cf: CachingConnectionFactory): SimpleRabbitListenerContainerFactory {
val factory = SimpleRabbitListenerContainerFactory()
factory.setConnectionFactory(cf)
return factory
}
fun rabbitListenerContainerFactory(cf: CachingConnectionFactory) =
SimpleRabbitListenerContainerFactory().also {
it.setAcknowledgeMode(AcknowledgeMode.MANUAL)
it.setConnectionFactory(cf)
}
@Bean
fun cf(): CachingConnectionFactory {
return CachingConnectionFactory(
RabbitAvailableCondition.getBrokerRunning().connectionFactory)
}
fun cf() = CachingConnectionFactory(RabbitAvailableCondition.getBrokerRunning().connectionFactory)
@Bean
fun multi(): Multi {
return Multi()
}
fun multi() = Multi()
@Bean
fun proxyListenerPostProcessor(): BeanPostProcessor? {

6
src/reference/asciidoc/amqp.adoc
View File

@@ -3085,7 +3085,7 @@ Each queue needed a separate property.
====== Reply Management
The existing support in `MessageListenerAdapter` already lets your method have a non-void return type.
When that is the case, the result of the invocation is encapsulated in a message sent to the the address specified in the `ReplyToAddress` header of the original message, or to the default address configured on the listener.
When that is the case, the result of the invocation is encapsulated in a message sent to the address specified in the `ReplyToAddress` header of the original message, or to the default address configured on the listener.
You can set that default address by using the `@SendTo` annotation of the messaging abstraction.
Assuming our `processOrder` method should now return an `OrderStatus`, we can write it as follows to automatically send a reply:
@@ -3660,6 +3660,10 @@ If some exception occurs within the listener method that prevents creation of th
Starting with versions 2.2.21, 2.3.13, 2.4.1, the `AcknowledgeMode` will be automatically set the `MANUAL` when async return types are detected.
In addition, incoming messages with fatal exceptions will be negatively acknowledged individually, previously any prior unacknowledged message were also negatively acknowledged.
Starting with version 3.0.5, the `@RabbitListener` (and `@RabbitHandler`) methods can be marked with Kotlin `suspend` and the whole handling process and reply producing (optional) happens on respective Kotlin coroutine.
All the mentioned rules about `AcknowledgeMode.MANUAL` are still apply.
The `org.jetbrains.kotlinx:kotlinx-coroutines-reactor` dependency must be present in classpath to allow `suspend` function invocations.
[[threading]]
===== Threading and Asynchronous Consumers

8
src/reference/asciidoc/whats-new.adoc
View File

@@ -40,7 +40,7 @@ See <<stream-support>> for more information.
Batch listeners can now consume `Collection<?>` as well as `List<?>`.
The batch messaging adapter now ensures that the method is suitable for consuming batches.
When setting the container factory `consumerBatchEnabled` to `true`, the `batchListener` property is also set to `true`.
See <<receiving-batch>> for more infoprmation.
See <<receiving-batch>> for more information.
`MessageConverter` s can now return `Optional.empty()` for a null value; this is currently implemented by the `Jackson2JsonMessageConverter`.
See <<Jackson2JsonMessageConverter-from-message>> for more information
@@ -48,9 +48,13 @@ See <<Jackson2JsonMessageConverter-from-message>> for more information
You can now configure a `ReplyPostProcessor` via the container factory rather than via a property on `@RabbitListener`.
See <<async-annotation-driven-reply>> for more information.
The `@RabbitListener` (and `@RabbitHandler`) methods can now be as a Kotlin `suspend` functions.
See <<async-returns>> for more information.
==== Connection Factory Changes
The default `addressShuffleMode` in `AbstractConnectionFactory` is now `RANDOM`. This results in connecting to a random host when multiple addresses are provided.
The default `addressShuffleMode` in `AbstractConnectionFactory` is now `RANDOM`.
This results in connecting to a random host when multiple addresses are provided.
See <<cluster>> for more information.
The `LocalizedQueueConnectionFactory` no longer uses the RabbitMQ `http-client` library to determine which node is the leader for a queue.