From 0ce832f2bc8e5e9be60bc98a38f26ce3bb5239b0 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 28 Jul 2021 12:03:03 -0400 Subject: [PATCH] GH-325: Initial Support for RabbitMQ Stream Plugin Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/325 * Do not propagate `streamContext` header on output. * Fix test broken by previous commit. * Remove overrides from POMs. --- docs/src/main/asciidoc/overview.adoc | 57 ++++ .../properties/RabbitConsumerProperties.java | 27 +- .../RabbitExchangeQueueProvisioner.java | 4 + .../test/junit/rabbit/RabbitTestSupport.java | 10 +- spring-cloud-stream-binder-rabbit/pom.xml | 6 + .../rabbit/RabbitMessageChannelBinder.java | 83 +++--- .../binder/rabbit/StreamContainerUtils.java | 257 ++++++++++++++++++ ...bbitMessageChannelBinderConfiguration.java | 4 +- .../binder/rabbit/RabbitBinderTests.java | 8 +- .../integration/RabbitBinderModuleTests.java | 5 +- .../stream/RabbitStreamBinderModuleTests.java | 97 +++++++ 11 files changed, 515 insertions(+), 43 deletions(-) create mode 100644 spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamContainerUtils.java create mode 100644 spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index 76eb85fe6..62485c7d1 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -164,6 +164,7 @@ Default: none - the broker will generate random consumer tags. containerType:: Select the type of listener container to be used. See https://docs.spring.io/spring-amqp/reference/html/_reference.html#choose-container[Choosing a Container] in the Spring AMQP documentation for more information. +Also see <>. + Default: `simple` deadLetterQueueName:: @@ -413,6 +414,62 @@ Not supported when the `containerType` is `direct`. + Default: `1`. +[[rabbitmq-stream]] +=== Initial Support for the RabbitMQ Stream Plugin + +Basic support for the https://rabbitmq.com/stream.html[RabbitMQ Stream Plugin] is now provided. +To enable this feature, you must add the `spring-rabbit-stream` jar to the class path - it must be the same version as `spring-amqp` and `spring-rabbit`. + +IMPORTANT: The consumer properties described above are not supported when you set the `containerType` property to `stream`; `concurrency` is also not supported at this time. +Only a single stream queue can be consumed by each binding. + +To configure the binder to use `containerType=stream`, you must add an `Environment` `@Bean` and, optionally, a customizer to customize the listener container. + +==== +[source, java] +---- +@Bean +Environment streamEnv() { + return Environment.builder() + .build(); +} + +@Bean +ListenerContainerCustomizer customizer() { + return (cont, dest, group) -> { + StreamListenerContainer container = (StreamListenerContainer) cont; + container.setConsumerCustomizer(builder -> { + builder.offset(OffsetSpecification.first()); + }); + // ... + }; +} +---- +==== + +The stream name (for the purpose of offset tracking) is set to the binding `destination + '.' + group`. +If you decide to use manual offset tracking, the `Context` is available as a message header: + +==== +[source, java] +---- +int count; + +@Bean +public Consumer> input() { + return msg -> { + System.out.println(msg); + if (++count % 1000 == 0) { + Context context = msg.getHeaders().get("rabbitmq_streamContext", Context.class); + context.consumer().store(context.offset()); + } + }; +} +---- +==== + +Refer to the https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/[RabbitMQ Stream Java Client documentation] for information about configuring the environment and consumer builder. + === Advanced Listener Container Configuration To set listener container properties that are not exposed as binder or binding properties, add a single bean of type `ListenerContainerCustomizer` to the application context. diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java index 720647d5a..9d2d0fda5 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java @@ -20,7 +20,6 @@ import javax.validation.constraints.Min; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.MessageDeliveryMode; -import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ContainerType; import org.springframework.util.Assert; /** @@ -348,4 +347,30 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { this.receiveTimeout = receiveTimeout; } + /** + * Container type. + * @author Gary Russell + * @since 3.2 + * + */ + public enum ContainerType { + + /** + * Container where the RabbitMQ consumer dispatches messages to an invoker thread. + */ + SIMPLE, + + /** + * Container where the listener is invoked directly on the RabbitMQ consumer + * thread. + */ + DIRECT, + + /** + * Container that uses the RabbitMQ Stream Client. + */ + STREAM + + } + } diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index fd7d863fd..46264fffc 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -51,6 +51,7 @@ import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties.QuorumConfig; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties.ContainerType; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; @@ -268,6 +269,9 @@ public class RabbitExchangeQueueProvisioner } Binding binding = null; if (properties.getExtension().isBindQueue()) { + if (properties.getExtension().getContainerType().equals(ContainerType.STREAM)) { + queue.getArguments().put("x-queue-type", "stream"); + } declareQueue(queueName, queue); String[] routingKeys = bindingRoutingKeys(properties.getExtension()); if (ObjectUtils.isEmpty(routingKeys)) { diff --git a/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java b/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java index 61740c55a..e44cadf8a 100644 --- a/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java +++ b/spring-cloud-stream-binder-rabbit-test-support/src/main/java/org/springframework/cloud/stream/binder/test/junit/rabbit/RabbitTestSupport.java @@ -27,6 +27,9 @@ import java.util.concurrent.Executors; import javax.net.ServerSocketFactory; import javax.net.SocketFactory; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.cloud.stream.test.junit.AbstractExternalResourceTestSupport; @@ -72,6 +75,8 @@ public class RabbitTestSupport */ public static class RabbitProxy { + private static final Log LOGGER = LogFactory.getLog(RabbitProxy.class); + private final int port; private final ExecutorService serverExec = Executors.newSingleThreadExecutor(); @@ -93,7 +98,8 @@ public class RabbitTestSupport public void start() throws IOException { this.serverSocket = ServerSocketFactory.getDefault() - .createServerSocket(this.port); + .createServerSocket(this.port, 10); + LOGGER.info("Proxy started"); this.serverExec.execute(new Runnable() { @Override @@ -101,6 +107,7 @@ public class RabbitTestSupport try { while (true) { final Socket socket = serverSocket.accept(); + LOGGER.info("Accepted Connection"); socketExec.execute(new Runnable() { @Override @@ -113,6 +120,7 @@ public class RabbitTestSupport @Override public void run() { + LOGGER.info("Running: " + rabbitSocket.getLocalPort()); try { InputStream is = rabbitSocket .getInputStream(); diff --git a/spring-cloud-stream-binder-rabbit/pom.xml b/spring-cloud-stream-binder-rabbit/pom.xml index f3c7ed2b6..45875901a 100644 --- a/spring-cloud-stream-binder-rabbit/pom.xml +++ b/spring-cloud-stream-binder-rabbit/pom.xml @@ -55,6 +55,12 @@ org.springframework.boot spring-boot-starter-amqp + + org.springframework.amqp + spring-rabbit-stream + 2.4.0-M1 + true + org.springframework.integration spring-integration-jmx diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 206646921..99810408b 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -52,6 +52,7 @@ import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; import org.springframework.amqp.rabbit.listener.DirectMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer; import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer; @@ -66,7 +67,6 @@ import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPos import org.springframework.amqp.support.postprocessor.GZipPostProcessor; import org.springframework.beans.factory.DisposableBean; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; -import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ContainerType; import org.springframework.boot.autoconfigure.amqp.RabbitProperties.Retry; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; import org.springframework.cloud.stream.binder.BinderHeaders; @@ -78,6 +78,7 @@ import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; import org.springframework.cloud.stream.binder.HeaderMode; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties.ContainerType; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; @@ -186,7 +187,7 @@ public class RabbitMessageChannelBinder extends public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties, RabbitExchangeQueueProvisioner provisioningProvider, - ListenerContainerCustomizer containerCustomizer) { + ListenerContainerCustomizer containerCustomizer) { this(connectionFactory, rabbitProperties, provisioningProvider, containerCustomizer, null); } @@ -194,7 +195,7 @@ public class RabbitMessageChannelBinder extends public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties, RabbitExchangeQueueProvisioner provisioningProvider, - ListenerContainerCustomizer containerCustomizer, + ListenerContainerCustomizer containerCustomizer, MessageSourceCustomizer sourceCustomizer) { super(new String[0], provisioningProvider, containerCustomizer, sourceCustomizer); @@ -361,6 +362,7 @@ public class RabbitMessageChannelBinder extends headerPatterns.add("!" + BinderHeaders.PARTITION_HEADER); headerPatterns.add("!" + IntegrationMessageHeaderAccessor.SOURCE_DATA); headerPatterns.add("!" + IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT); + headerPatterns.add("!rabbitmq_streamContext"); headerPatterns.addAll(Arrays.asList(extendedProperties.getHeaderPatterns())); mapper.setRequestHeaderNames( headerPatterns.toArray(new String[headerPatterns.size()])); @@ -460,6 +462,50 @@ public class RabbitMessageChannelBinder extends "the RabbitMQ binder does not support embedded headers since RabbitMQ supports headers natively"); String destination = consumerDestination.getName(); RabbitConsumerProperties extension = properties.getExtension(); + MessageListenerContainer listenerContainer = createAndConfigureContainer(consumerDestination, group, + properties, destination, extension); + String[] queues = StringUtils.tokenizeToStringArray(destination, ",", true, true); + listenerContainer.setQueueNames(queues); + getContainerCustomizer().configure(listenerContainer, + consumerDestination.getName(), group); + listenerContainer.afterPropertiesSet(); + + AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter(listenerContainer); + adapter.setBindSourceMessage(true); + adapter.setBeanFactory(this.getBeanFactory()); + adapter.setBeanName("inbound." + destination); + DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper(); + mapper.setRequestHeaderNames(extension.getHeaderPatterns()); + adapter.setHeaderMapper(mapper); + ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure( + consumerDestination, group, properties); + if (properties.getMaxAttempts() > 1) { + adapter.setRetryTemplate(buildRetryTemplate(properties)); + adapter.setRecoveryCallback(errorInfrastructure.getRecoverer()); + } + else { + adapter.setErrorMessageStrategy(errorMessageStrategy); + adapter.setErrorChannel(errorInfrastructure.getErrorChannel()); + } + adapter.setMessageConverter(passThoughConverter); + if (properties.isBatchMode() && extension.isEnableBatching() + && ContainerType.SIMPLE.equals(extension.getContainerType())) { + adapter.setBatchMode(BatchMode.EXTRACT_PAYLOADS_WITH_HEADERS); + } + if (extension.getContainerType().equals(ContainerType.STREAM)) { + StreamContainerUtils.configureAdapter(adapter); + } + return adapter; + } + + private MessageListenerContainer createAndConfigureContainer(ConsumerDestination consumerDestination, + String group, ExtendedConsumerProperties properties, String destination, + RabbitConsumerProperties extension) { + + if (extension.getContainerType().equals(ContainerType.STREAM)) { + return StreamContainerUtils.createContainer(consumerDestination, group, properties, destination, extension, + getApplicationContext()); + } boolean directContainer = extension.getContainerType() .equals(ContainerType.DIRECT); AbstractMessageListenerContainer listenerContainer = directContainer @@ -485,8 +531,6 @@ public class RabbitMessageChannelBinder extends .setRecoveryInterval(extension.getRecoveryInterval()); listenerContainer.setTaskExecutor( new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-")); - String[] queues = StringUtils.tokenizeToStringArray(destination, ",", true, true); - listenerContainer.setQueueNames(queues); listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor); listenerContainer.setMessagePropertiesConverter( RabbitMessageChannelBinder.inboundMessagePropertiesConverter); @@ -504,40 +548,13 @@ public class RabbitMessageChannelBinder extends else if (getApplicationContext() != null) { listenerContainer.setApplicationEventPublisher(getApplicationContext()); } - getContainerCustomizer().configure(listenerContainer, - consumerDestination.getName(), group); if (StringUtils.hasText(extension.getConsumerTagPrefix())) { final AtomicInteger index = new AtomicInteger(); listenerContainer.setConsumerTagStrategy( q -> extension.getConsumerTagPrefix() + "#" + index.getAndIncrement()); } - listenerContainer.afterPropertiesSet(); - - AmqpInboundChannelAdapter adapter = new AmqpInboundChannelAdapter( - listenerContainer); - adapter.setBindSourceMessage(true); - adapter.setBeanFactory(this.getBeanFactory()); - adapter.setBeanName("inbound." + destination); - DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper(); - mapper.setRequestHeaderNames(extension.getHeaderPatterns()); - adapter.setHeaderMapper(mapper); - ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure( - consumerDestination, group, properties); - if (properties.getMaxAttempts() > 1) { - adapter.setRetryTemplate(buildRetryTemplate(properties)); - adapter.setRecoveryCallback(errorInfrastructure.getRecoverer()); - } - else { - adapter.setErrorMessageStrategy(errorMessageStrategy); - adapter.setErrorChannel(errorInfrastructure.getErrorChannel()); - } - adapter.setMessageConverter(passThoughConverter); - if (properties.isBatchMode() && extension.isEnableBatching() - && ContainerType.SIMPLE.equals(extension.getContainerType())) { - adapter.setBatchMode(BatchMode.EXTRACT_PAYLOADS_WITH_HEADERS); - } - return adapter; + return listenerContainer; } private void setSMLCProperties( diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamContainerUtils.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamContainerUtils.java new file mode 100644 index 000000000..d44845f49 --- /dev/null +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamContainerUtils.java @@ -0,0 +1,257 @@ +/* + * Copyright 2021-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.cloud.stream.binder.rabbit; + +import java.nio.charset.Charset; +import java.nio.charset.StandardCharsets; +import java.util.Map; +import java.util.UUID; +import java.util.function.Consumer; +import java.util.function.Supplier; + +import com.rabbitmq.stream.Codec; +import com.rabbitmq.stream.ConsumerBuilder; +import com.rabbitmq.stream.Environment; +import com.rabbitmq.stream.MessageBuilder; +import com.rabbitmq.stream.MessageBuilder.ApplicationPropertiesBuilder; +import com.rabbitmq.stream.MessageBuilder.PropertiesBuilder; +import com.rabbitmq.stream.Properties; +import com.rabbitmq.stream.codec.WrapperMessageBuilder; + +import org.springframework.amqp.core.Message; +import org.springframework.amqp.core.MessageProperties; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; +import org.springframework.amqp.support.converter.MessageConversionException; +import org.springframework.amqp.utils.JavaUtils; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; +import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.context.support.AbstractApplicationContext; +import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; +import org.springframework.integration.amqp.support.AmqpHeaderMapper; +import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; +import org.springframework.lang.Nullable; +import org.springframework.messaging.MessageHeaders; +import org.springframework.rabbit.stream.listener.StreamListenerContainer; +import org.springframework.rabbit.stream.support.StreamMessageProperties; +import org.springframework.rabbit.stream.support.converter.StreamMessageConverter; +import org.springframework.util.Assert; + +/** + * Utilities for stream containers. Used to prevent a hard runtime dependency on + * spring-rabbit-stream. + * + * @author Gary Russell + * @since 3.2 + * + */ +public final class StreamContainerUtils { + + private StreamContainerUtils() { + } + + /** + * Create a {@link StreamListenerContainer}. + * + * @param consumerDestination the destination. + * @param group the group. + * @param properties the properties. + * @param destination the destination. + * @param extension the properties extension. + * @param applicationContext the application context. + * @return the container. + */ + public static MessageListenerContainer createContainer(ConsumerDestination consumerDestination, String group, + ExtendedConsumerProperties properties, String destination, + RabbitConsumerProperties extension, AbstractApplicationContext applicationContext) { + + StreamListenerContainer container = new StreamListenerContainer(applicationContext.getBean(Environment.class)) { + + @Override + public synchronized void setConsumerCustomizer(Consumer consumerCustomizer) { + super.setConsumerCustomizer(builder -> { + builder.name(consumerDestination.getName()); + consumerCustomizer.accept(builder); + }); + } + + + }; + container.setMessageConverter(new DefaultStreamMessageConverter()); + return container; + } + + /** + * Configure the channel adapter for streams support. + * @param adapter the adapter. + */ + public static void configureAdapter(AmqpInboundChannelAdapter adapter) { + adapter.setHeaderMapper(new AmqpHeaderMapper() { + + AmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.inboundMapper(); + + @Override + public Map toHeadersFromRequest(MessageProperties source) { + Map headers = this.mapper.toHeadersFromRequest(source); + headers.put("rabbitmq_streamContext", ((StreamMessageProperties) source).getContext()); + return headers; + } + + @Override + public Map toHeadersFromReply(MessageProperties source) { + return null; + } + + @Override + public void fromHeadersToRequest(MessageHeaders headers, MessageProperties target) { + } + + @Override + public void fromHeadersToReply(MessageHeaders headers, MessageProperties target) { + } + + }); + } + +} + +/** + * Temporary work-around for a bug in spring-rabbit-stream 2.4.0-M1. + */ +class DefaultStreamMessageConverter implements StreamMessageConverter { + + private final Supplier builderSupplier; + + private final Charset charset = StandardCharsets.UTF_8; + + /** + * Construct an instance using a {@link WrapperMessageBuilder}. + */ + DefaultStreamMessageConverter() { + this.builderSupplier = () -> new WrapperMessageBuilder(); + } + + /** + * Construct an instance using the provided codec. + * @param codec the codec. + */ + DefaultStreamMessageConverter(@Nullable Codec codec) { + this.builderSupplier = () -> codec.messageBuilder(); + } + + @Override + public Message toMessage(Object object, StreamMessageProperties 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 org.springframework.amqp.core.MessageBuilder.withBody(streamMessage.getBodyAsBinary()) + .andProperties(messageProperties) + .build(); + } + + @Override + public com.rabbitmq.stream.Message fromMessage(Message message) throws MessageConversionException { + MessageBuilder builder = this.builderSupplier.get(); + PropertiesBuilder propsBuilder = builder.properties(); + MessageProperties props = message.getMessageProperties(); + Assert.isInstanceOf(StreamMessageProperties.class, props); + StreamMessageProperties mProps = (StreamMessageProperties) props; + JavaUtils.INSTANCE + .acceptIfNotNull(mProps.getMessageId(), propsBuilder::messageId) // TODO different types + .acceptIfNotNull(mProps.getUserId(), usr -> propsBuilder.userId(usr.getBytes(this.charset))) + .acceptIfNotNull(mProps.getTo(), propsBuilder::to) + .acceptIfNotNull(mProps.getSubject(), propsBuilder::subject) + .acceptIfNotNull(mProps.getReplyTo(), propsBuilder::replyTo) + .acceptIfNotNull(mProps.getCorrelationId(), propsBuilder::correlationId) // TODO different types + .acceptIfNotNull(mProps.getContentType(), propsBuilder::contentType) + .acceptIfNotNull(mProps.getContentEncoding(), propsBuilder::contentEncoding) + .acceptIfNotNull(mProps.getExpiration(), exp -> propsBuilder.absoluteExpiryTime(Long.parseLong(exp))) + .acceptIfNotNull(mProps.getCreationTime(), propsBuilder::creationTime) + .acceptIfNotNull(mProps.getGroupId(), propsBuilder::groupId) + .acceptIfNotNull(mProps.getGroupSequence(), propsBuilder::groupSequence) + .acceptIfNotNull(mProps.getReplyToGroupId(), propsBuilder::replyToGroupId); + if (mProps.getHeaders().size() > 0) { + ApplicationPropertiesBuilder appPropsBuilder = builder.applicationProperties(); + mProps.getHeaders().forEach((key, val) -> { + mapProp(key, val, appPropsBuilder); + }); + } + builder.addData(message.getBody()); + return builder.build(); + } + + private void mapProp(String key, Object val, ApplicationPropertiesBuilder builder) { // NOSONAR - complexity + if (val instanceof String) { + builder.entry(key, (String) val); + } + else if (val instanceof Long) { + builder.entry(key, (Long) val); + } + else if (val instanceof Integer) { + builder.entry(key, (Integer) val); + } + else if (val instanceof Short) { + builder.entry(key, (Short) val); + } + else if (val instanceof Byte) { + builder.entry(key, (Byte) val); + } + else if (val instanceof Double) { + builder.entry(key, (Double) val); + } + else if (val instanceof Float) { + builder.entry(key, (Float) val); + } + else if (val instanceof Character) { + builder.entry(key, (Character) val); + } + else if (val instanceof UUID) { + builder.entry(key, (UUID) val); + } + else if (val instanceof byte[]) { + builder.entry(key, (byte[]) val); + } + } + + private void toMessageProperties(com.rabbitmq.stream.Message streamMessage, + StreamMessageProperties mProps) { + + Properties properties = streamMessage.getProperties(); + if (properties != null) { + JavaUtils.INSTANCE + .acceptIfNotNull(properties.getMessageIdAsString(), mProps::setMessageId) + .acceptIfNotNull(properties.getUserId(), usr -> mProps.setUserId(new String(usr, this.charset))) + .acceptIfNotNull(properties.getTo(), mProps::setTo) + .acceptIfNotNull(properties.getSubject(), mProps::setSubject) + .acceptIfNotNull(properties.getReplyTo(), mProps::setReplyTo) + .acceptIfNotNull(properties.getCorrelationIdAsString(), mProps::setCorrelationId) + .acceptIfNotNull(properties.getContentType(), mProps::setContentType) + .acceptIfNotNull(properties.getContentEncoding(), mProps::setContentEncoding) + .acceptIfNotNull(properties.getAbsoluteExpiryTime(), + exp -> mProps.setExpiration(Long.toString(exp))) + .acceptIfNotNull(properties.getCreationTime(), mProps::setCreationTime) + .acceptIfNotNull(properties.getGroupId(), mProps::setGroupId) + .acceptIfNotNull(properties.getGroupSequence(), mProps::setGroupSequence) + .acceptIfNotNull(properties.getReplyToGroupId(), mProps::setReplyToGroupId); + } + Map applicationProperties = streamMessage.getApplicationProperties(); + if (applicationProperties != null) { + mProps.getHeaders().putAll(applicationProperties); + } + } + +} diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java index 20b571ccc..3bfd1d53f 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitMessageChannelBinderConfiguration.java @@ -24,7 +24,7 @@ import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; -import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor; import org.springframework.amqp.support.postprocessor.GZipPostProcessor; import org.springframework.beans.factory.annotation.Autowired; @@ -77,7 +77,7 @@ public class RabbitMessageChannelBinderConfiguration { @Bean RabbitMessageChannelBinder rabbitMessageChannelBinder( - @Nullable ListenerContainerCustomizer listenerContainerCustomizer, + @Nullable ListenerContainerCustomizer listenerContainerCustomizer, @Nullable MessageSourceCustomizer sourceCustomizer, @Nullable ProducerMessageHandlerCustomizer producerMessageHandlerCustomizer, @Nullable ConsumerEndpointCustomizer consumerCustomizer, diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java index 3ecc5b6e7..8ec092827 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java @@ -83,7 +83,6 @@ import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; -import org.springframework.boot.autoconfigure.amqp.RabbitProperties.ContainerType; import org.springframework.cloud.stream.binder.BinderException; import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.Binding; @@ -99,6 +98,7 @@ import org.springframework.cloud.stream.binder.RequeueCurrentMessageException; import org.springframework.cloud.stream.binder.Spy; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitCommonProperties.QuorumConfig; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties.ContainerType; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; @@ -801,7 +801,7 @@ public class RabbitBinderTests extends assertThat(mode).isEqualTo(MessageDeliveryMode.PERSISTENT); List requestHeaders = TestUtils.getPropertyValue(endpoint, "headerMapper.requestHeaderMatcher.matchers", List.class); - assertThat(requestHeaders).hasSize(4); + assertThat(requestHeaders).hasSize(5); producerBinding.unbind(); assertThat(endpoint.isRunning()).isFalse(); assertThat(TestUtils.getPropertyValue(endpoint, "amqpTemplate.transactional", @@ -2340,8 +2340,8 @@ public class RabbitBinderTests extends private void verifyFooRequestProducer(Lifecycle endpoint) { List requestMatchers = TestUtils.getPropertyValue(endpoint, "headerMapper.requestHeaderMatcher.matchers", List.class); - assertThat(requestMatchers).hasSize(4); - assertThat(TestUtils.getPropertyValue(requestMatchers.get(3), "pattern")) + assertThat(requestMatchers).hasSize(5); + assertThat(TestUtils.getPropertyValue(requestMatchers.get(4), "pattern")) .isEqualTo("foo"); } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java index b82637ceb..e5814e042 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/integration/RabbitBinderModuleTests.java @@ -39,6 +39,7 @@ import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer; +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.utils.test.TestUtils; import org.springframework.beans.DirectFieldAccessor; @@ -381,8 +382,8 @@ public class RabbitBinderModuleTests { public static class SimpleProcessor { @Bean - public ListenerContainerCustomizer containerCustomizer() { - return (c, q, g) -> c.setBeanName( + public ListenerContainerCustomizer containerCustomizer() { + return (c, q, g) -> ((AbstractMessageListenerContainer) c).setBeanName( "setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g)); } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java new file mode 100644 index 000000000..9517023e9 --- /dev/null +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java @@ -0,0 +1,97 @@ +/* + * Copyright 2021-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.cloud.stream.binder.rabbit.stream; + +import com.rabbitmq.stream.ConsumerBuilder; +import com.rabbitmq.stream.Environment; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import org.springframework.amqp.rabbit.listener.MessageListenerContainer; +import org.springframework.boot.WebApplicationType; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.builder.SpringApplicationBuilder; +import org.springframework.cloud.stream.binder.BinderFactory; +import org.springframework.cloud.stream.binder.Binding; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties.ContainerType; +import org.springframework.cloud.stream.config.ListenerContainerCustomizer; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.messaging.MessageChannel; +import org.springframework.rabbit.stream.listener.StreamListenerContainer; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.mock; + +/** + * @author Gary Russell + */ +public class RabbitStreamBinderModuleTests { + + private ConfigurableApplicationContext context; + + @AfterEach + public void tearDown() { + if (context != null) { + context.close(); + context = null; + } + } + + @Test + public void testExtendedProperties() { + context = new SpringApplicationBuilder(SimpleProcessor.class) + .web(WebApplicationType.NONE) + .run("--server.port=0"); + BinderFactory binderFactory = context.getBean(BinderFactory.class); + RabbitMessageChannelBinder rabbitBinder = (RabbitMessageChannelBinder) binderFactory.getBinder(null, + MessageChannel.class); + RabbitConsumerProperties rProps = new RabbitConsumerProperties(); + rProps.setContainerType(ContainerType.STREAM); + ExtendedConsumerProperties props = + new ExtendedConsumerProperties(rProps); + props.setAutoStartup(false); + Binding binding = rabbitBinder.bindConsumer("testStream", "grp", new QueueChannel(), props); + assertThat(TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer")) + .isInstanceOf(StreamListenerContainer.class); + } + + @SpringBootApplication + public static class SimpleProcessor { + + @Bean + public ListenerContainerCustomizer containerCustomizer() { + return (c, q, g) -> ((StreamListenerContainer) c).setBeanName( + "setByCustomizerForQueue:" + q + (g == null ? "" : ",andGroup:" + g)); + } + + @Bean + Environment env() { + Environment env = mock(Environment.class); + given(env.consumerBuilder()).willReturn(mock(ConsumerBuilder.class)); + return env; + } + + } + +}