diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/pom.xml b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/pom.xml index 0b3e0c5de..b3d24ec9d 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/pom.xml +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/pom.xml @@ -22,6 +22,12 @@ org.springframework.boot spring-boot-starter-amqp + + org.springframework.amqp + spring-rabbit-stream + 3.0.0-SNAPSHOT + true + org.springframework.integration spring-integration-amqp diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java index 639d2bcf2..25ae65a84 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java @@ -115,7 +115,7 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { private int frameMaxHeadroom = 20_000; /** - * The container type, SIMPLE or DIRECT. + * The container type, SIMPLE, DIRECT, or STREAM. */ private ContainerType containerType = ContainerType.SIMPLE; @@ -139,6 +139,12 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { */ private Long receiveTimeout; + /** + * When the container type is STREAM, set this to true to create a super stream with + * competing consumers. + */ + private boolean superStream; + public boolean isTransacted() { return transacted; } @@ -347,6 +353,14 @@ public class RabbitConsumerProperties extends RabbitCommonProperties { this.receiveTimeout = receiveTimeout; } + public boolean isSuperStream() { + return this.superStream; + } + + public void setSuperStream(boolean superStream) { + this.superStream = superStream; + } + /** * Container type. * @author Gary Russell diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java index 46d789657..678eb7c9d 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java @@ -157,6 +157,12 @@ public class RabbitProducerProperties extends RabbitCommonProperties { */ private AlternateExchange alternateExchange; + /** + * When the producer type is STREAM_*, set this to true to publish to a super stream. + * Also requires a partition key. + */ + private boolean superStream; + /** * @param requestHeaderPatterns the patterns. * @deprecated - use {@link #setHeaderPatterns(String[])}. @@ -319,6 +325,14 @@ public class RabbitProducerProperties extends RabbitCommonProperties { this.alternateExchange = alternate; } + public boolean isSuperStream() { + return this.superStream; + } + + public void setSuperStream(boolean superStream) { + this.superStream = superStream; + } + public static class AlternateExchange { /** diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index f6f85b44d..c93e87467 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -22,6 +22,8 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.Collectors; +import java.util.stream.IntStream; import java.util.stream.Stream; import org.apache.commons.logging.Log; @@ -64,6 +66,7 @@ import org.springframework.cloud.stream.provisioning.ProvisioningProvider; import org.springframework.context.ApplicationListener; import org.springframework.context.support.GenericApplicationContext; import org.springframework.lang.Nullable; +import org.springframework.rabbit.stream.config.SuperStream; import org.springframework.util.Assert; import org.springframework.util.ObjectUtils; import org.springframework.util.StringUtils; @@ -245,9 +248,11 @@ public class RabbitExchangeQueueProvisioner + ", bound to: " + name); } String prefix = properties.getExtension().getPrefix(); - final String exchangeName = applyPrefix(prefix, name); + String exchangeName = applyPrefix(prefix, name); + ContainerType containerType = properties.getExtension().getContainerType(); + boolean superStream = containerType.equals(ContainerType.STREAM) && properties.getExtension().isSuperStream(); Exchange exchange = buildExchange(properties.getExtension(), exchangeName, null, null); - if (properties.getExtension().isDeclareExchange()) { + if (!superStream && properties.getExtension().isDeclareExchange()) { declareExchange(exchangeName, anonymous ? anonymousGroup : group, exchange); } String queueName = applyPrefix(prefix, baseQueueName); @@ -258,6 +263,7 @@ public class RabbitExchangeQueueProvisioner String anonQueueName = queueName; queue = new AnonymousQueue((org.springframework.amqp.core.NamingStrategy) () -> anonQueueName, queueArgs(queueName, properties.getExtension(), false)); + queueName = queue.getName(); } else { if (partitioned) { @@ -275,25 +281,68 @@ 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)) { - binding = declareConsumerBindings(name, null, properties, exchange, partitioned, queue); + if (superStream) { + provisionSuperStream(properties, name); } else { - for (String routingKey : routingKeys) { - binding = declareConsumerBindings(name, routingKey, properties, exchange, partitioned, queue); + if (containerType.equals(ContainerType.STREAM)) { + queue.getArguments().put("x-queue-type", "stream"); + } + declareQueue(queueName, queue); + String[] routingKeys = bindingRoutingKeys(properties.getExtension()); + if (ObjectUtils.isEmpty(routingKeys)) { + binding = declareConsumerBindings(name, null, properties, exchange, partitioned, queue); + } + else { + for (String routingKey : routingKeys) { + binding = declareConsumerBindings(name, routingKey, properties, exchange, partitioned, queue); + } } } } - if (durable) { + if (durable && !superStream) { autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName, group, properties.getExtension()); } - return new RabbitConsumerDestination(queue.getName(), binding, anonymous ? baseQueueName : group, name); + if (superStream) { + queueName = name; // group is used in the consumer for super streams so not part of the name. + } + return new RabbitConsumerDestination(queueName, binding, anonymous ? baseQueueName : group, name); + } + + private void provisionSuperStream(ExtendedConsumerProperties properties, + String name) { + + String routingKey = properties.getExtension().getBindingRoutingKey(); + String rk = routingKey == null ? name : routingKey; + SuperStream ss = new SuperStream(name, properties.getInstanceCount(), (q, i) -> IntStream.range(0, i) + .mapToObj(j -> rk + "-" + j) + .collect(Collectors.toList())); + synchronized (this.autoDeclareContext) { + if (!this.autoDeclareContext.containsBean(name + ".superStream")) { + this.autoDeclareContext.getBeanFactory().registerSingleton(name + ".superStream", ss); + } + } + try { + ss.getDeclarables().forEach(dec -> { + if (dec instanceof Exchange exch) { + this.rabbitAdmin.declareExchange(exch); + } + else if (dec instanceof Queue queue) { + this.rabbitAdmin.declareQueue(queue); + } + else if (dec instanceof Binding binding) { + this.rabbitAdmin.declareBinding(binding); + } + }); + } + catch (AmqpConnectException e) { + if (this.logger.isDebugEnabled()) { + this.logger.debug("Declaration of super stream: " + name + + " deferred - connection not available"); + } + } + } /** diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/pom.xml b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/pom.xml index 28b88b37b..cd1d30493 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/pom.xml +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/pom.xml @@ -64,11 +64,23 @@ org.springframework.amqp spring-rabbit-stream + 3.0.0-SNAPSHOT true + + org.springframework.integration + spring-integration-amqp + 6.0.0-SNAPSHOT + + + org.springframework.integration + spring-integration-core + 6.0.0-SNAPSHOT + org.springframework.integration spring-integration-jmx + 6.0.0-SNAPSHOT org.springframework.integration diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java index 6f33e2dae..8db8fddb1 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java @@ -422,7 +422,10 @@ public class RabbitMessageChannelBinder extends private AmqpHeaderMapper configureHeaderMapper(RabbitProducerProperties extendedProperties) { DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.outboundMapper(); List headerPatterns = new ArrayList<>(extendedProperties.getHeaderPatterns().length + 3); - headerPatterns.add("!" + BinderHeaders.PARTITION_HEADER); + if (!extendedProperties.isSuperStream()) { + // need to keep this header until later + headerPatterns.add("!" + BinderHeaders.PARTITION_HEADER); + } headerPatterns.add("!" + IntegrationMessageHeaderAccessor.SOURCE_DATA); headerPatterns.add("!" + IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT); headerPatterns.add("!rabbitmq_streamContext"); @@ -500,7 +503,10 @@ public class RabbitMessageChannelBinder extends MessageListenerContainer listenerContainer = createAndConfigureContainer(consumerDestination, group, properties, destination, extension); String[] queues = StringUtils.tokenizeToStringArray(destination, ",", true, true); - listenerContainer.setQueueNames(queues); + if (properties.getExtension().getContainerType() != ContainerType.STREAM + || !properties.getExtension().isSuperStream()) { + listenerContainer.setQueueNames(queues); + } getContainerCustomizer().configure(listenerContainer, consumerDestination.getName(), group); listenerContainer.afterPropertiesSet(); @@ -523,11 +529,12 @@ public class RabbitMessageChannelBinder extends adapter.setErrorChannel(errorInfrastructure.getErrorChannel()); } adapter.setMessageConverter(passThoughConverter); + ContainerType containerType = extension.getContainerType(); if (properties.isBatchMode() && extension.isEnableBatching() - && ContainerType.SIMPLE.equals(extension.getContainerType())) { + && ContainerType.SIMPLE.equals(containerType)) { adapter.setBatchMode(BatchMode.EXTRACT_PAYLOADS_WITH_HEADERS); } - if (extension.getContainerType().equals(ContainerType.STREAM)) { + if (containerType.equals(ContainerType.STREAM)) { StreamUtils.configureAdapter(adapter); } return adapter; diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitStreamMessageHandler.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitStreamMessageHandler.java deleted file mode 100644 index 36f587d73..000000000 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitStreamMessageHandler.java +++ /dev/null @@ -1,275 +0,0 @@ -/* - * Copyright 2021-2022 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.util.concurrent.CompletableFuture; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.TimeoutException; - -import org.springframework.amqp.core.MessageProperties; -import org.springframework.amqp.support.AmqpHeaders; -import org.springframework.amqp.support.converter.ContentTypeDelegatingMessageConverter; -import org.springframework.amqp.support.converter.MessageConverter; -import org.springframework.context.Lifecycle; -import org.springframework.integration.amqp.support.AmqpHeaderMapper; -import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; -import org.springframework.integration.handler.AbstractMessageHandler; -import org.springframework.lang.Nullable; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessageHandlingException; -import org.springframework.messaging.MessageHeaders; -import org.springframework.rabbit.stream.producer.RabbitStreamOperations; -import org.springframework.rabbit.stream.support.StreamMessageProperties; -import org.springframework.util.Assert; -import org.springframework.util.MimeType; - -/** - * {@link MessageHandler} based on {@link RabbitStreamOperations}. - * - * TODO: This class will move to Spring Integration in 6.0. - * - * @author Gary Russell - * @author Chris Bono - * @since 3.2 - * - */ -public class RabbitStreamMessageHandler extends AbstractMessageHandler implements Lifecycle { - - private static final int DEFAULT_CONFIRM_TIMEOUT = 10_000; - - private final RabbitStreamOperations streamOperations; - - private boolean sync; - - private long confirmTimeout = DEFAULT_CONFIRM_TIMEOUT; - - private SuccessCallback> successCallback = msg -> { }; - - private FailureCallback failureCallback = (msg, ex) -> { }; - - private AmqpHeaderMapper headerMapper = DefaultAmqpHeaderMapper.outboundMapper(); - - private boolean headersMappedLast; - - /** - * Create an instance with the provided {@link RabbitStreamOperations}. - * @param streamOperations the operations. - */ - public RabbitStreamMessageHandler(RabbitStreamOperations streamOperations) { - Assert.notNull(streamOperations, "'streamOperations' cannot be null"); - this.streamOperations = streamOperations; - } - - /** - * Set a callback to be invoked when a send is successful. - * @param successCallback the callback. - */ - public void setSuccessCallback(SuccessCallback> successCallback) { - Assert.notNull(successCallback, "'successCallback' cannot be null"); - this.successCallback = successCallback; - } - - /** - * Set a callback to be invoked when a send fails. - * @param failureCallback the callback. - */ - public void setFailureCallback(FailureCallback failureCallback) { - Assert.notNull(failureCallback, "'failureCallback' cannot be null"); - this.failureCallback = failureCallback; - } - - /** - * Set to true to wait for a confirmation. - * @param sync true to wait. - * @see #setConfirmTimeout(long) - */ - public void setSync(boolean sync) { - this.sync = sync; - } - - /** - * Set the confirm timeout. - * @param confirmTimeout the timeout. - * @see #setSync(boolean) - */ - public void setConfirmTimeout(long confirmTimeout) { - this.confirmTimeout = confirmTimeout; - } - - /** - * Set a custom {@link AmqpHeaderMapper} for mapping request and reply headers. - * Defaults to {@link DefaultAmqpHeaderMapper#outboundMapper()}. - * @param headerMapper the {@link AmqpHeaderMapper} to use. - */ - public void setHeaderMapper(AmqpHeaderMapper headerMapper) { - Assert.notNull(headerMapper, "headerMapper must not be null"); - this.headerMapper = headerMapper; - } - - /** - * When mapping headers for the outbound message, determine whether the headers are - * mapped before the message is converted, or afterwards. This only affects headers - * that might be added by the message converter. When false, the converter's headers - * win; when true, any headers added by the converter will be overridden (if the - * source message has a header that maps to those headers). You might wish to set this - * to true, for example, when using a - * {@link org.springframework.amqp.support.converter.SimpleMessageConverter} with a - * String payload that contains json; the converter will set the content type to - * {@code text/plain} which can be overridden to {@code application/json} by setting - * the {@link AmqpHeaders#CONTENT_TYPE} message header. Default: false. - * @param headersMappedLast true if headers are mapped after conversion. - */ - public void setHeadersMappedLast(boolean headersMappedLast) { - this.headersMappedLast = headersMappedLast; - } - - /** - * Return the {@link RabbitStreamOperations}. - * @return the operations. - */ - public RabbitStreamOperations getStreamOperations() { - return this.streamOperations; - } - - @Override - protected void handleMessageInternal(Message requestMessage) { - CompletableFuture future; - com.rabbitmq.stream.Message streamMessage; - if (requestMessage.getPayload() instanceof com.rabbitmq.stream.Message) { - streamMessage = (com.rabbitmq.stream.Message) requestMessage.getPayload(); - } - else { - MessageConverter converter = streamOperations.messageConverter(); - org.springframework.amqp.core.Message amqpMessage = mapMessage(requestMessage, converter, - this.headerMapper, this.headersMappedLast); - streamMessage = this.streamOperations.streamMessageConverter().fromMessage(amqpMessage); - } - future = this.streamOperations.send(streamMessage); - handleConfirms(requestMessage, future); - } - - private void handleConfirms(Message message, CompletableFuture future) { - future.whenComplete((bool, ex) -> { - if (ex != null) { - this.failureCallback.failure(message, ex); - } - else { - this.successCallback.onSuccess(message); - } - }); - if (this.sync) { - try { - future.get(this.confirmTimeout, TimeUnit.MILLISECONDS); - } - catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - throw new MessageHandlingException(message, ex); - } - catch (ExecutionException | TimeoutException ex) { - throw new MessageHandlingException(message, ex); - } - } - } - - /* - * TODO Copied/modified from MapppingUtils until SI 6.0 - */ - private static org.springframework.amqp.core.Message mapMessage(Message message, - MessageConverter converter, AmqpHeaderMapper headerMapper, boolean headersMappedLast) { - - MessageProperties amqpMessageProperties = new StreamMessageProperties(); - org.springframework.amqp.core.Message amqpMessage; - if (!headersMappedLast) { - mapHeaders(message.getHeaders(), amqpMessageProperties, headerMapper); - } - if (converter instanceof ContentTypeDelegatingMessageConverter && headersMappedLast) { - String contentType = contentTypeAsString(message.getHeaders()); - if (contentType != null) { - amqpMessageProperties.setContentType(contentType); - } - } - amqpMessage = converter.toMessage(message.getPayload(), amqpMessageProperties); - if (headersMappedLast) { - mapHeaders(message.getHeaders(), amqpMessageProperties, headerMapper); - } - return amqpMessage; - } - - private static void mapHeaders(MessageHeaders messageHeaders, MessageProperties amqpMessageProperties, - AmqpHeaderMapper headerMapper) { - - headerMapper.fromHeadersToRequest(messageHeaders, amqpMessageProperties); - } - - private static String contentTypeAsString(MessageHeaders headers) { - Object contentType = headers.get(AmqpHeaders.CONTENT_TYPE); - if (contentType instanceof MimeType) { - contentType = contentType.toString(); - } - if (contentType instanceof String) { - return (String) contentType; - } - else if (contentType != null) { - throw new IllegalArgumentException(AmqpHeaders.CONTENT_TYPE - + " header must be a MimeType or String, found: " + contentType.getClass().getName()); - } - return null; - } - /* - * End copied/modified from MappingUtils - */ - - @Override - public void start() { - } - - @Override - public void stop() { - this.streamOperations.close(); - } - - @Override - public boolean isRunning() { - return true; - } - - /** - * Callback for when publishing succeeds. - */ - interface SuccessCallback { - /** - * Called when the future completes with success. - * Note that Exceptions raised by this method are ignored. - * @param result the result of the future - */ - void onSuccess(@Nullable T result); - } - - /** - * Callback for when publishing fails. - */ - interface FailureCallback { - /** - * Message publish failure. - * @param message the message. - * @param throwable the throwable. - */ - void failure(Message message, Throwable throwable); - } -} diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamUtils.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamUtils.java index aee1466ac..70a432509 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamUtils.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/StreamUtils.java @@ -24,6 +24,7 @@ import com.rabbitmq.stream.Environment; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.amqp.support.converter.MessageConverter; +import org.springframework.cloud.stream.binder.BinderHeaders; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; @@ -33,13 +34,12 @@ import org.springframework.cloud.stream.provisioning.ConsumerDestination; import org.springframework.cloud.stream.provisioning.ProducerDestination; import org.springframework.context.support.AbstractApplicationContext; import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; +import org.springframework.integration.amqp.outbound.RabbitStreamMessageHandler; import org.springframework.integration.amqp.support.AmqpHeaderMapper; import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.MessageHandlingException; import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.support.ErrorMessage; import org.springframework.rabbit.stream.listener.ConsumerCustomizer; import org.springframework.rabbit.stream.listener.StreamListenerContainer; import org.springframework.rabbit.stream.producer.RabbitStreamTemplate; @@ -79,7 +79,9 @@ public final class StreamUtils { @Override public synchronized void setConsumerCustomizer(ConsumerCustomizer consumerCustomizer) { super.setConsumerCustomizer((id, builder) -> { - builder.name(consumerDestination.getName() + "." + group); + if (!properties.getExtension().isSuperStream()) { + builder.name(consumerDestination.getName() + "." + group); + } consumerCustomizer.accept(id, builder); }); } @@ -91,6 +93,9 @@ public final class StreamUtils { if (beanName != null) { container.setStreamConverter(applicationContext.getBean(beanName, StreamMessageConverter.class)); } + if (properties.getExtension().isSuperStream()) { + container.superStream(consumerDestination.getName(), consumerDestination.getName() + "." + group); + } return container; } @@ -146,6 +151,13 @@ public final class StreamUtils { RabbitStreamTemplate template = new RabbitStreamTemplate(applicationContext.getBean(Environment.class), producerDestination.getName()); + if (extendedProperties.isSuperStream()) { + template.setSuperStreamRouting(message -> { + Object property = message.getApplicationProperties().getOrDefault(BinderHeaders.PARTITION_HEADER, "0"); + message.getApplicationProperties().remove(BinderHeaders.PARTITION_HEADER); + return "" + property; + }); + } String beanName = extendedProperties.getStreamMessageConverterBeanName(); if (beanName != null) { template.setMessageConverter(applicationContext.getBean(beanName, MessageConverter.class)); @@ -156,9 +168,11 @@ public final class StreamUtils { } RabbitStreamMessageHandler handler = new RabbitStreamMessageHandler(template); if (errorChannel != null) { - handler.setFailureCallback((msg, ex) -> { - errorChannel.send(new ErrorMessage(new MessageHandlingException(msg, ex))); - }); + handler.setSendFailureChannel(errorChannel); + } + beanName = extendedProperties.getConfirmAckChannel(); + if (beanName != null) { + handler.setSendSuccessChannelName(beanName); } handler.setHeaderMapper(headerMapperFunction.apply(extendedProperties)); handler.setSync(ProducerType.STREAM_SYNC.equals(producerProperties.getExtension().getProducerType())); diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java index 18f0f015f..5a7b6be7d 100644 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java +++ b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamBinderModuleTests.java @@ -22,7 +22,10 @@ import com.rabbitmq.stream.OffsetSpecification; import com.rabbitmq.stream.ProducerBuilder; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; +import org.testcontainers.containers.RabbitMQContainer; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.listener.MessageListenerContainer; import org.springframework.boot.WebApplicationType; import org.springframework.boot.autoconfigure.SpringBootApplication; @@ -33,7 +36,7 @@ import org.springframework.cloud.stream.binder.Binding; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; -import org.springframework.cloud.stream.binder.rabbit.RabbitStreamMessageHandler; +import org.springframework.cloud.stream.binder.rabbit.RabbitTestContainer; 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; @@ -42,6 +45,7 @@ import org.springframework.cloud.stream.config.ListenerContainerCustomizer; import org.springframework.cloud.stream.config.ProducerMessageHandlerCustomizer; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.integration.amqp.outbound.RabbitStreamMessageHandler; import org.springframework.integration.channel.DirectChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.test.util.TestUtils; @@ -50,6 +54,7 @@ import org.springframework.messaging.MessageHandler; import org.springframework.rabbit.stream.listener.StreamListenerContainer; import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.BDDMockito.given; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.verify; @@ -59,6 +64,8 @@ import static org.mockito.Mockito.verify; */ public class RabbitStreamBinderModuleTests { + private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance(); + private ConfigurableApplicationContext context; @AfterEach @@ -90,6 +97,32 @@ public class RabbitStreamBinderModuleTests { ((StreamListenerContainer) container).stop(); } + @Test + void testSuperStreamContainer() { + 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); + rProps.setSuperStream(true); + ExtendedConsumerProperties props = + new ExtendedConsumerProperties(rProps); + props.setAutoStartup(false); + props.setInstanceCount(1); + Binding binding = rabbitBinder.bindConsumer("testSuperStream", "grp", new QueueChannel(), props); + Object container = TestUtils.getPropertyValue(binding, "lifecycle.messageListenerContainer"); + assertThat(container).isInstanceOf(StreamListenerContainer.class); + ((StreamListenerContainer) container).start(); + ConsumerBuilder builder = this.context.getBean(ConsumerBuilder.class); + verify(builder).singleActiveConsumer(); + verify(builder).superStream("testSuperStream"); + verify(builder).name("testSuperStream.grp"); + ((StreamListenerContainer) container).stop(); + } + @Test void testStreamHandler() { context = new SpringApplicationBuilder(SimpleProcessor.class) @@ -103,13 +136,18 @@ public class RabbitStreamBinderModuleTests { ExtendedProducerProperties props = new ExtendedProducerProperties(rProps); Binding binding = rabbitBinder.bindProducer("testStream", new DirectChannel(), props); - Object handler = TestUtils.getPropertyValue(binding, "lifecycle"); + Object handler = TestUtils.getPropertyValue(binding, "val$producerMessageHandler"); assertThat(handler).isInstanceOf(RabbitStreamMessageHandler.class); } @SpringBootApplication(proxyBeanMethods = false) public static class SimpleProcessor { + @Bean + ConnectionFactory cf() { + return new CachingConnectionFactory(RABBITMQ.getMappedPort(5672)); + } + @Bean ProducerMessageHandlerCustomizer handlerCustomizer() { return (hand, dest) -> { @@ -139,7 +177,10 @@ public class RabbitStreamBinderModuleTests { @Bean ConsumerBuilder consumerBuilder() { - return mock(ConsumerBuilder.class); + ConsumerBuilder mock = mock(ConsumerBuilder.class); + given(mock.superStream(anyString())).willReturn(mock); + given(mock.singleActiveConsumer()).willReturn(mock); + return mock; } @Bean diff --git a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamMessageHandlerTests.java b/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamMessageHandlerTests.java deleted file mode 100644 index 575de2ccd..000000000 --- a/binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/stream/RabbitStreamMessageHandlerTests.java +++ /dev/null @@ -1,120 +0,0 @@ -/* - * Copyright 2021-2022 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 java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicReference; - -import com.rabbitmq.stream.Address; -import com.rabbitmq.stream.Consumer; -import com.rabbitmq.stream.Environment; -import com.rabbitmq.stream.OffsetSpecification; -import org.junit.jupiter.api.Disabled; -import org.junit.jupiter.api.Test; -import org.testcontainers.containers.RabbitMQContainer; - -import org.springframework.cloud.stream.binder.rabbit.RabbitStreamMessageHandler; -import org.springframework.cloud.stream.binder.rabbit.RabbitTestContainer; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.rabbit.stream.producer.RabbitStreamTemplate; - -import static org.assertj.core.api.Assertions.assertThat; - -/** - * @author Gary Russell - * @author Chris Bono - * @since 3.2 - */ -public class RabbitStreamMessageHandlerTests { - - private static final RabbitMQContainer RABBITMQ = RabbitTestContainer.sharedInstance(); - - @Test - void convertAndSend() throws InterruptedException { - Environment env = Environment.builder() - .lazyInitialization(true) - .addressResolver(add -> new Address("localhost", RABBITMQ.getMappedPort(5552))) - .build(); - try { - env.deleteStream("stream.stream"); - } - catch (Exception e) { - } - env.streamCreator().stream("stream.stream").create(); - RabbitStreamTemplate streamTemplate = new RabbitStreamTemplate(env, "stream.stream"); - RabbitStreamMessageHandler handler = new RabbitStreamMessageHandler(streamTemplate); - handler.setSync(true); - handler.handleMessage(MessageBuilder.withPayload("foo") - .setHeader("bar", "baz") - .build()); - CountDownLatch latch = new CountDownLatch(1); - AtomicReference received = new AtomicReference<>(); - Consumer consumer = env.consumerBuilder().stream("stream.stream") - .offset(OffsetSpecification.first()) - .messageHandler((context, msg) -> { - received.set(msg); - latch.countDown(); - }) - .build(); - assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(received.get()).isNotNull(); - assertThat(received.get().getBodyAsBinary()).isEqualTo("foo".getBytes()); - assertThat((String) received.get().getApplicationProperties().get("bar")).isEqualTo("baz"); - consumer.close(); - handler.stop(); - } - - @Test - @Disabled - void sendNative() throws InterruptedException { - Environment env = Environment.builder() - .lazyInitialization(true) - .build(); - try { - env.deleteStream("stream.stream"); - } - catch (Exception e) { - } - env.streamCreator().stream("stream.stream").create(); - RabbitStreamTemplate streamTemplate = new RabbitStreamTemplate(env, "stream.stream"); - RabbitStreamMessageHandler handler = new RabbitStreamMessageHandler(streamTemplate); - handler.setSync(true); - handler.handleMessage(MessageBuilder.withPayload(streamTemplate.messageBuilder() - .addData("foo".getBytes()) - .applicationProperties().entry("bar", "baz") - .messageBuilder() - .build()) - .build()); - CountDownLatch latch = new CountDownLatch(1); - AtomicReference received = new AtomicReference<>(); - Consumer consumer = env.consumerBuilder().stream("stream.stream") - .offset(OffsetSpecification.first()) - .messageHandler((context, msg) -> { - received.set(msg); - latch.countDown(); - }) - .build(); - assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); - assertThat(received.get()).isNotNull(); - assertThat(received.get().getBodyAsBinary()).isEqualTo("foo".getBytes()); - assertThat((String) received.get().getApplicationProperties().get("bar")).isEqualTo("baz"); - consumer.close(); - handler.stop(); - } - -} diff --git a/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc b/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc index a5a952ba0..3bcaa2854 100644 --- a/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc +++ b/docs/src/main/asciidoc/rabbit/rabbit_overview.adoc @@ -423,17 +423,12 @@ To enable this feature, you must add the `spring-rabbit-stream` jar to the class 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. +To configure the binder to use `containerType=stream`, Spring Boot will automatically configure an `Environment` `@Bean` from the application properties. +You can, optionally, add a customizer to customize the listener container. ==== [source, java] ---- -@Bean -Environment streamEnv() { - return Environment.builder() - .build(); -} - @Bean ListenerContainerCustomizer customizer() { return (cont, dest, group) -> { @@ -473,6 +468,39 @@ public Consumer> input() { 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. +[[rabbitmq-super-stream-consumer]] +==== Consumer Support for the RabbitMQ Super Streams + +See https://blog.rabbitmq.com/posts/2022/07/rabbitmq-3-11-feature-preview-super-streams[Super Streams] for information about super streams. + +Use of super streams allows for automatic scale-up scale-down with a single active consumer on each partition of a super stream. + +Configuration example: + +==== +[source, java] +---- +@Bean +public Consumer input() { + ... +} +---- +==== + +==== +[source, properties] +---- +spring.cloud.stream.bindings.input-in-0.destination=super +spring.cloud.stream.bindings.input-in-0.group=test +spring.cloud.stream.bindings.input-in-0.consumer.instance-count=3 +spring.cloud.stream.rabbit.bindings.input-in-0.consumer.container-type=STREAM +spring.cloud.stream.rabbit.bindings.input-in-0.consumer.super-stream=true +---- +==== + +The framework will create a super stream named `super`, with 3 partitions. +Up to 3 instances of this application can be deployed. + === 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. @@ -1101,6 +1129,74 @@ IMPORTANT: The correlation data must be provided with a unique `id` so that the You cannot set both `useConfirmHeader` and `confirmAckChannel` but you can still receive returned messages in the error channel when `useConfirmHeader` is true, but using the correlation header is more convenient. +[[rabbitmq-stream-producer]] +=== Initial Producer 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 producer properties described above are not supported when you set the `producerType` property to `STREAM_SYNC` or `STREAM_ASYNC`. + +To configure the binder to use a stream `ProducerType`, Spring Boot will configure an `Environment` `@Bean` from the applicaation properties. +You can, optionally, add a customizer to customize the message handler. + +==== +[source, java] +---- +@Bean +ProducerMessageHandlerCustomizer handlerCustomizer() { + return (hand, dest) -> { + RabbitStreamMessageHandler handler = (RabbitStreamMessageHandler) hand; + handler.setConfirmTimeout(5000); + ((RabbitStreamTemplate) handler.getStreamOperations()).setProducerCustomizer( + (name, builder) -> { + ... + }); + }; +} +---- +==== + +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 producer builder. + +[[rabbitmq-super-stream-producer]] +==== Producer Support for the RabbitMQ Super Streams + +See https://blog.rabbitmq.com/posts/2022/07/rabbitmq-3-11-feature-preview-super-streams[Super Streams] for information about super streams. + +Use of super streams allows for automatic scale-up scale-down with a single active consumer on each partition of a super stream. +Using Spring Cloud Stream, you can publish to a super stream either over AMQP, or using the stream client. + +IMPORTANT: The super stream must already exist; creating a super stream is not supported by producer bindings. + +Publishing to a super stream over AMQP: + +==== +[source, properties] +---- +spring.cloud.stream.bindings.output.destination=super +spring.cloud.stream.bindings.output.producer.partition-count=3 +spring.cloud.stream.bindings.output.producer.partition-key-expression=headers['cust-no'] +spring.cloud.stream.rabbit.bindings.output.producer.declare-exchange=false +---- +==== + +Publishing to a super stream using the stream client: + +==== +[source, properties] +---- +spring.cloud.stream.bindings.output.destination=super +spring.cloud.stream.bindings.output.producer.partition-count=3 +spring.cloud.stream.bindings.output.producer.partition-key-expression=headers['cust-no'] +spring.cloud.stream.rabbit.bindings.output.producer.producer-type=stream-async +spring.cloud.stream.rabbit.bindings.output.producer.super-stream=true +spring.cloud.stream.rabbit.bindings.output.producer.declare-exchange=false +---- +==== + +When using the stream client, if you set a `confirmAckChannel`, a copy of a successfully sent message will be sent to that channel. + == Using Existing Queues/Exchanges By default, the binder will automatically provision a topic exchange with the name being derived from the value of the destination binding property ``. @@ -1242,41 +1338,6 @@ For negatively acknowledged confirmations, the payload is a `NackedAmqpMessageEx There is no automatic handling of these exceptions (such as sending to a <>). You can consume these exceptions with your own Spring Integration flow. -[[rabbitmq-stream-producer]] -=== Initial Producer 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 producer properties described above are not supported when you set the `producerType` property to `STREAM_SYNC` or `STREAM_ASYNC`. - -To configure the binder to use a stream `ProducerType`, you must add an `Environment` `@Bean` and, optionally, a customizer to customize the message handler. - -==== -[source, java] ----- -@Bean -Environment streamEnv() { - return Environment.builder() - .build(); -} - -@Bean -ProducerMessageHandlerCustomizer handlerCustomizer() { - return (hand, dest) -> { - RabbitStreamMessageHandler handler = (RabbitStreamMessageHandler) hand; - handler.setConfirmTimeout(5000); - ((RabbitStreamTemplate) handler.getStreamOperations()).setProducerCustomizer( - (name, builder) -> { - ... - }); - }; -} ----- -==== - -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 producer builder. -======= [[rabbit-binder-health-indicator]] == Rabbit Binder Health Indicator