From b654afa5517058eb55d73a901354f0ea4b787cde Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 12 Jan 2017 15:43:19 -0500 Subject: [PATCH] GH-34: Support User Infrastructure and Delayed Ex. Resolves #34 Resolves #26 - Add properties to support user configuration of exchanges/queues -- Disable binding and exchange declaration, giving complete control -- Allow the exchange type to be specified and, for non-partitioned destinations, the routing key. - Add support for the Delayed Message Exchange broker plugin Doc Polishing Polishing; add delay header; add outbound routingKey config --- .../src/main/asciidoc/overview.adoc | 58 ++++++- .../binder/rabbit/RabbitCommonProperties.java | 93 ++++++++++ .../rabbit/RabbitConsumerProperties.java | 5 +- .../rabbit/RabbitMessageChannelBinder.java | 163 ++++++++++++++---- .../rabbit/RabbitProducerProperties.java | 30 +++- .../binder/rabbit/RabbitBinderTests.java | 100 ++++++++++- .../binder/rabbit/RabbitTestBinder.java | 7 + 7 files changed, 406 insertions(+), 50 deletions(-) create mode 100644 spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java diff --git a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc index de2ea4bef..4d9dacdeb 100644 --- a/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc +++ b/spring-cloud-stream-binder-rabbit-docs/src/main/asciidoc/overview.adoc @@ -100,11 +100,33 @@ autoBindDlq:: Whether to automatically declare the DLQ and bind it to the binder DLX. + Default: `false`. +bindQueue:: + Whether to bind the queue to the destination exchange; set to `false` if you have set up your own infrastructure and have previously created/bound the queue. ++ +Default: `true`. +declareExchange:: + Whether to declare the exchange for the destination. ++ +Default: `true`. +delayedExchange:: + Whether to declare the exchange as a `Delayed Message Exchange` - requires the delayed message exchange plugin on the broker. + The `x-delayed-type` argument is set to the `exchangeType`. ++ +Default: `false`. durableSubscription:: Whether subscription should be durable. Only effective if `group` is also set. + Default: `true`. +bindingRoutingKey:: + The routing key with which to bind the queue to the exchange (if `bindQueue` is `true`). + for partitioned destinations `-` will be appended. ++ +Default: `#`. +exchangeType:: + The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations. ++ +Default: `topic`. maxConcurrency:: Default: `1`. prefetch:: @@ -167,18 +189,42 @@ batchBufferLimit:: Default: `10000`. batchTimeout:: Default: `5000`. +bindQueue:: + Whether to bind the queue to the destination exchange; set to `false` if you have set up your own infrastructure and have previously created/bound the queue. + Only applies if `requiredGroups` are provided and then only to those groups. ++ +Default: `true`. compress:: Whether data should be compressed when sent. + Default: `false`. -transacted:: - Whether to use transacted channels. +declareExchange:: + Whether to declare the exchange for the destination. ++ +Default: `true`. +delay:: + A SpEL expression to evaluate the delay to apply to the message (`x-delay` header) - has no effect if the exchange is not a delayed message exchange. ++ +Default: No `x-delay` header is set. +delayedExchange:: + Whether to declare the exchange as a `Delayed Message Exchange` - requires the delayed message exchange plugin on the broker. + The `x-delayed-type` argument is set to the `exchangeType`. + Default: `false`. deliveryMode:: Delivery mode. + Default: `PERSISTENT`. +exchangeRoutingKey:: + The routing key with which to bind the queue to the exchange (if `bindQueue` is `true`). + Only applies to non-partitioned destinations. + Only applies if `requiredGroups` are provided. ++ +Default: `#`. +exchangeType:: + The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations. ++ +Default: `topic`. prefix:: A prefix to be added to the name of the `destination` exchange. + @@ -191,6 +237,14 @@ replyHeaderPatterns:: The reply headers to be transported. + Default: `[STANDARD_REPLY_HEADERS,'*']`. +routingKeyExpression:: + A SpEL expression to determine the routing key to use when publishing messages. ++ +Default: `destination` or `destination-` for partitioned destinations. +transacted:: + Whether to use transacted channels. ++ +Default: `false`. [NOTE] ==== diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java new file mode 100644 index 000000000..a2426d511 --- /dev/null +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java @@ -0,0 +1,93 @@ +/* + * Copyright 2017 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 + * + * http://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 org.springframework.amqp.core.ExchangeTypes; + +/** + * @author Gary Russell + * @since 1.2 + * + */ +public abstract class RabbitCommonProperties { + + /** + * type of exchange to declare (if necessary, and declareExchange is true). + */ + private String exchangeType = ExchangeTypes.TOPIC; + + /** + * whether to declare the exchange + */ + private boolean declareExchange = true; + + /** + * whether a delayed message exchange should be used + */ + private boolean delayedExchange = false; + + /** + * whether to bind a queue (or queues when partitioned) to the exchange + */ + private boolean bindQueue = true; + + /** + * The routing key to bind (default # for non-partitioned, destination-instanceIndex for partitioned) + */ + private String bindingRoutingKey; + + public String getExchangeType() { + return this.exchangeType; + } + + public void setExchangeType(String exchangeType) { + this.exchangeType = exchangeType; + } + + public boolean isDeclareExchange() { + return this.declareExchange; + } + + public void setDeclareExchange(boolean declareExchange) { + this.declareExchange = declareExchange; + } + + public boolean isDelayedExchange() { + return this.delayedExchange; + } + + public void setDelayedExchange(boolean delayedExchange) { + this.delayedExchange = delayedExchange; + } + + public boolean isBindQueue() { + return this.bindQueue; + } + + public void setBindQueue(boolean bindQueue) { + this.bindQueue = bindQueue; + } + + public String getBindingRoutingKey() { + return this.bindingRoutingKey; + } + + public void setBindingRoutingKey(String routingKey) { + this.bindingRoutingKey = routingKey; + } + +} diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java index 53f9885eb..b786cb583 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -23,8 +23,9 @@ import org.springframework.util.Assert; /** * @author Marius Bogoevici + * @author Gary Russell */ -public class RabbitConsumerProperties { +public class RabbitConsumerProperties extends RabbitCommonProperties { private String prefix = ""; 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 4423a8135..b8f1518f3 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 @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-2017 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,17 +16,18 @@ package org.springframework.cloud.stream.binder.rabbit; +import java.lang.reflect.Constructor; import java.util.Arrays; import java.util.HashMap; import java.util.Map; -import org.aopalliance.aop.Advice; - import org.springframework.amqp.AmqpConnectException; import org.springframework.amqp.core.AnonymousQueue; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Exchange; +import org.springframework.amqp.core.ExchangeBuilder; +import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; import org.springframework.amqp.core.Queue; @@ -67,6 +68,7 @@ import org.springframework.messaging.MessageHandler; import org.springframework.retry.interceptor.RetryOperationsInterceptor; import org.springframework.scheduling.TaskScheduler; import org.springframework.util.Assert; +import org.springframework.util.ReflectionUtils; import org.springframework.util.StringUtils; import com.rabbitmq.client.AMQP; @@ -84,7 +86,7 @@ import com.rabbitmq.client.Envelope; */ public class RabbitMessageChannelBinder extends AbstractMessageChannelBinder, - ExtendedProducerProperties, Queue, TopicExchange> + ExtendedProducerProperties, Queue, Exchange> implements ExtendedPropertiesBinder { private static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR @@ -228,7 +230,7 @@ public class RabbitMessageChannelBinder .recoverer(determineRecoverer(baseQueueName, properties.getExtension().getPrefix(), properties.getExtension().isRepublishToDlq())) .build(); - listenerContainer.setAdviceChain(new Advice[] {retryInterceptor}); + listenerContainer.setAdviceChain(retryInterceptor); } listenerContainer.setAfterReceivePostProcessors(this.decompressingPostProcessor); listenerContainer.setMessagePropertiesConverter(RabbitMessageChannelBinder.inboundMessagePropertiesConverter); @@ -262,9 +264,10 @@ public class RabbitMessageChannelBinder } String prefix = properties.getExtension().getPrefix(); String exchangeName = applyPrefix(prefix, name); - TopicExchange exchange = new TopicExchange(exchangeName); - declareExchange(exchangeName, exchange); - + Exchange exchange = buildExchange(properties.getExtension(), exchangeName); + if (properties.getExtension().isDeclareExchange()) { + declareExchange(exchangeName, exchange); + } String queueName = applyPrefix(prefix, baseQueueName); boolean partitioned = !anonymous && properties.isPartitioned(); boolean durable = !anonymous && properties.getExtension().isDurableSubscription(); @@ -288,12 +291,8 @@ public class RabbitMessageChannelBinder } } declareQueue(queueName, queue); - if (partitioned) { - String bindingKey = String.format("%s-%d", name, properties.getInstanceIndex()); - declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with(bindingKey)); - } - else { - declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange).with("#")); + if (properties.getExtension().isBindQueue()) { + declareConsumerBindings(name, properties, exchange, partitioned, queue); } if (durable) { autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName, @@ -329,16 +328,18 @@ public class RabbitMessageChannelBinder } @Override - protected TopicExchange createProducerDestinationIfNecessary(String name, + protected Exchange createProducerDestinationIfNecessary(String name, ExtendedProducerProperties producerProperties) { String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name); - TopicExchange exchange = new TopicExchange(exchangeName); - declareExchange(exchangeName, exchange); + Exchange exchange = buildExchange(producerProperties.getExtension(), exchangeName); + if (producerProperties.getExtension().isDeclareExchange()) { + declareExchange(exchangeName, exchange); + } return exchange; } @Override - protected MessageHandler createProducerMessageHandler(final TopicExchange exchange, + protected MessageHandler createProducerMessageHandler(final Exchange exchange, ExtendedProducerProperties properties) throws Exception { String prefix = properties.getExtension().getPrefix(); @@ -346,24 +347,38 @@ public class RabbitMessageChannelBinder String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length()); final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(buildRabbitTemplate(properties.getExtension())); endpoint.setExchangeName(exchange.getName()); + RabbitProducerProperties extendedProperties = properties.getExtension(); + String routingKeyExpression = extendedProperties.getRoutingKeyExpression(); if (!properties.isPartitioned()) { - endpoint.setRoutingKey(destination); + if (routingKeyExpression == null) { + endpoint.setRoutingKey(destination); + } + else { + endpoint.setRoutingKeyExpressionString(routingKeyExpression); + } } else { - endpoint.setRoutingKeyExpression(EXPRESSION_PARSER.parseExpression(buildPartitionRoutingExpression( - destination))); + if (routingKeyExpression == null) { + endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(destination)); + } + else { + endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(routingKeyExpression)); + } + } + if (extendedProperties.getDelayExpression() != null) { + endpoint.setDelayExpressionString(extendedProperties.getDelayExpression()); } for (String requiredGroupName : properties.getRequiredGroups()) { String baseQueueName = exchangeName + "." + requiredGroupName; if (!properties.isPartitioned()) { Queue queue = new Queue(baseQueueName, true, false, false, - queueArgs(baseQueueName, prefix, properties.getExtension().isAutoBindDlq())); + queueArgs(baseQueueName, prefix, extendedProperties.isAutoBindDlq())); declareQueue(baseQueueName, queue); - autoBindDLQ(baseQueueName, baseQueueName, properties.getExtension().getPrefix(), - properties.getExtension().isAutoBindDlq()); - org.springframework.amqp.core.Binding binding = BindingBuilder.bind(queue).to(exchange).with( - destination); - declareBinding(baseQueueName, binding); + autoBindDLQ(baseQueueName, baseQueueName, extendedProperties.getPrefix(), + extendedProperties.isAutoBindDlq()); + if (extendedProperties.isBindQueue()) { + notPartitionedBinding(exchange, queue, extendedProperties); + } } else { // if the stream is partitioned, create one queue for each target partition for the default group @@ -371,21 +386,22 @@ public class RabbitMessageChannelBinder String partitionSuffix = "-" + i; String partitionQueueName = baseQueueName + partitionSuffix; Queue queue = new Queue(partitionQueueName, true, false, false, - queueArgs(partitionQueueName, properties.getExtension().getPrefix(), - properties.getExtension().isAutoBindDlq())); + queueArgs(partitionQueueName, extendedProperties.getPrefix(), + extendedProperties.isAutoBindDlq())); declareQueue(queue.getName(), queue); - autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, properties.getExtension().getPrefix(), - properties.getExtension().isAutoBindDlq()); - declareBinding(queue.getName(), BindingBuilder.bind(queue).to(exchange) - .with(destination + partitionSuffix)); + autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, extendedProperties.getPrefix(), + extendedProperties.isAutoBindDlq()); + if (extendedProperties.isBindQueue()) { + partitionedBinding(destination, exchange, queue, extendedProperties, i); + } } } } DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.outboundMapper(); - mapper.setRequestHeaderNames(properties.getExtension().getRequestHeaderPatterns()); - mapper.setReplyHeaderNames(properties.getExtension().getReplyHeaderPatterns()); + mapper.setRequestHeaderNames(extendedProperties.getRequestHeaderPatterns()); + mapper.setReplyHeaderNames(extendedProperties.getReplyHeaderPatterns()); endpoint.setHeaderMapper(mapper); - endpoint.setDefaultDeliveryMode(properties.getExtension().getDeliveryMode()); + endpoint.setDefaultDeliveryMode(extendedProperties.getDeliveryMode()); endpoint.setBeanFactory(this.getBeanFactory()); endpoint.afterPropertiesSet(); return endpoint; @@ -452,6 +468,22 @@ public class RabbitMessageChannelBinder addToAutoDeclareContext(beanName, queue); } + private Exchange buildExchange(RabbitCommonProperties properties, String exchangeName) { + try { + // TODO Make the ctor public in Spring-AMQP - AMQP-695 + Constructor ctor = ExchangeBuilder.class.getDeclaredConstructor(String.class, String.class); + ReflectionUtils.makeAccessible(ctor); + ExchangeBuilder builder = ctor.newInstance(exchangeName, properties.getExchangeType()); + if (properties.isDelayedExchange()) { + builder.delayed(); + } + return builder.build(); + } + catch (Exception e) { + throw new IllegalStateException("Failed to create exchange object", e); + } + } + private void declareExchange(final String rootName, final Exchange exchange) { try { this.rabbitAdmin.declareExchange(exchange); @@ -465,6 +497,65 @@ public class RabbitMessageChannelBinder addToAutoDeclareContext(rootName + ".exchange", exchange); } + private void declareConsumerBindings(String name, ExtendedConsumerProperties properties, + Exchange exchange, boolean partitioned, Queue queue) { + if (partitioned) { + partitionedBinding(name, exchange, queue, properties.getExtension(), properties.getInstanceIndex()); + } + else { + notPartitionedBinding(exchange, queue, properties.getExtension()); + } + } + + private void partitionedBinding(String destination, Exchange exchange, Queue queue, + RabbitCommonProperties extendedProperties, int index) { + String bindingKey = extendedProperties.getBindingRoutingKey(); + if (bindingKey == null) { + bindingKey = destination; + } + bindingKey += "-" + index; + if (exchange instanceof TopicExchange) { + declareBinding(queue.getName(), BindingBuilder.bind(queue) + .to((TopicExchange) exchange) + .with(bindingKey)); + } + else if (exchange instanceof DirectExchange) { + declareBinding(queue.getName(), BindingBuilder.bind(queue) + .to((DirectExchange) exchange) + .with(bindingKey)); + } + else if (exchange instanceof FanoutExchange) { + throw new IllegalStateException("A fanout exchange is not appropriate for partitioned apps"); + } + else { + throw new IllegalStateException("Cannot bind to a " + exchange.getType() + " exchange"); + } + } + + private void notPartitionedBinding(Exchange exchange, Queue queue, RabbitCommonProperties extendedProperties) { + String routingKey = extendedProperties.getBindingRoutingKey(); + if (routingKey == null) { + routingKey = "#"; + } + if (exchange instanceof TopicExchange) { + declareBinding(queue.getName(), BindingBuilder.bind(queue) + .to((TopicExchange) exchange) + .with(routingKey)); + } + else if (exchange instanceof DirectExchange) { + declareBinding(queue.getName(), BindingBuilder.bind(queue) + .to((DirectExchange) exchange) + .with(routingKey)); + } + else if (exchange instanceof FanoutExchange) { + declareBinding(queue.getName(), BindingBuilder.bind(queue) + .to((FanoutExchange) exchange)); + } + else { + throw new IllegalStateException("Cannot bind to a " + exchange.getType() + " exchange"); + } + } + private void declareBinding(String rootName, org.springframework.amqp.core.Binding binding) { try { this.rabbitAdmin.declareBinding(binding); diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java index 940ff586b..25eb37457 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java +++ b/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 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. @@ -24,7 +24,7 @@ import org.springframework.amqp.core.MessageDeliveryMode; * @author Marius Bogoevici * @author Gary Russell */ -public class RabbitProducerProperties { +public class RabbitProducerProperties extends RabbitCommonProperties { private String prefix = ""; @@ -48,6 +48,16 @@ public class RabbitProducerProperties { private String[] replyHeaderPatterns = new String[] {"STANDARD_REPLY_HEADERS", "*"}; + /** + * When using a delayed message exchange, a SpEL expression to determine the delay to apply to messages + */ + private String delayExpression; + + /** + * A custom routing key when publishing messages; default is the destination name; suffixed by "-partition" when partitioned + */ + private String routingKeyExpression; + public String getPrefix() { return prefix; } @@ -139,4 +149,20 @@ public class RabbitProducerProperties { this.transacted = transacted; } + public String getDelayExpression() { + return this.delayExpression; + } + + public void setDelayExpression(String delayExpression) { + this.delayExpression = delayExpression; + } + + public String getRoutingKeyExpression() { + return this.routingKeyExpression; + } + + public void setRoutingKeyExpression(String routingKeyExpression) { + this.routingKeyExpression = routingKeyExpression; + } + } 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 58f97aca8..a5368d861 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 @@ -1,5 +1,5 @@ /* - * Copyright 2013-2016 the original author or authors. + * Copyright 2013-2017 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. @@ -35,10 +35,15 @@ import org.junit.Test; import org.mockito.ArgumentCaptor; import org.springframework.amqp.core.AcknowledgeMode; +import org.springframework.amqp.core.BindingBuilder; +import org.springframework.amqp.core.DirectExchange; +import org.springframework.amqp.core.Exchange; +import org.springframework.amqp.core.ExchangeTypes; import org.springframework.amqp.core.MessageDeliveryMode; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.amqp.rabbit.core.RabbitManagementTemplate; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.amqp.support.AmqpHeaders; @@ -82,7 +87,7 @@ public class RabbitBinderTests extends public static final String TEST_PREFIX = "bindertest."; @Rule - public RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(); + public RabbitTestSupport rabbitAvailableRule = new RabbitTestSupport(true); @Override protected RabbitTestBinder getBinder() { @@ -185,13 +190,81 @@ public class RabbitBinderTests extends assertThat(endpoint.isRunning()).isFalse(); } + @Test + public void testConsumerPropertiesWithUserInfrastructureNoBind() throws Exception { + RabbitAdmin admin = new RabbitAdmin(this.rabbitAvailableRule.getResource()); + Queue queue = new Queue("propsUser1.infra"); + admin.declareQueue(queue); + DirectExchange exchange = new DirectExchange("propsUser1"); + admin.declareExchange(exchange); + admin.declareBinding(BindingBuilder.bind(queue).to(exchange).with("foo")); + + RabbitTestBinder binder = getBinder(); + ExtendedConsumerProperties properties = createConsumerProperties(); + properties.getExtension().setDeclareExchange(false); + properties.getExtension().setBindQueue(false); + + Binding consumerBinding = binder.bindConsumer("propsUser1", "infra", + createBindableChannel("input", new BindingProperties()), properties); + Lifecycle endpoint = extractEndpoint(consumerBinding); + SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer", + SimpleMessageListenerContainer.class); + assertThat(container.isRunning()).isTrue(); + consumerBinding.unbind(); + assertThat(container.isRunning()).isFalse(); + RabbitManagementTemplate rmt = new RabbitManagementTemplate(); + List bindings = rmt.getBindingsForExchange("/", exchange.getName()); + assertThat(bindings.size()).isEqualTo(1); + } + + @Test + public void testConsumerPropertiesWithUserInfrastructureCustomExchangeAndRK() throws Exception { + RabbitTestBinder binder = getBinder(); + ExtendedConsumerProperties properties = createConsumerProperties(); + properties.getExtension().setExchangeType(ExchangeTypes.DIRECT); + properties.getExtension().setBindingRoutingKey("foo"); +// properties.getExtension().setDelayedExchange(true); // requires delayed message exchange plugin; tested locally + + Binding consumerBinding = binder.bindConsumer("propsUser2", "infra", + createBindableChannel("input", new BindingProperties()), properties); + Lifecycle endpoint = extractEndpoint(consumerBinding); + SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint, "messageListenerContainer", + SimpleMessageListenerContainer.class); + assertThat(container.isRunning()).isTrue(); + consumerBinding.unbind(); + assertThat(container.isRunning()).isFalse(); + RabbitManagementTemplate rmt = new RabbitManagementTemplate(); + List bindings = rmt.getBindingsForExchange("/", "propsUser2"); + int n = 0; + while (n++ < 100 && bindings == null || bindings.size() < 1) { + Thread.sleep(100); + bindings = rmt.getBindingsForExchange("/", "propsUser2"); + } + assertThat(bindings.size()).isEqualTo(1); + assertThat(bindings.get(0).getExchange()).isEqualTo("propsUser2"); + assertThat(bindings.get(0).getDestination()).isEqualTo("propsUser2.infra"); + assertThat(bindings.get(0).getRoutingKey()).isEqualTo("foo"); + +// // TODO: AMQP-696 +// // Exchange exchange = rmt.getExchange("propsUser2"); +// ExchangeInfo ei = rmt.getClient().getExchange("/", "propsUser2"); // requires delayed message exchange plugin +// assertThat(ei.getType()).isEqualTo("x-delayed-message"); +// assertThat(ei.getArguments().get("x-delayed-type")).isEqualTo("direct"); + + Exchange exchange = rmt.getExchange("propsUser2"); + while (n++ < 100 && exchange == null) { + Thread.sleep(100); + exchange = rmt.getExchange("propsUser2"); + } + assertThat(exchange).isInstanceOf(DirectExchange.class); + } + @Test public void testProducerProperties() throws Exception { RabbitTestBinder binder = getBinder(); Binding producerBinding = binder.bindProducer("props.0", createBindableChannel("input", new BindingProperties()), createProducerProperties()); - @SuppressWarnings("unchecked") Lifecycle endpoint = extractEndpoint(producerBinding); MessageDeliveryMode mode = TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode", MessageDeliveryMode.class); @@ -214,18 +287,28 @@ public class RabbitBinderTests extends producerProperties.setPartitionSelectorClass(TestPartitionSelectorClass.class); producerProperties.setPartitionCount(1); producerProperties.getExtension().setTransacted(true); + producerProperties.getExtension().setDelayExpression("42"); + producerProperties.setRequiredGroups("prodPropsRequired"); BindingProperties producerBindingProperties = createProducerBindingProperties(producerProperties); - producerBinding = binder.bindProducer("props.0", createBindableChannel("output", producerBindingProperties), + DirectChannel channel = createBindableChannel("output", producerBindingProperties); + producerBinding = binder.bindProducer("props.0", channel, producerProperties); endpoint = extractEndpoint(producerBinding); assertThat(TestUtils.getPropertyValue(endpoint, "routingKeyExpression", SpelExpression.class) .getExpressionString()).isEqualTo("'props.0-' + headers['partition']"); + assertThat(TestUtils.getPropertyValue(endpoint, "delayExpression", SpelExpression.class) + .getExpressionString()).isEqualTo("42"); mode = TestUtils.getPropertyValue(endpoint, "defaultDeliveryMode", MessageDeliveryMode.class); assertThat(mode).isEqualTo(MessageDeliveryMode.NON_PERSISTENT); assertThat(TestUtils.getPropertyValue(endpoint, "amqpTemplate.transactional", Boolean.class)) .isTrue(); verifyFooRequestProducer(endpoint); + channel.send(new GenericMessage<>("foo")); + org.springframework.amqp.core.Message received = new RabbitTemplate(this.rabbitAvailableRule.getResource()) + .receive("foo.props.0.prodPropsRequired-0", 10_000); + assertThat(received).isNotNull(); + assertThat(received.getMessageProperties().getReceivedDelay()).isEqualTo(42); producerBinding.unbind(); assertThat(endpoint.isRunning()).isFalse(); @@ -663,7 +746,6 @@ public class RabbitBinderTests extends @SuppressWarnings("unchecked") @Test public void testBatchingAndCompression() throws Exception { - RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource()); RabbitTestBinder binder = getBinder(); ExtendedProducerProperties producerProperties = createProducerProperties(); producerProperties.getExtension().setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT); @@ -910,20 +992,22 @@ public class RabbitBinderTests extends }; } - private static class TestPartitionKeyExtractorClass implements PartitionKeyExtractorStrategy { + public static class TestPartitionKeyExtractorClass implements PartitionKeyExtractorStrategy { @Override public Object extractKey(Message message) { - return null; + return 0; } + } - private static class TestPartitionSelectorClass implements PartitionSelectorStrategy { + public static class TestPartitionSelectorClass implements PartitionSelectorStrategy { @Override public int selectPartition(Object key, int partitionCount) { return 0; } + } } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java index 4b9e35b81..4aa869af4 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitTestBinder.java @@ -74,6 +74,7 @@ public class RabbitTestBinder extends AbstractTestBinder properties) { this.queues.add(properties.getExtension().getPrefix() + name + ".default"); this.exchanges.add(properties.getExtension().getPrefix() + name); + if (properties.getRequiredGroups() != null) { + for (String group : properties.getRequiredGroups()) { + this.queues.add(properties.getExtension().getPrefix() + name + "." + group); + } + } + this.prefixes.add(properties.getExtension().getPrefix()); return super.bindProducer(name, moduleOutputChannel, properties); }