diff --git a/build.gradle b/build.gradle index 08b4492b..78323311 100644 --- a/build.gradle +++ b/build.gradle @@ -55,6 +55,8 @@ ext { logbackVersion = '1.2.3' micrometerVersion = '1.7.0' mockitoVersion = '3.9.0' + protonJVersion = '0.33.8' + rabbitmqStreamVersion = '0.1.0-SNAPSHOT' rabbitmqVersion = project.hasProperty('rabbitmqVersion') ? project.rabbitmqVersion : '5.12.0' rabbitmqHttpClientVersion = '3.9.0.RELEASE' reactorVersion = '2020.0.7' @@ -96,6 +98,7 @@ allprojects { maven { url 'https://repo.spring.io/libs-milestone' } if (version.endsWith('-SNAPSHOT')) { maven { url 'https://repo.spring.io/libs-snapshot' } + maven { url 'https://oss.sonatype.org/content/repositories/snapshots' } } // maven { url 'https://repo.spring.io/libs-staging-local' } } @@ -385,6 +388,28 @@ project('spring-rabbit') { } +project('spring-rabbit-stream') { + description = 'Spring RabbitMQ Stream Support' + + dependencies { + + api project(':spring-rabbit') + api "com.rabbitmq:stream-client:$rabbitmqStreamVersion" + optionalApi "com.rabbitmq:http-client:$rabbitmqHttpClientVersion" + + testApi project(':spring-rabbit-junit') + testRuntimeOnly 'com.fasterxml.jackson.core:jackson-core' + testRuntimeOnly 'com.fasterxml.jackson.core:jackson-databind' + testRuntimeOnly 'com.fasterxml.jackson.dataformat:jackson-dataformat-xml' + testRuntimeOnly 'com.fasterxml.jackson.module:jackson-module-kotlin' + testRuntimeOnly "org.apache.httpcomponents:httpclient:$commonsHttpClientVersion" + testRuntimeOnly "org.apache.qpid:proton-j:$protonJVersion" + testImplementation "org.testcontainers:rabbitmq:1.15.3" + testImplementation "org.apache.logging.log4j:log4j-slf4j-impl:$log4jVersion" + } + +} + project('spring-rabbit-junit') { description = 'Spring Rabbit JUnit Support' diff --git a/settings.gradle b/settings.gradle index a0f0f963..5d950cd7 100644 --- a/settings.gradle +++ b/settings.gradle @@ -2,5 +2,6 @@ rootProject.name = 'spring-amqp-dist' include 'spring-amqp' include 'spring-rabbit' +include 'spring-rabbit-stream' include 'spring-rabbit-junit' include 'spring-rabbit-test' diff --git a/spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java b/spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java index 932eb93d..f76bbdb9 100644 --- a/spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java +++ b/spring-amqp/src/main/java/org/springframework/amqp/core/QueueBuilder.java @@ -267,6 +267,15 @@ public final class QueueBuilder extends AbstractBuilder { return withArgument("x-queue-type", "quorum"); } + /** + * Set the queue argument to declare a queue of type 'stream' instead of 'classic'. + * @return the builder. + * @since 2.4 + */ + public QueueBuilder stream() { + return withArgument("x-queue-type", "stream"); + } + /** * Set the delivery limit; only applies to quorum queues. * @param limit the limit. diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/StreamRabbitListenerContainerFactory.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/StreamRabbitListenerContainerFactory.java new file mode 100644 index 00000000..411a24be --- /dev/null +++ b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/config/StreamRabbitListenerContainerFactory.java @@ -0,0 +1,118 @@ +/* + * Copyright 2021 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.rabbit.stream.config; + +import java.lang.reflect.Method; +import java.util.function.Consumer; + +import org.springframework.amqp.rabbit.batch.BatchingStrategy; +import org.springframework.amqp.rabbit.config.BaseRabbitListenerContainerFactory; +import org.springframework.amqp.rabbit.config.ContainerCustomizer; +import org.springframework.amqp.rabbit.listener.MethodRabbitListenerEndpoint; +import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint; +import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler; +import org.springframework.lang.Nullable; +import org.springframework.rabbit.stream.listener.StreamListenerContainer; +import org.springframework.rabbit.stream.listener.adapter.StreamMessageListenerAdapter; +import org.springframework.util.Assert; + +import com.rabbitmq.stream.ConsumerBuilder; +import com.rabbitmq.stream.Environment; + +/** + * Factory for StreamListenerContainer. + * + * @author Gary Russell + * @since 2.4 + * + */ +public class StreamRabbitListenerContainerFactory + extends BaseRabbitListenerContainerFactory { + + private final Environment environment; + + private boolean nativeListener; + + private Consumer consumerCustomizer; + + private ContainerCustomizer containerCustomizer; + + /** + * Construct an instance using the provided environment. + * @param environment the environment. + */ + public StreamRabbitListenerContainerFactory(Environment environment) { + Assert.notNull(environment, "'environment' cannot be null"); + this.environment = environment; + } + + /** + * Set to true to create a container supporting a native RabbitMQ Stream message. + * @param nativeListener true for native listeners. + */ + public void setNativeListener(boolean nativeListener) { + this.nativeListener = nativeListener; + } + + /** + * Customize the consumer builder before it is built. + * @param consumerCustomizer the customizer. + */ + public void setConsumerCustomizer(java.util.function.Consumer consumerCustomizer) { + this.consumerCustomizer = consumerCustomizer; + } + + /** + * Set a {@link ContainerCustomizer} that is invoked after a container is created and + * configured to enable further customization of the container. + * @param containerCustomizer the customizer. + */ + public void setContainerCustomizer(ContainerCustomizer containerCustomizer) { + this.containerCustomizer = containerCustomizer; + } + + @Override + public StreamListenerContainer createListenerContainer(RabbitListenerEndpoint endpoint) { + if (endpoint instanceof MethodRabbitListenerEndpoint && this.nativeListener) { + ((MethodRabbitListenerEndpoint) endpoint).setAdapterProvider( + (boolean batch, Object bean, Method method, boolean returnExceptions, + RabbitListenerErrorHandler errorHandler, @Nullable BatchingStrategy batchingStrategy) -> { + + Assert.isTrue(!batch, "Batch listeners are not supported by the stream container"); + return new StreamMessageListenerAdapter(bean, method, returnExceptions, errorHandler); + }); + } + StreamListenerContainer container = createContainerInstance(); + if (this.consumerCustomizer != null) { + container.setConsumerCustomizer(this.consumerCustomizer); + } + applyCommonOverrides(endpoint, container); + if (this.containerCustomizer != null) { + this.containerCustomizer.configure(container); + } + return container; + } + + /** + * Create an instance of the listener container. + * @return the container. + */ + protected StreamListenerContainer createContainerInstance() { + return new StreamListenerContainer(this.environment); + } + +} diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java new file mode 100644 index 00000000..fa9db0f2 --- /dev/null +++ b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamListenerContainer.java @@ -0,0 +1,201 @@ +/* + * Copyright 2021 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.rabbit.stream.listener; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageListener; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; +import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.lang.Nullable; +import org.springframework.rabbit.stream.support.converter.DefaultStreamMessageConverter; +import org.springframework.rabbit.stream.support.converter.StreamMessageConverter; +import org.springframework.util.Assert; + +import com.rabbitmq.stream.Consumer; +import com.rabbitmq.stream.ConsumerBuilder; +import com.rabbitmq.stream.Environment; + +/** + * A listener container for RabbitMQ Streams. + * + * @author Gary Russell + * @since 2.4 + * + */ +public class StreamListenerContainer implements MessageListenerContainer, BeanNameAware { + + protected Log logger = LogFactory.getLog(getClass()); + + private final Environment environment; + + private final ConsumerBuilder builder; + + private StreamMessageConverter messageConverter; + + private java.util.function.Consumer consumerCustomizer = c -> { }; + + private String stream; + + private Consumer consumer; + + private String listenerId; + + private String beanName; + + private boolean autoStartup = true; + + private MessageListener messageListener; + + /** + * Construct an instance using the provided environment. + * @param environment the environment. + */ + public StreamListenerContainer(Environment environment) { + Assert.notNull(environment, "'environment' cannot be null"); + this.environment = environment; + this.builder = environment.consumerBuilder(); + this.messageConverter = new DefaultStreamMessageConverter(environment); + } + + @Override + public void setQueueNames(String... queueNames) { + Assert.isTrue(queueNames != null && queueNames.length == 1, "Only one stream is supported"); + this.stream = queueNames[0]; + this.builder.stream(this.stream); + } + + /** + * Get a {@link StreamMessageConverter} used to convert a + * {@link com.rabbitmq.stream.Message} to a + * {@link org.springframework.amqp.core.Message}. + * @return the converter. + */ + public StreamMessageConverter getMessageConverter() { + return this.messageConverter; + } + + /** + * Set a {@link StreamMessageConverter} used to convert a + * {@link com.rabbitmq.stream.Message} to a + * {@link org.springframework.amqp.core.Message}. + * @param messageConverter the converter. + */ + public void setMessageConverter(StreamMessageConverter messageConverter) { + this.messageConverter = messageConverter; + } + + /** + * Customize the consumer builder before it is built. + * @param consumerCustomizer the customizer. + */ + public void setConsumerCustomizer(java.util.function.Consumer consumerCustomizer) { + this.consumerCustomizer = consumerCustomizer; + } + + /** + * The 'id' attribute of the listener. + * @return the id (or the container bean name if no id set). + */ + @Nullable + public String getListenerId() { + return this.listenerId != null ? this.listenerId : this.beanName; + } + + @Override + public void setListenerId(String listenerId) { + this.listenerId = listenerId; + } + + /** + * Return the bean name. + * @return the bean name. + */ + @Nullable + public String getBeanName() { + return this.beanName; + } + + @Override + public void setBeanName(String beanName) { + this.beanName = beanName; + } + + @Override + public void setAutoStartup(boolean autoStart) { + this.autoStartup = autoStart; + } + + @Override + public boolean isAutoStartup() { + return this.autoStartup; + } + @Override + @Nullable + public Object getMessageListener() { + return this.messageListener; + } + + @Override + public synchronized boolean isRunning() { + return this.consumer != null; + } + + @Override + public synchronized void start() { + if (this.consumer == null) { + this.consumerCustomizer.accept(this.builder); + this.consumer = this.builder.build(); + } + } + + @Override + public synchronized void stop() { + if (this.consumer != null) { + this.consumer.close(); + this.consumer = null; + } + } + + @Override + public void setupMessageListener(MessageListener messageListener) { + this.messageListener = messageListener; + this.builder.messageHandler((context, message) -> { + if (messageListener instanceof StreamMessageListener) { + ((StreamMessageListener) messageListener).onStreamMessage(message, context); + } + else { + Message message2 = this.messageConverter.toMessage(message, new StreamMessageProperties(context)); + if (messageListener instanceof ChannelAwareMessageListener) { + try { + ((ChannelAwareMessageListener) messageListener).onMessage(message2, null); + } + catch (Exception e) { // NOSONAR + this.logger.error("Listner threw an exception", e); + } + } + else { + messageListener.onMessage(message2); + } + } + }); + } + +} diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamMessageListener.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamMessageListener.java new file mode 100644 index 00000000..679f6525 --- /dev/null +++ b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamMessageListener.java @@ -0,0 +1,40 @@ +/* + * Copyright 2021 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.rabbit.stream.listener; + +import org.springframework.amqp.core.MessageListener; + +import com.rabbitmq.stream.Message; +import com.rabbitmq.stream.MessageHandler.Context; + +/** + * A message listener that receives native stream messages. + * + * @author Gary Russell + * @since 2.4 + * + */ +public interface StreamMessageListener extends MessageListener { + + /** + * Process a message. + * @param message the message. + * @param context the stream context. + */ + void onStreamMessage(Message message, Context context); + +} diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamMessageProperties.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamMessageProperties.java new file mode 100644 index 00000000..5eb724d2 --- /dev/null +++ b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/StreamMessageProperties.java @@ -0,0 +1,53 @@ +/* + * Copyright 2021 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.rabbit.stream.listener; + +import org.springframework.amqp.core.MessageProperties; + +import com.rabbitmq.stream.MessageHandler.Context; + +/** + * {@link MessageProperties} extension for stream messages. + * + * @author Gary Russell + * @since 2.4 + * + */ +public class StreamMessageProperties extends MessageProperties { + + private static final long serialVersionUID = 1L; + + private final Context context; + + /** + * Create a new instance with the provided context. + * @param context the context. + */ + public StreamMessageProperties(Context context) { + this.context = context; + } + + /** + * Return the stream {@link Context} for the message. + * @return the context. + */ + public Context getContext() { + return this.context; + } + + +} diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/adapter/StreamMessageListenerAdapter.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/adapter/StreamMessageListenerAdapter.java new file mode 100644 index 00000000..3db6b3a3 --- /dev/null +++ b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/listener/adapter/StreamMessageListenerAdapter.java @@ -0,0 +1,67 @@ +/* + * Copyright 2021 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.rabbit.stream.listener.adapter; + +import java.lang.reflect.Method; + +import org.springframework.amqp.rabbit.listener.adapter.InvocationResult; +import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter; +import org.springframework.amqp.rabbit.listener.api.RabbitListenerErrorHandler; +import org.springframework.rabbit.stream.listener.StreamMessageListener; + +import com.rabbitmq.stream.Message; +import com.rabbitmq.stream.MessageHandler.Context; + +/** + * A listener adapter that receives native stream messages. + * + * @author Gary Russell + * @since 2.4 + * + */ +public class StreamMessageListenerAdapter extends MessagingMessageListenerAdapter implements StreamMessageListener { + + /** + * Construct an instance with the provided arguments. + * @param bean the bean. + * @param method the method. + * @param returnExceptions true to return exceptions. + * @param errorHandler the error handler. + */ + public StreamMessageListenerAdapter(Object bean, Method method, boolean returnExceptions, + RabbitListenerErrorHandler errorHandler) { + + super(bean, method, returnExceptions, errorHandler); + } + + @Override + public void onStreamMessage(Message message, Context context) { + try { + InvocationResult result = getHandlerAdapter().invoke(null, message, context); + if (result.getReturnValue() != null) { + logger.warn("Replies are not currently supported with native Stream listeners"); + } + else { + logger.trace("No result object given - no result to handle"); + } + } + catch (Exception ex) { + this.logger.error("Failed to invoke listener", ex); + } + } + +} diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/converter/DefaultStreamMessageConverter.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/converter/DefaultStreamMessageConverter.java new file mode 100644 index 00000000..f5f7d411 --- /dev/null +++ b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/converter/DefaultStreamMessageConverter.java @@ -0,0 +1,85 @@ +/* + * Copyright 2021 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.rabbit.stream.support.converter; + +import java.util.Map; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageBuilder; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.support.converter.MessageConversionException; +import org.springframework.amqp.utils.JavaUtils; +import org.springframework.util.Assert; + +import com.rabbitmq.stream.Environment; +import com.rabbitmq.stream.MessageBuilder.PropertiesBuilder; +import com.rabbitmq.stream.Properties; +import com.rabbitmq.stream.codec.WrapperMessageBuilder; + +/** + * Default {@link StreamMessageConverter}. + * + * @author Gary Russell + * @since 2.4 + * + */ +public class DefaultStreamMessageConverter implements StreamMessageConverter { + + private final Environment environment; + + /** + * Construct an instance using the provided environment. + * @param environment the environment. + */ + public DefaultStreamMessageConverter(Environment environment) { + Assert.notNull(environment, "'environment' cannot be null"); + this.environment = environment; + } + + @Override + public Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException { + Assert.isInstanceOf(com.rabbitmq.stream.Message.class, object); + com.rabbitmq.stream.Message streamMessage = (com.rabbitmq.stream.Message) object; + toMessageProperties(streamMessage, messageProperties); + return MessageBuilder.withBody(streamMessage.getBodyAsBinary()).andProperties(messageProperties).build(); + } + + @Override + public com.rabbitmq.stream.Message fromMessage(Message message) throws MessageConversionException { + // TODO get the builder from the environment's codec + WrapperMessageBuilder builder = new WrapperMessageBuilder(); + PropertiesBuilder propsBuilder = builder.properties(); + MessageProperties mProps = message.getMessageProperties(); + JavaUtils.INSTANCE + .acceptIfNotNull(mProps.getMessageId(), propsBuilder::messageId); + // TODO ... + builder.addData(message.getBody()); + return builder.build(); + } + + private void toMessageProperties(com.rabbitmq.stream.Message streamMessage, MessageProperties messageProperties) { + Properties properties = streamMessage.getProperties(); + JavaUtils.INSTANCE + .acceptIfNotNull(properties.getMessageIdAsString(), messageProperties::setMessageId); + // TODO ... + Map applicationProperties = streamMessage.getApplicationProperties(); + if (applicationProperties != null) { + messageProperties.getHeaders().putAll(applicationProperties); + } + } + +} diff --git a/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/converter/StreamMessageConverter.java b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/converter/StreamMessageConverter.java new file mode 100644 index 00000000..92ca6243 --- /dev/null +++ b/spring-rabbit-stream/src/main/java/org/springframework/rabbit/stream/support/converter/StreamMessageConverter.java @@ -0,0 +1,40 @@ +/* + * Copyright 2021 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.rabbit.stream.support.converter; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.support.converter.MessageConversionException; +import org.springframework.amqp.support.converter.MessageConverter; + +/** + * Converts between {@link com.rabbitmq.stream.Message} and + * {@link org.springframework.amqp.core.Message}. + * + * @author Gary Russell + * @since 2.4 + * + */ +public interface StreamMessageConverter extends MessageConverter { + + @Override + Message toMessage(Object object, MessageProperties messageProperties) throws MessageConversionException; + + @Override + com.rabbitmq.stream.Message fromMessage(Message message) throws MessageConversionException; + +} diff --git a/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/AbstractIntegrationTests.java b/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/AbstractIntegrationTests.java new file mode 100644 index 00000000..cb4e1b82 --- /dev/null +++ b/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/AbstractIntegrationTests.java @@ -0,0 +1,40 @@ +/* + * Copyright 2021 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.rabbit.stream.listener; + +import java.time.Duration; + +import org.testcontainers.containers.GenericContainer; +import org.testcontainers.utility.DockerImageName; + +/** + * @author Gary Russell + * @since 2.4 + * + */ +public abstract class AbstractIntegrationTests { + + static final GenericContainer RABBITMQ = new GenericContainer<>( + DockerImageName.parse("pivotalrabbitmq/rabbitmq-stream")) + .withExposedPorts(5672, 15672, 5552) + .withStartupTimeout(Duration.ofMinutes(2)); + + static { + RABBITMQ.start(); + } + +} diff --git a/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java b/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java new file mode 100644 index 00000000..b33e0bb0 --- /dev/null +++ b/spring-rabbit-stream/src/test/java/org/springframework/rabbit/stream/listener/RabbitListenerTests.java @@ -0,0 +1,181 @@ +/* + * Copyright 2021 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.rabbit.stream.listener; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.core.Queue; +import org.springframework.amqp.core.QueueBuilder; +import org.springframework.amqp.rabbit.annotation.EnableRabbit; +import org.springframework.amqp.rabbit.annotation.RabbitListener; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.SmartLifecycle; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.rabbit.stream.config.StreamRabbitListenerContainerFactory; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import com.rabbitmq.http.client.Client; +import com.rabbitmq.http.client.domain.QueueInfo; +import com.rabbitmq.stream.Address; +import com.rabbitmq.stream.Environment; +import com.rabbitmq.stream.Message; +import com.rabbitmq.stream.MessageHandler.Context; +import com.rabbitmq.stream.OffsetSpecification; + +/** + * @author Gary Russell + * @since 2.4 + * + */ +@SpringJUnitConfig +@DirtiesContext +public class RabbitListenerTests extends AbstractIntegrationTests { + + @Autowired + Config config; + + @Test + void simple(@Autowired RabbitTemplate template) throws InterruptedException { + + template.convertAndSend("test.stream.queue1", "foo"); + assertThat(this.config.latch1.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.received).isEqualTo("foo"); + } + + @Test + void nativeMsg(@Autowired RabbitTemplate template) throws InterruptedException { + + template.convertAndSend("test.stream.queue2", "foo"); + assertThat(this.config.latch2.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(this.config.receivedNative).isNotNull(); + assertThat(this.config.context).isNotNull(); + } + + @Test + void queueOverAmqp() throws Exception { + Client client = new Client("http://guest:guest@localhost:" + RABBITMQ.getMappedPort(15672) + "/api"); + QueueInfo queue = client.getQueue("/", "stream.created.over.amqp"); + assertThat(queue.getArguments().get("x-queue-type")).isEqualTo("stream"); + } + + @Configuration(proxyBeanMethods = false) + @EnableRabbit + public static class Config { + + final CountDownLatch latch1 = new CountDownLatch(1); + + final CountDownLatch latch2 = new CountDownLatch(1); + + volatile String received; + + volatile Message receivedNative; + + volatile Context context; + + @Bean + Environment environment() { + return Environment.builder() + .addressResolver(add -> new Address("localhost", RABBITMQ.getMappedPort(5552))) + .build(); + } + + @Bean + SmartLifecycle creator(Environment env) { + return new SmartLifecycle() { + + @Override + public void stop() { + } + + @Override + public void start() { + env.streamCreator().stream("test.stream.queue1").create(); + env.streamCreator().stream("test.stream.queue2").create(); + } + + @Override + public boolean isRunning() { + return false; + } + }; + } + + @Bean + RabbitListenerContainerFactory rabbitListenerContainerFactory(Environment env) { + return new StreamRabbitListenerContainerFactory(env); + } + + @RabbitListener(queues = "test.stream.queue1") + void listen(String in) { + this.received = in; + this.latch1.countDown(); + } + + @Bean + RabbitListenerContainerFactory nativeFactory(Environment env) { + StreamRabbitListenerContainerFactory factory = new StreamRabbitListenerContainerFactory(env); + factory.setNativeListener(true); + factory.setConsumerCustomizer(builder -> builder.name("myConsumer") + .offset(OffsetSpecification.first()) + .manualCommitStrategy()); + return factory; + } + + @RabbitListener(queues = "test.stream.queue2", containerFactory = "nativeFactory") + void nativeMsg(Message in, Context context) { + this.receivedNative = in; + this.context = context; + this.latch2.countDown(); + context.commit(); + } + + @Bean + CachingConnectionFactory cf() { + return new CachingConnectionFactory(RABBITMQ.getContainerIpAddress(), RABBITMQ.getFirstMappedPort()); + } + + @Bean + RabbitTemplate template(CachingConnectionFactory cf) { + return new RabbitTemplate(cf); + } + + @Bean + RabbitAdmin admin(CachingConnectionFactory cf) { + return new RabbitAdmin(cf); + } + + @Bean + Queue queue() { + return QueueBuilder.durable("stream.created.over.amqp") + .stream() + .build(); + } + + } + +} diff --git a/spring-rabbit-stream/src/test/resources/log4j2-test.xml b/spring-rabbit-stream/src/test/resources/log4j2-test.xml new file mode 100644 index 00000000..86ded2e3 --- /dev/null +++ b/spring-rabbit-stream/src/test/resources/log4j2-test.xml @@ -0,0 +1,15 @@ + + + + + + + + + + + + + + + diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java index ce3b05e4..d9cbdcfb 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/AbstractRabbitListenerContainerFactory.java @@ -32,7 +32,6 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory; import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint; -import org.springframework.amqp.rabbit.listener.adapter.AbstractAdaptableMessageListener; import org.springframework.amqp.support.ConsumerTagStrategy; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.amqp.utils.JavaUtils; @@ -42,8 +41,6 @@ import org.springframework.context.ApplicationContextAware; import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.lang.Nullable; -import org.springframework.retry.RecoveryCallback; -import org.springframework.retry.support.RetryTemplate; import org.springframework.transaction.PlatformTransactionManager; import org.springframework.util.Assert; import org.springframework.util.ErrorHandler; @@ -64,7 +61,8 @@ import org.springframework.util.backoff.FixedBackOff; * @see AbstractMessageListenerContainer */ public abstract class AbstractRabbitListenerContainerFactory - implements RabbitListenerContainerFactory, ApplicationContextAware, ApplicationEventPublisherAware { + extends BaseRabbitListenerContainerFactory + implements ApplicationContextAware, ApplicationEventPublisherAware { protected final Log logger = LogFactory.getLog(getClass()); // NOSONAR @@ -88,8 +86,6 @@ public abstract class AbstractRabbitListenerContainerFactory recoveryCallback; - private ContainerCustomizer containerCustomizer; private boolean batchListener; @@ -192,14 +182,6 @@ public abstract class AbstractRabbitListenerContainerFactory recoveryCallback) { - this.recoveryCallback = recoveryCallback; - } - /** * Set a {@link ContainerCustomizer} that is invoked after a container is created and * configured to enable further customization of the container. @@ -437,29 +381,14 @@ public abstract class AbstractRabbitListenerContainerFactory the container type that the factory creates. + * + * @author Gary Russell + * @since 2.4 + * + */ +public abstract class BaseRabbitListenerContainerFactory + implements RabbitListenerContainerFactory { + + protected Boolean defaultRequeueRejected; + + private MessagePostProcessor[] beforeSendReplyPostProcessors; + + private RetryTemplate retryTemplate; + + private RecoveryCallback recoveryCallback; + + @Override + public abstract C createListenerContainer(RabbitListenerEndpoint endpoint); + + /** + * @param requeueRejected true to reject by default. + * @see AbstractMessageListenerContainer#setDefaultRequeueRejected + */ + public void setDefaultRequeueRejected(Boolean requeueRejected) { + this.defaultRequeueRejected = requeueRejected; + } + + /** + * Set post processors that will be applied before sending replies; added to each + * message listener adapter. + * @param postProcessors the post processors. + * @since 2.0.3 + * @see AbstractAdaptableMessageListener#setBeforeSendReplyPostProcessors(MessagePostProcessor...) + */ + public void setBeforeSendReplyPostProcessors(MessagePostProcessor... postProcessors) { + Assert.notNull(postProcessors, "'postProcessors' cannot be null"); + Assert.noNullElements(postProcessors, "'postProcessors' cannot have null elements"); + this.beforeSendReplyPostProcessors = Arrays.copyOf(postProcessors, postProcessors.length); + } + + /** + * Set a {@link RetryTemplate} to use when sending replies; added to each message + * listener adapter. + * @param retryTemplate the template. + * @since 2.0.6 + * @see #setReplyRecoveryCallback(RecoveryCallback) + * @see AbstractAdaptableMessageListener#setRetryTemplate(RetryTemplate) + */ + public void setRetryTemplate(RetryTemplate retryTemplate) { + this.retryTemplate = retryTemplate; + } + + /** + * Set a {@link RecoveryCallback} to invoke when retries are exhausted. Added to each + * message listener adapter. Only used if a {@link #setRetryTemplate(RetryTemplate) + * retryTemplate} is provided. + * @param recoveryCallback the recovery callback. + * @since 2.0.6 + * @see #setRetryTemplate(RetryTemplate) + * @see AbstractAdaptableMessageListener#setRecoveryCallback(RecoveryCallback) + */ + public void setReplyRecoveryCallback(RecoveryCallback recoveryCallback) { + this.recoveryCallback = recoveryCallback; + } + + protected void applyCommonOverrides(@Nullable RabbitListenerEndpoint endpoint, C instance) { + if (endpoint != null) { // endpoint settings overriding default factory settings + JavaUtils.INSTANCE + .acceptIfNotNull(endpoint.getAutoStartup(), instance::setAutoStartup); + instance.setListenerId(endpoint.getId()); + endpoint.setupListenerContainer(instance); + } + if (instance.getMessageListener() instanceof AbstractAdaptableMessageListener) { + AbstractAdaptableMessageListener messageListener = (AbstractAdaptableMessageListener) instance + .getMessageListener(); + JavaUtils.INSTANCE + .acceptIfNotNull(this.beforeSendReplyPostProcessors, + messageListener::setBeforeSendReplyPostProcessors) + .acceptIfNotNull(this.retryTemplate, messageListener::setRetryTemplate) + .acceptIfCondition(this.retryTemplate != null && this.recoveryCallback != null, + this.recoveryCallback, messageListener::setRecoveryCallback) + .acceptIfNotNull(this.defaultRequeueRejected, messageListener::setDefaultRequeueRejected) + .acceptIfNotNull(endpoint.getReplyPostProcessor(), messageListener::setReplyPostProcessor) + .acceptIfNotNull(endpoint.getReplyContentType(), messageListener::setReplyContentType); + messageListener.setConverterWinsContentType(endpoint.isConverterWinsContentType()); + } + } + +} diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ContainerCustomizer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ContainerCustomizer.java index 7ee01957..9e39d59b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ContainerCustomizer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/config/ContainerCustomizer.java @@ -1,5 +1,5 @@ /* - * Copyright 2019 the original author or authors. + * Copyright 2019-2021 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. @@ -16,7 +16,7 @@ package org.springframework.amqp.rabbit.config; -import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; /** * Called by the container factory after the container is created and configured. @@ -28,7 +28,7 @@ import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer * */ @FunctionalInterface -public interface ContainerCustomizer { +public interface ContainerCustomizer { /** * Configure the container. diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java index d18ccdee..5d3c0530 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractMessageListenerContainer.java @@ -296,6 +296,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * Set the name of the queue(s) to receive messages from. * @param queueName the desired queueName(s) (can not be null) */ + @Override public void setQueueNames(String... queueName) { Assert.noNullElements(queueName, "Queue name(s) cannot be null"); setQueues(Arrays.stream(queueName) @@ -454,9 +455,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor } } - /** - * @return The message listener object to register. - */ + @Override public Object getMessageListener() { return this.messageListener; } @@ -557,6 +556,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * * @param autoStartup true for auto startup. */ + @Override public void setAutoStartup(boolean autoStartup) { this.autoStartup = autoStartup; } @@ -1189,7 +1189,7 @@ public abstract class AbstractMessageListenerContainer extends RabbitAccessor * Delegates to {@link #validateConfiguration()} and {@link #initialize()}. */ @Override - public final void afterPropertiesSet() { + public void afterPropertiesSet() { super.afterPropertiesSet(); Assert.state( this.exposeListenerChannel || !getAcknowledgeMode().isManual(), diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java index 2b3e417d..bcd7dffc 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/AbstractRabbitListenerEndpoint.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-2021 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. @@ -372,31 +372,38 @@ public abstract class AbstractRabbitListenerEndpoint implements RabbitListenerEn @Override public void setupListenerContainer(MessageListenerContainer listenerContainer) { - AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) listenerContainer; + Collection qNames = getQueueNames(); + boolean queueNamesEmpty = qNames.isEmpty(); + if (listenerContainer instanceof AbstractMessageListenerContainer) { + AbstractMessageListenerContainer container = (AbstractMessageListenerContainer) listenerContainer; - boolean queuesEmpty = getQueues().isEmpty(); - boolean queueNamesEmpty = getQueueNames().isEmpty(); - if (!queuesEmpty && !queueNamesEmpty) { - throw new IllegalStateException("Queues or queue names must be provided but not both for " + this); - } - if (queuesEmpty) { - Collection names = getQueueNames(); - container.setQueueNames(names.toArray(new String[names.size()])); + boolean queuesEmpty = getQueues().isEmpty(); + if (!queuesEmpty && !queueNamesEmpty) { + throw new IllegalStateException("Queues or queue names must be provided but not both for " + this); + } + if (queuesEmpty) { + Collection names = qNames; + container.setQueueNames(names.toArray(new String[0])); + } + else { + Collection instances = getQueues(); + container.setQueues(instances.toArray(new Queue[0])); + } + + container.setExclusive(isExclusive()); + if (getPriority() != null) { + Map args = container.getConsumerArguments(); + args.put("x-priority", getPriority()); + container.setConsumerArguments(args); + } + + if (getAdmin() != null) { + container.setAmqpAdmin(getAdmin()); + } } else { - Collection instances = getQueues(); - container.setQueues(instances.toArray(new Queue[instances.size()])); - } - - container.setExclusive(isExclusive()); - if (getPriority() != null) { - Map args = container.getConsumerArguments(); - args.put("x-priority", getPriority()); - container.setConsumerArguments(args); - } - - if (getAdmin() != null) { - container.setAmqpAdmin(getAdmin()); + Assert.state(!queueNamesEmpty, "At least one queue name is required"); + listenerContainer.setQueueNames(qNames.toArray(new String[0])); } setupMessageListener(listenerContainer); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MessageListenerContainer.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MessageListenerContainer.java index e4d8edac..60b56a8b 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MessageListenerContainer.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MessageListenerContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2020 the original author or authors. + * Copyright 2014-2021 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. @@ -18,6 +18,7 @@ package org.springframework.amqp.rabbit.listener; import org.springframework.amqp.core.MessageListener; import org.springframework.context.SmartLifecycle; +import org.springframework.lang.Nullable; /** * Internal abstraction used by the framework representing a message @@ -57,4 +58,33 @@ public interface MessageListenerContainer extends SmartLifecycle { return false; } + /** + * Set the queue names. + * @param queues the queue names. + * @since 2.4 + */ + void setQueueNames(String... queues); + + /** + * Set auto startup. + * @param autoStart true to auto start. + * @since 2.4 + */ + void setAutoStartup(boolean autoStart); + + /** + * Get the message listener. + * @return The message listener object. + * @since 2.4 + */ + @Nullable + Object getMessageListener(); + + /** + * Set the listener id. + * @param id the id. + * @since 2.4 + */ + void setListenerId(String id); + } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java index d7b55035..baea9e49 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MethodRabbitListenerEndpoint.java @@ -19,6 +19,7 @@ package org.springframework.amqp.rabbit.listener; import java.lang.reflect.Method; import java.util.Arrays; +import org.springframework.amqp.rabbit.batch.BatchingStrategy; import org.springframework.amqp.rabbit.listener.adapter.BatchMessagingMessageListenerAdapter; import org.springframework.amqp.rabbit.listener.adapter.HandlerAdapter; import org.springframework.amqp.rabbit.listener.adapter.MessagingMessageListenerAdapter; @@ -53,6 +54,8 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint private RabbitListenerErrorHandler errorHandler; + private AdapterProvider adapterProvider = new DefaultAdapterProvider(); + /** * Set the object instance that should manage this endpoint. * @param bean the target bean instance. @@ -114,6 +117,15 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint return this.messageHandlerMethodFactory; } + /** + * Set a provider to create adapter instances. + * @param adapterProvider the provider. + */ + public void setAdapterProvider(AdapterProvider adapterProvider) { + Assert.notNull(adapterProvider, "'adapterProvider' cannot be null"); + this.adapterProvider = adapterProvider; + } + @Override protected MessagingMessageListenerAdapter createMessageListener(MessageListenerContainer container) { Assert.state(this.messageHandlerMethodFactory != null, @@ -150,14 +162,8 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint * @return the {@link MessagingMessageListenerAdapter} instance. */ protected MessagingMessageListenerAdapter createMessageListenerInstance() { - if (isBatchListener()) { - return new BatchMessagingMessageListenerAdapter(this.bean, this.method, this.returnExceptions, + return this.adapterProvider.getAdapter(isBatchListener(), this.bean, this.method, this.returnExceptions, this.errorHandler, getBatchingStrategy()); - } - else { - return new MessagingMessageListenerAdapter(this.bean, this.method, this.returnExceptions, - this.errorHandler); - } } @Nullable @@ -196,4 +202,43 @@ public class MethodRabbitListenerEndpoint extends AbstractRabbitListenerEndpoint .append(" | method='").append(this.method).append("'"); } + /** + * Provider of listener adapters. + * @since 2.4 + * + */ + public interface AdapterProvider { + + /** + * Get an adapter instance. + * @param batch true for a batch listener. + * @param bean the bean. + * @param method the method. + * @param returnExceptions true to return exceptions. + * @param errorHandler the error handler. + * @param batchingStrategy the batching strategy for batch listeners. + * @return the adapter. + */ + MessagingMessageListenerAdapter getAdapter(boolean batch, Object bean, Method method, boolean returnExceptions, + RabbitListenerErrorHandler errorHandler, @Nullable BatchingStrategy batchingStrategy); + } + + private static final class DefaultAdapterProvider implements AdapterProvider { + + @Override + public MessagingMessageListenerAdapter getAdapter(boolean batch, Object bean, Method method, + boolean returnExceptions, RabbitListenerErrorHandler errorHandler, + @Nullable BatchingStrategy batchingStrategy) { + + if (batch) { + return new BatchMessagingMessageListenerAdapter(bean, method, returnExceptions, errorHandler, + batchingStrategy); + } + else { + return new MessagingMessageListenerAdapter(bean, method, returnExceptions, errorHandler); + } + } + + } + } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MultiMethodRabbitListenerEndpoint.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MultiMethodRabbitListenerEndpoint.java index 5bcd80c9..8a57710a 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MultiMethodRabbitListenerEndpoint.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/MultiMethodRabbitListenerEndpoint.java @@ -44,7 +44,9 @@ public class MultiMethodRabbitListenerEndpoint extends MethodRabbitListenerEndpo * Construct an instance for the provided methods and bean. * @param methods the methods. * @param bean the bean. + * @deprecated - no longer used. */ + @Deprecated public MultiMethodRabbitListenerEndpoint(List methods, Object bean) { this(methods, null, bean); } diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java index d6919b03..09bee504 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/adapter/MessagingMessageListenerAdapter.java @@ -103,6 +103,10 @@ public class MessagingMessageListenerAdapter extends AbstractAdaptableMessageLis this.handlerAdapter = handlerAdapter; } + protected HandlerAdapter getHandlerAdapter() { + return this.handlerAdapter; + } + /** * Set the {@link AmqpHeaderMapper} implementation to use to map the standard * AMQP headers. By default, a {@link org.springframework.amqp.support.SimpleAmqpHeaderMapper diff --git a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/ChannelAwareMessageListener.java b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/ChannelAwareMessageListener.java index 58266a3a..3ab039d2 100644 --- a/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/ChannelAwareMessageListener.java +++ b/spring-rabbit/src/main/java/org/springframework/amqp/rabbit/listener/api/ChannelAwareMessageListener.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -20,6 +20,7 @@ import java.util.List; import org.springframework.amqp.core.Message; import org.springframework.amqp.core.MessageListener; +import org.springframework.lang.Nullable; import com.rabbitmq.client.Channel; @@ -37,10 +38,11 @@ public interface ChannelAwareMessageListener extends MessageListener { *

Implementors are supposed to process the given Message, * typically sending reply messages through the given Session. * @param message the received AMQP message (never null) - * @param channel the underlying Rabbit Channel (never null) + * @param channel the underlying Rabbit Channel (never null + * unless called by the stream listener container). * @throws Exception Any. */ - void onMessage(Message message, Channel channel) throws Exception; // NOSONAR + void onMessage(Message message, @Nullable Channel channel) throws Exception; // NOSONAR @Override default void onMessage(Message message) { diff --git a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MessageListenerTestContainer.java b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MessageListenerTestContainer.java index b42baf5d..16c6e296 100644 --- a/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MessageListenerTestContainer.java +++ b/spring-rabbit/src/test/java/org/springframework/amqp/rabbit/config/MessageListenerTestContainer.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2019 the original author or authors. + * Copyright 2002-2021 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. @@ -21,6 +21,7 @@ import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.amqp.rabbit.listener.RabbitListenerEndpoint; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; +import org.springframework.lang.Nullable; /** * @author Stephane Nicoll @@ -47,6 +48,24 @@ public class MessageListenerTestContainer return endpoint; } + @Override + public void setQueueNames(String... queues) { + } + + @Override + public void setAutoStartup(boolean autoStart) { + } + + @Override + @Nullable + public Object getMessageListener() { + return null; + } + + @Override + public void setListenerId(String id) { + } + public boolean isStarted() { return startInvoked && initializationInvoked; }