From d9c404402741f5308f31f93acb1207bfcd41d2fc Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Wed, 1 Feb 2017 19:22:24 -0500 Subject: [PATCH] Provisioning Changes - Separate Rabbit exchange/queue provisioning from the binder using the SPI provided in core - Refactor the common entities needed for Rabbit into a new core module - Refactor binder code to reflect the SPI changes Fixes #40 Addressing PR review comments Addressing PR review comments cleanup - Addressing PR review Using ProvisioningException Addressing PR comments --- pom.xml | 1 + .../pom.xml | 30 ++ .../rabbit/admin}/RabbitAdminException.java | 2 +- .../rabbit/admin}/RabbitBindingCleaner.java | 2 +- .../rabbit/admin}/RabbitManagementUtils.java | 2 +- .../RabbitBinderConfigurationProperties.java | 4 +- .../properties}/RabbitBindingProperties.java | 2 +- .../properties}/RabbitCommonProperties.java | 4 +- .../properties}/RabbitConsumerProperties.java | 2 +- .../RabbitExtendedBindingProperties.java | 2 +- .../properties}/RabbitProducerProperties.java | 2 +- .../RabbitExchangeQueueProvisioner.java | 510 ++++++++++++++++++ spring-cloud-stream-binder-rabbit/pom.xml | 5 + .../rabbit/RabbitMessageChannelBinder.java | 501 +++-------------- ...bbitMessageChannelBinderConfiguration.java | 12 +- .../rabbit/RabbitBinderCleanerTests.java | 3 + .../binder/rabbit/RabbitBinderTests.java | 20 +- .../binder/rabbit/RabbitTestBinder.java | 7 +- 18 files changed, 665 insertions(+), 446 deletions(-) create mode 100644 spring-cloud-stream-binder-rabbit-core/pom.xml rename {spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit => spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin}/RabbitAdminException.java (94%) rename {spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit => spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin}/RabbitBindingCleaner.java (99%) rename {spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit => spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin}/RabbitManagementUtils.java (98%) rename {spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config => spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties}/RabbitBinderConfigurationProperties.java (93%) rename {spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit => spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties}/RabbitBindingProperties.java (94%) rename {spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit => spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties}/RabbitCommonProperties.java (98%) rename {spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit => spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties}/RabbitConsumerProperties.java (98%) rename {spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit => spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties}/RabbitExtendedBindingProperties.java (96%) rename {spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit => spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties}/RabbitProducerProperties.java (98%) create mode 100644 spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java diff --git a/pom.xml b/pom.xml index a222f2343..fe23369ea 100644 --- a/pom.xml +++ b/pom.xml @@ -51,6 +51,7 @@ + spring-cloud-stream-binder-rabbit-core spring-cloud-stream-binder-rabbit spring-cloud-starter-stream-rabbit spring-cloud-stream-binder-rabbit-test-support diff --git a/spring-cloud-stream-binder-rabbit-core/pom.xml b/spring-cloud-stream-binder-rabbit-core/pom.xml new file mode 100644 index 000000000..1686f6739 --- /dev/null +++ b/spring-cloud-stream-binder-rabbit-core/pom.xml @@ -0,0 +1,30 @@ + + + 4.0.0 + + spring-cloud-stream-binder-rabbit-core + jar + spring-cloud-stream-binder-rabbit-core + RabbitMQ binder core + + + org.springframework.cloud + spring-cloud-stream-binder-rabbit-parent + 1.2.0.BUILD-SNAPSHOT + + + + + org.springframework.cloud + spring-cloud-stream + + + org.springframework.boot + spring-boot-starter-amqp + + + org.springframework.integration + spring-integration-amqp + + + diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitAdminException.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitAdminException.java similarity index 94% rename from spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitAdminException.java rename to spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitAdminException.java index 191b2e2e1..a9b69e1f7 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitAdminException.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitAdminException.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.rabbit; +package org.springframework.cloud.stream.binder.rabbit.admin; /** diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingCleaner.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java similarity index 99% rename from spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingCleaner.java rename to spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java index 250439595..1a0b493cf 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingCleaner.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitBindingCleaner.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.rabbit; +package org.springframework.cloud.stream.binder.rabbit.admin; import java.net.URI; import java.util.ArrayList; diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitManagementUtils.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitManagementUtils.java similarity index 98% rename from spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitManagementUtils.java rename to spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitManagementUtils.java index 6ed3dcecf..d44c62419 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitManagementUtils.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/admin/RabbitManagementUtils.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.rabbit; +package org.springframework.cloud.stream.binder.rabbit.admin; import java.net.URI; import java.net.URISyntaxException; diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBinderConfigurationProperties.java similarity index 93% rename from spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java rename to spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBinderConfigurationProperties.java index d49b70073..bb3797638 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/config/RabbitBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBinderConfigurationProperties.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.rabbit.config; +package org.springframework.cloud.stream.binder.rabbit.properties; import org.springframework.boot.context.properties.ConfigurationProperties; @@ -22,7 +22,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; * @author David Turanski */ @ConfigurationProperties(prefix = "spring.cloud.stream.rabbit.binder") -class RabbitBinderConfigurationProperties { +public class RabbitBinderConfigurationProperties { private String[] adminAddresses = new String[0]; diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBindingProperties.java similarity index 94% rename from spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingProperties.java rename to spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBindingProperties.java index 5ed04136d..ec5500d25 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitBindingProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBindingProperties.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.rabbit; +package org.springframework.cloud.stream.binder.rabbit.properties; /** * @author Marius Bogoevici 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-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java similarity index 98% rename from spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java rename to spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java index 0dabe1085..f4cf247d3 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitCommonProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.rabbit; +package org.springframework.cloud.stream.binder.rabbit.properties; import org.hibernate.validator.constraints.Range; @@ -27,6 +27,8 @@ import org.springframework.amqp.core.ExchangeTypes; */ public abstract class RabbitCommonProperties { + public static final String DEAD_LETTER_EXCHANGE = "DLX"; + /** * type of exchange to declare (if necessary, and declareExchange is true). */ 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-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java similarity index 98% rename from spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java rename to spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java index f6d122ca5..ae5999499 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitConsumerProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.rabbit; +package org.springframework.cloud.stream.binder.rabbit.properties; import javax.validation.constraints.Min; diff --git a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExtendedBindingProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitExtendedBindingProperties.java similarity index 96% rename from spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExtendedBindingProperties.java rename to spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitExtendedBindingProperties.java index afc821c4f..1c008bb14 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitExtendedBindingProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitExtendedBindingProperties.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.rabbit; +package org.springframework.cloud.stream.binder.rabbit.properties; import java.util.HashMap; import java.util.Map; 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-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java similarity index 98% rename from spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java rename to spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java index d1d98aada..5a420693d 100644 --- a/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitProducerProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitProducerProperties.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.cloud.stream.binder.rabbit; +package org.springframework.cloud.stream.binder.rabbit.properties; import javax.validation.constraints.Min; 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 new file mode 100644 index 000000000..b196bf331 --- /dev/null +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -0,0 +1,510 @@ +/* + * 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. + * 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.provisioning; + +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.amqp.AmqpConnectException; +import org.springframework.amqp.core.AnonymousQueue; +import org.springframework.amqp.core.Binding; +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.Queue; +import org.springframework.amqp.core.TopicExchange; +import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.core.RabbitAdmin; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; +import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; +import org.springframework.cloud.stream.binder.ExtendedProducerProperties; +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.RabbitProducerProperties; +import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.cloud.stream.provisioning.ProducerDestination; +import org.springframework.cloud.stream.provisioning.ProvisioningException; +import org.springframework.cloud.stream.provisioning.ProvisioningProvider; +import org.springframework.context.support.GenericApplicationContext; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * AMQP implementation for {@link ProvisioningProvider} + * + * @author Soby Chacko + */ +public class RabbitExchangeQueueProvisioner implements ProvisioningProvider, + ExtendedProducerProperties> { + + private static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR + = new AnonymousQueue.Base64UrlNamingStrategy("anonymous."); + + /** + * The delimiter between a group and index when constructing a binder + * consumer/producer. + */ + private static final String GROUP_INDEX_DELIMITER = "."; + + protected final Log logger = LogFactory.getLog(getClass()); + + private final RabbitAdmin rabbitAdmin; + + private final GenericApplicationContext autoDeclareContext = new GenericApplicationContext(); + + public RabbitExchangeQueueProvisioner(ConnectionFactory connectionFactory) { + this.rabbitAdmin = new RabbitAdmin(connectionFactory); + this.autoDeclareContext.refresh(); + this.rabbitAdmin.setApplicationContext(this.autoDeclareContext); + this.rabbitAdmin.setIgnoreDeclarationExceptions(true); + this.rabbitAdmin.afterPropertiesSet(); + } + + @Override + public ProducerDestination provisionProducerDestination(String name, + ExtendedProducerProperties producerProperties) { + final String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name); + Exchange exchange = buildExchange(producerProperties.getExtension(), exchangeName); + if (producerProperties.getExtension().isDeclareExchange()) { + declareExchange(exchangeName, exchange); + } + Binding binding = null; + for (String requiredGroupName : producerProperties.getRequiredGroups()) { + String baseQueueName = exchangeName + "." + requiredGroupName; + if (!producerProperties.isPartitioned()) { + Queue queue = new Queue(baseQueueName, true, false, false, + queueArgs(baseQueueName, producerProperties.getExtension(), false)); + declareQueue(baseQueueName, queue); + autoBindDLQ(baseQueueName, baseQueueName, producerProperties.getExtension()); + if (producerProperties.getExtension().isBindQueue()) { + binding = notPartitionedBinding(exchange, queue, producerProperties.getExtension()); + } + } + else { + // if the stream is partitioned, create one queue for each target partition for the default group + for (int i = 0; i < producerProperties.getPartitionCount(); i++) { + String partitionSuffix = "-" + i; + String partitionQueueName = baseQueueName + partitionSuffix; + Queue queue = new Queue(partitionQueueName, true, false, false, + queueArgs(partitionQueueName, producerProperties.getExtension(), false)); + declareQueue(queue.getName(), queue); + autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, producerProperties.getExtension()); + if (producerProperties.getExtension().isBindQueue()) { + String prefix = producerProperties.getExtension().getPrefix(); + String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length()); + binding = partitionedBinding(destination, exchange, queue, producerProperties.getExtension(), i); + } + } + } + } + return new RabbitProducerDestination(exchange, binding); + } + + @Override + public ConsumerDestination provisionConsumerDestination(String name, String group, ExtendedConsumerProperties properties) { + boolean anonymous = !StringUtils.hasText(group); + String baseQueueName = anonymous ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName()) + : groupedName(name, group); + if (this.logger.isInfoEnabled()) { + this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name); + } + String prefix = properties.getExtension().getPrefix(); + final String exchangeName = applyPrefix(prefix, name); + 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(); + Queue queue; + if (anonymous) { + queue = new Queue(queueName, false, true, true, queueArgs(queueName, properties.getExtension(), false)); + } + else { + if (partitioned) { + String partitionSuffix = "-" + properties.getInstanceIndex(); + queueName += partitionSuffix; + } + if (durable) { + queue = new Queue(queueName, true, false, false, + queueArgs(queueName, properties.getExtension(), false)); + } + else { + queue = new Queue(queueName, false, false, true, + queueArgs(queueName, properties.getExtension(), false)); + } + } + declareQueue(queueName, queue); + Binding binding = null; + if (properties.getExtension().isBindQueue()) { + binding = declareConsumerBindings(name, properties, exchange, partitioned, queue); + } + if (durable) { + autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName, + properties.getExtension()); + } + return new RabbitConsumerDestination(queue, binding); + } + + /** + * Construct a name comprised of the name and group. + * + * @param name the name. + * @param group the group. + * @return the constructed name. + */ + protected final String groupedName(String name, String group) { + return name + GROUP_INDEX_DELIMITER + (StringUtils.hasText(group) ? group : "default"); + } + + private Binding 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) { + Binding binding = BindingBuilder.bind(queue) + .to((TopicExchange) exchange) + .with(bindingKey); + declareBinding(queue.getName(), binding); + return binding; + } + else if (exchange instanceof DirectExchange) { + Binding binding = BindingBuilder.bind(queue) + .to((DirectExchange) exchange) + .with(bindingKey); + declareBinding(queue.getName(), binding); + return binding; + } + else if (exchange instanceof FanoutExchange) { + throw new ProvisioningException("A fanout exchange is not appropriate for partitioned apps"); + } + else { + throw new ProvisioningException("Cannot bind to a " + exchange.getType() + " exchange"); + } + } + + private Binding declareConsumerBindings(String name, ExtendedConsumerProperties properties, + Exchange exchange, boolean partitioned, Queue queue) { + if (partitioned) { + return partitionedBinding(name, exchange, queue, properties.getExtension(), properties.getInstanceIndex()); + } + else { + return notPartitionedBinding(exchange, queue, properties.getExtension()); + } + } + + private Binding notPartitionedBinding(Exchange exchange, Queue queue, RabbitCommonProperties extendedProperties) { + String routingKey = extendedProperties.getBindingRoutingKey(); + if (routingKey == null) { + routingKey = "#"; + } + if (exchange instanceof TopicExchange) { + Binding binding = BindingBuilder.bind(queue) + .to((TopicExchange) exchange) + .with(routingKey); + declareBinding(queue.getName(), binding); + return binding; + } + else if (exchange instanceof DirectExchange) { + Binding binding = BindingBuilder.bind(queue) + .to((DirectExchange) exchange) + .with(routingKey); + declareBinding(queue.getName(), binding); + return binding; + } + else if (exchange instanceof FanoutExchange) { + Binding binding = BindingBuilder.bind(queue) + .to((FanoutExchange) exchange); + declareBinding(queue.getName(), binding); + return binding; + } + else { + throw new ProvisioningException("Cannot bind to a " + exchange.getType() + " exchange"); + } + } + + /** + * If so requested, declare the DLX/DLQ and bind it. The DLQ is bound to the DLX with a routing key of the original + * queue name because we use default exchange routing by queue name for the original message. + * @param baseQueueName The base name for the queue (including the binder prefix, if any). + * @param routingKey The routing key for the queue. + * @param properties the properties. + */ + private void autoBindDLQ(final String baseQueueName, String routingKey, RabbitCommonProperties properties) { + boolean autoBindDlq = properties.isAutoBindDlq(); + if (this.logger.isDebugEnabled()) { + this.logger.debug("autoBindDLQ=" + autoBindDlq + + " for: " + baseQueueName); + } + if (autoBindDlq) { + String dlqName; + if (properties.getDeadLetterQueueName() == null) { + dlqName = constructDLQName(baseQueueName); + } + else { + dlqName = properties.getDeadLetterQueueName(); + } + Queue dlq = new Queue(dlqName, true, false, false, queueArgs(dlqName, properties, true)); + declareQueue(dlqName, dlq); + String dlxName = deadLetterExchangeName(properties); + final DirectExchange dlx = new DirectExchange(dlxName); + declareExchange(dlxName, dlx); + BindingBuilder.DirectExchangeRoutingKeyConfigurer bindingBuilder = BindingBuilder.bind(dlq).to(dlx); + Binding dlqBinding; + if (properties.getDeadLetterRoutingKey() == null) { + dlqBinding = bindingBuilder.with(routingKey); + } + else { + dlqBinding = bindingBuilder.with(properties.getDeadLetterRoutingKey()); + } + declareBinding(dlqName, dlqBinding); + if (properties instanceof RabbitConsumerProperties && + ((RabbitConsumerProperties) properties).isRepublishToDlq()) { + /* + * Also bind with the base queue name when republishToDlq is used, which does not know about + * partitioning + */ + declareBinding(dlqName, BindingBuilder.bind(dlq).to(dlx).with(baseQueueName)); + } + } + } + + /** + * For binder implementations that support dead lettering, construct the name of the + * dead letter entity for the underlying pipe name. + * + * @param name the name. + */ + public static String constructDLQName(String name) { + return name + ".dlq"; + } + + + private String deadLetterExchangeName(RabbitCommonProperties properties) { + if (properties.getDeadLetterExchange() == null) { + return properties.getPrefix() + RabbitCommonProperties.DEAD_LETTER_EXCHANGE; + } + else { + return properties.getDeadLetterExchange(); + } + } + + + private void declareQueue(String beanName, Queue queue) { + try { + this.rabbitAdmin.declareQueue(queue); + } + catch (AmqpConnectException e) { + if (this.logger.isDebugEnabled()) { + this.logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available"); + } + } + addToAutoDeclareContext(beanName, queue); + } + + private Map queueArgs(String queueName, RabbitCommonProperties properties, boolean isDlq) { + Map args = new HashMap<>(); + if (!isDlq) { + if (properties.isAutoBindDlq()) { + String dlx; + if (properties.getDeadLetterExchange() != null) { + dlx = properties.getDeadLetterExchange(); + } + else { + dlx = applyPrefix(properties.getPrefix(), "DLX"); + } + args.put("x-dead-letter-exchange", dlx); + String dlRk; + if (properties.getDeadLetterRoutingKey() != null) { + dlRk = properties.getDeadLetterRoutingKey(); + } + else { + dlRk = queueName; + } + args.put("x-dead-letter-routing-key", dlRk); + } + additionalArgs(args, properties.getExpires(), properties.getMaxLength(), properties.getMaxLengthBytes(), + properties.getMaxPriority(), properties.getTtl()); + } + else { + if (properties.getDlqDeadLetterExchange() != null) { + args.put("x-dead-letter-exchange", properties.getDlqDeadLetterExchange()); + } + if (properties.getDlqDeadLetterRoutingKey() != null) { + args.put("x-dead-letter-routing-key", properties.getDlqDeadLetterRoutingKey()); + } + additionalArgs(args, properties.getDlqExpires(), properties.getDlqMaxLength(), + properties.getDlqMaxLengthBytes(), properties.getDlqMaxPriority(), properties.getDlqTtl()); + } + return args; + } + + private void additionalArgs(Map args, Integer expires, Integer maxLength, Integer maxLengthBytes, + Integer maxPriority, Integer ttl) { + if (expires != null) { + args.put("x-expires", expires); + } + if (maxLength != null) { + args.put("x-max-length", maxLength); + } + if (maxLengthBytes != null) { + args.put("x-max-length-bytes", maxLengthBytes); + } + if (maxPriority != null) { + args.put("x-max-priority", maxPriority); + } + if (ttl != null) { + args.put("x-message-ttl", ttl); + } + } + + + public static String applyPrefix(String prefix, String name) { + return prefix + name; + } + + private Exchange buildExchange(RabbitCommonProperties properties, String exchangeName) { + try { + ExchangeBuilder builder = new ExchangeBuilder(exchangeName, properties.getExchangeType()); + if (properties.isDelayedExchange()) { + builder.delayed(); + } + return builder.build(); + } + catch (Exception e) { + throw new ProvisioningException("Failed to create exchange object", e); + } + } + + private void declareExchange(final String rootName, final Exchange exchange) { + try { + this.rabbitAdmin.declareExchange(exchange); + } + catch (AmqpConnectException e) { + if (this.logger.isDebugEnabled()) { + this.logger.debug( + "Declaration of exchange: " + exchange.getName() + " deferred - connection not available"); + } + } + addToAutoDeclareContext(rootName + ".exchange", exchange); + } + + private void addToAutoDeclareContext(String name, Object bean) { + synchronized (this.autoDeclareContext) { + if (!this.autoDeclareContext.containsBean(name)) { + this.autoDeclareContext.getBeanFactory().registerSingleton(name, bean); + } + } + } + + private void declareBinding(String rootName, org.springframework.amqp.core.Binding binding) { + try { + this.rabbitAdmin.declareBinding(binding); + } + catch (AmqpConnectException e) { + if (this.logger.isDebugEnabled()) { + this.logger.debug( + "Declaration of binding: " + rootName + ".binding deferred - connection not available"); + } + } + addToAutoDeclareContext(rootName + ".binding", binding); + } + + public void cleanAutoDeclareContext(String name) { + synchronized (this.autoDeclareContext) { + removeSingleton(name + ".binding"); + removeSingleton(name); + String dlq = name + ".dlq"; + removeSingleton(dlq + ".binding"); + removeSingleton(dlq); + } + } + + private void removeSingleton(String name) { + if (this.autoDeclareContext.containsBean(name)) { + ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext.getBeanFactory(); + if (beanFactory instanceof DefaultListableBeanFactory) { + ((DefaultListableBeanFactory) beanFactory).destroySingleton(name); + } + } + } + + private final class RabbitProducerDestination implements ProducerDestination { + + private final Exchange exchange; + private final Binding binding; + + private RabbitProducerDestination(Exchange exchange, Binding binding) { + Assert.notNull(exchange, "exchange must not be null"); + this.exchange = exchange; + this.binding = binding; + } + + @Override + public String getName() { + return this.exchange.getName(); + } + + @Override + public String getNameForPartition(int partition) { + return this.exchange.getName(); + } + + @Override + public String toString() { + return "RabbitProducerDestination{" + + "exchange=" + exchange + + ", binding=" + binding + + '}'; + } + } + + private final class RabbitConsumerDestination implements ConsumerDestination { + + private final Queue queue; + private final Binding binding; + + private RabbitConsumerDestination(Queue queue, Binding binding) { + Assert.notNull(queue, "queue must not be null"); + this.queue = queue; + this.binding = binding; + } + + @Override + public String toString() { + return "RabbitConsumerDestination{" + + "queue=" + queue + + ", binding=" + binding + + '}'; + } + + @Override + public String getName() { + return this.queue.getName(); + } + } +} diff --git a/spring-cloud-stream-binder-rabbit/pom.xml b/spring-cloud-stream-binder-rabbit/pom.xml index 6fa97be97..1cb0ed265 100644 --- a/spring-cloud-stream-binder-rabbit/pom.xml +++ b/spring-cloud-stream-binder-rabbit/pom.xml @@ -14,6 +14,11 @@ + + org.springframework.cloud + spring-cloud-stream-binder-rabbit-core + ${project.version} + org.springframework.boot spring-boot-configuration-processor 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 5b1cf983a..5d067482d 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 @@ -16,29 +16,17 @@ 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.springframework.amqp.AmqpConnectException; -import org.springframework.amqp.core.AnonymousQueue; -import org.springframework.amqp.core.Binding; -import org.springframework.amqp.core.BindingBuilder; -import org.springframework.amqp.core.BindingBuilder.DirectExchangeRoutingKeyConfigurer; -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 com.rabbitmq.client.AMQP; +import com.rabbitmq.client.Envelope; + import org.springframework.amqp.core.MessagePostProcessor; import org.springframework.amqp.core.MessageProperties; -import org.springframework.amqp.core.Queue; -import org.springframework.amqp.core.TopicExchange; import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory; import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate; -import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.amqp.rabbit.core.support.BatchingStrategy; import org.springframework.amqp.rabbit.core.support.SimpleBatchingStrategy; @@ -50,14 +38,18 @@ import org.springframework.amqp.rabbit.support.DefaultMessagePropertiesConverter import org.springframework.amqp.rabbit.support.MessagePropertiesConverter; import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor; import org.springframework.amqp.support.postprocessor.GZipPostProcessor; -import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; -import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.cloud.stream.binder.AbstractMessageChannelBinder; import org.springframework.cloud.stream.binder.ExtendedConsumerProperties; import org.springframework.cloud.stream.binder.ExtendedProducerProperties; import org.springframework.cloud.stream.binder.ExtendedPropertiesBinder; -import org.springframework.context.support.GenericApplicationContext; +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.RabbitExtendedBindingProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; +import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; +import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.cloud.stream.provisioning.ProducerDestination; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint; @@ -70,12 +62,8 @@ 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; -import com.rabbitmq.client.Envelope; - /** * A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by RabbitMQ. * @author Mark Fisher @@ -88,14 +76,9 @@ import com.rabbitmq.client.Envelope; */ public class RabbitMessageChannelBinder extends AbstractMessageChannelBinder, - ExtendedProducerProperties, Queue, Exchange> + ExtendedProducerProperties> implements ExtendedPropertiesBinder { - private static final AnonymousQueue.Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR - = new AnonymousQueue.Base64UrlNamingStrategy("anonymous."); - - private static final String DEAD_LETTER_EXCHANGE = "DLX"; - private static final MessagePropertiesConverter inboundMessagePropertiesConverter = new DefaultMessagePropertiesConverter() { @@ -108,10 +91,6 @@ public class RabbitMessageChannelBinder } }; - private final RabbitAdmin rabbitAdmin; - - private final GenericApplicationContext autoDeclareContext = new GenericApplicationContext(); - private final RabbitProperties rabbitProperties; private ConnectionFactory connectionFactory; @@ -128,17 +107,16 @@ public class RabbitMessageChannelBinder private RabbitExtendedBindingProperties extendedBindingProperties = new RabbitExtendedBindingProperties(); - public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties) { - super(true, new String[0]); + RabbitExchangeQueueProvisioner provisioningProvider; + + public RabbitMessageChannelBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties, + RabbitExchangeQueueProvisioner provisioningProvider) { + super(true, new String[0], provisioningProvider); Assert.notNull(connectionFactory, "connectionFactory must not be null"); Assert.notNull(rabbitProperties, "rabbitProperties must not be null"); this.connectionFactory = connectionFactory; this.rabbitProperties = rabbitProperties; - this.rabbitAdmin = new RabbitAdmin(connectionFactory); - this.autoDeclareContext.refresh(); - this.rabbitAdmin.setApplicationContext(this.autoDeclareContext); - this.rabbitAdmin.setIgnoreDeclarationExceptions(true); - this.rabbitAdmin.afterPropertiesSet(); + this.provisioningProvider = provisioningProvider; } /** @@ -194,21 +172,65 @@ public class RabbitMessageChannelBinder return this.extendedBindingProperties.getExtendedConsumerProperties(channelName); } - @Override public RabbitProducerProperties getExtendedProducerProperties(String channelName) { return this.extendedBindingProperties.getExtendedProducerProperties(channelName); } @Override - protected MessageProducer createConsumerEndpoint(String name, String group, Queue destination, - ExtendedConsumerProperties properties) { + protected MessageHandler createProducerMessageHandler(final ProducerDestination producerDestination, + ExtendedProducerProperties producerProperties) + throws Exception { + String prefix = producerProperties.getExtension().getPrefix(); + String exchangeName = producerDestination.getName(); + String destination = StringUtils.isEmpty(prefix) ? exchangeName : exchangeName.substring(prefix.length()); + final AmqpOutboundEndpoint endpoint = new AmqpOutboundEndpoint(buildRabbitTemplate(producerProperties.getExtension())); + endpoint.setExchangeName(producerDestination.getName()); + RabbitProducerProperties extendedProperties = producerProperties.getExtension(); + String routingKeyExpression = extendedProperties.getRoutingKeyExpression(); + if (!producerProperties.isPartitioned()) { + if (routingKeyExpression == null) { + endpoint.setRoutingKey(destination); + } + else { + endpoint.setRoutingKeyExpressionString(routingKeyExpression); + } + } + else { + if (routingKeyExpression == null) { + endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(destination)); + } + else { + endpoint.setRoutingKeyExpressionString(buildPartitionRoutingExpression(routingKeyExpression)); + } + } + if (extendedProperties.getDelayExpression() != null) { + endpoint.setDelayExpressionString(extendedProperties.getDelayExpression()); + } + DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.outboundMapper(); + mapper.setRequestHeaderNames(extendedProperties.getRequestHeaderPatterns()); + mapper.setReplyHeaderNames(extendedProperties.getReplyHeaderPatterns()); + endpoint.setHeaderMapper(mapper); + endpoint.setDefaultDeliveryMode(extendedProperties.getDeliveryMode()); + endpoint.setBeanFactory(this.getBeanFactory()); + endpoint.afterPropertiesSet(); + return endpoint; + } + + @Override + protected MessageProducer createConsumerEndpoint(ConsumerDestination consumerDestination, String group, + ExtendedConsumerProperties properties) { DirectChannel convertingBridgeChannel = new DirectChannel(); convertingBridgeChannel.setBeanFactory(this.getBeanFactory()); - final String baseQueueName = baseQueueName(name, group); - convertingBridgeChannel.setBeanName(baseQueueName + ".bridge"); + String prefix = properties.getExtension().getPrefix(); + String destination = consumerDestination.getName(); + String prefixStripped = (StringUtils.isEmpty(prefix) || !destination.startsWith(prefix)) ? destination + : destination.substring(prefix.length()); + String baseQueueName = StringUtils.hasText(group) ? prefixStripped.substring(0, prefixStripped.indexOf(group)) + group : prefixStripped; + + convertingBridgeChannel.setBeanName(baseQueueName + ".bridge"); SimpleMessageListenerContainer listenerContainer = new SimpleMessageListenerContainer( this.connectionFactory); listenerContainer.setAcknowledgeMode(properties.getExtension().getAcknowledgeMode()); @@ -224,8 +246,8 @@ public class RabbitMessageChannelBinder listenerContainer.setPrefetchCount(properties.getExtension().getPrefetch()); listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval()); listenerContainer.setTxSize(properties.getExtension().getTxSize()); - listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(destination.getName() + "-")); - listenerContainer.setQueues(destination); + listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(consumerDestination.getName() + "-")); + listenerContainer.setQueueNames(consumerDestination.getName()); if (properties.getMaxAttempts() > 1 || properties.getExtension().isRepublishToDlq()) { RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless() .retryOperations(buildRetryTemplate(properties)) @@ -249,63 +271,19 @@ public class RabbitMessageChannelBinder return adapter; } - @Override - protected void afterUnbindConsumer(String name, String group, - ExtendedConsumerProperties consumerProperties) { - cleanAutoDeclareContext(consumerProperties.getExtension().getPrefix(), baseQueueName(name, group)); - } - - @Override - protected Queue createConsumerDestinationIfNecessary(String name, String group, - ExtendedConsumerProperties properties) { - boolean anonymous = !StringUtils.hasText(group); - String baseQueueName = anonymous ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName()) - : groupedName(name, group); - if (this.logger.isInfoEnabled()) { - this.logger.info("declaring queue for inbound: " + baseQueueName + ", bound to: " + name); - } - String prefix = properties.getExtension().getPrefix(); - String exchangeName = applyPrefix(prefix, name); - 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(); - Queue queue; - - if (anonymous) { - queue = new Queue(queueName, false, true, true, queueArgs(queueName, properties.getExtension(), false)); + private String deadLetterExchangeName(RabbitCommonProperties properties) { + if (properties.getDeadLetterExchange() == null) { + return properties.getPrefix() + RabbitCommonProperties.DEAD_LETTER_EXCHANGE; } else { - if (partitioned) { - String partitionSuffix = "-" + properties.getInstanceIndex(); - queueName += partitionSuffix; - } - if (durable) { - queue = new Queue(queueName, true, false, false, - queueArgs(queueName, properties.getExtension(), false)); - } - else { - queue = new Queue(queueName, false, false, true, - queueArgs(queueName, properties.getExtension(), false)); - } + return properties.getDeadLetterExchange(); } - declareQueue(queueName, queue); - if (properties.getExtension().isBindQueue()) { - declareConsumerBindings(name, properties, exchange, partitioned, queue); - } - if (durable) { - autoBindDLQ(applyPrefix(properties.getExtension().getPrefix(), baseQueueName), queueName, - properties.getExtension()); - } - return queue; } - private String baseQueueName(String name, String group) { - return !StringUtils.hasText(group) ? groupedName(name, ANONYMOUS_GROUP_NAME_GENERATOR.generateName()) - : groupedName(name, group); + @Override + protected void afterUnbindConsumer(ConsumerDestination consumerDestination, String group, + ExtendedConsumerProperties consumerProperties) { + provisioningProvider.cleanAutoDeclareContext(consumerDestination.getName()); } private MessageRecoverer determineRecoverer(String name, RabbitCommonProperties properties, boolean republish) { @@ -320,139 +298,6 @@ public class RabbitMessageChannelBinder } } - @Override - protected Exchange createProducerDestinationIfNecessary(String name, - ExtendedProducerProperties producerProperties) { - String exchangeName = applyPrefix(producerProperties.getExtension().getPrefix(), name); - Exchange exchange = buildExchange(producerProperties.getExtension(), exchangeName); - if (producerProperties.getExtension().isDeclareExchange()) { - declareExchange(exchangeName, exchange); - } - return exchange; - } - - @Override - protected MessageHandler createProducerMessageHandler(final Exchange exchange, - ExtendedProducerProperties properties) - throws Exception { - String prefix = properties.getExtension().getPrefix(); - String exchangeName = exchange.getName(); - 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()) { - if (routingKeyExpression == null) { - endpoint.setRoutingKey(destination); - } - else { - endpoint.setRoutingKeyExpressionString(routingKeyExpression); - } - } - else { - 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, extendedProperties, false)); - declareQueue(baseQueueName, queue); - autoBindDLQ(baseQueueName, baseQueueName, extendedProperties); - if (extendedProperties.isBindQueue()) { - notPartitionedBinding(exchange, queue, extendedProperties); - } - } - else { - // if the stream is partitioned, create one queue for each target partition for the default group - for (int i = 0; i < properties.getPartitionCount(); i++) { - String partitionSuffix = "-" + i; - String partitionQueueName = baseQueueName + partitionSuffix; - Queue queue = new Queue(partitionQueueName, true, false, false, - queueArgs(partitionQueueName, extendedProperties, false)); - declareQueue(queue.getName(), queue); - autoBindDLQ(baseQueueName, baseQueueName + partitionSuffix, extendedProperties); - if (extendedProperties.isBindQueue()) { - partitionedBinding(destination, exchange, queue, extendedProperties, i); - } - } - } - } - DefaultAmqpHeaderMapper mapper = DefaultAmqpHeaderMapper.outboundMapper(); - mapper.setRequestHeaderNames(extendedProperties.getRequestHeaderPatterns()); - mapper.setReplyHeaderNames(extendedProperties.getReplyHeaderPatterns()); - endpoint.setHeaderMapper(mapper); - endpoint.setDefaultDeliveryMode(extendedProperties.getDeliveryMode()); - endpoint.setBeanFactory(this.getBeanFactory()); - endpoint.afterPropertiesSet(); - return endpoint; - } - - private Map queueArgs(String queueName, RabbitCommonProperties properties, boolean isDlq) { - Map args = new HashMap<>(); - if (!isDlq) { - if (properties.isAutoBindDlq()) { - String dlx; - if (properties.getDeadLetterExchange() != null) { - dlx = properties.getDeadLetterExchange(); - } - else { - dlx = applyPrefix(properties.getPrefix(), "DLX"); - } - args.put("x-dead-letter-exchange", dlx); - String dlRk; - if (properties.getDeadLetterRoutingKey() != null) { - dlRk = properties.getDeadLetterRoutingKey(); - } - else { - dlRk = queueName; - } - args.put("x-dead-letter-routing-key", dlRk); - } - additionalArgs(args, properties.getExpires(), properties.getMaxLength(), properties.getMaxLengthBytes(), - properties.getMaxPriority(), properties.getTtl()); - } - else { - if (properties.getDlqDeadLetterExchange() != null) { - args.put("x-dead-letter-exchange", properties.getDlqDeadLetterExchange()); - } - if (properties.getDlqDeadLetterRoutingKey() != null) { - args.put("x-dead-letter-routing-key", properties.getDlqDeadLetterRoutingKey()); - } - additionalArgs(args, properties.getDlqExpires(), properties.getDlqMaxLength(), - properties.getDlqMaxLengthBytes(), properties.getDlqMaxPriority(), properties.getDlqTtl()); - } - return args; - } - - private void additionalArgs(Map args, Integer expires, Integer maxLength, Integer maxLengthBytes, - Integer maxPriority, Integer ttl) { - if (expires != null) { - args.put("x-expires", expires); - } - if (maxLength != null) { - args.put("x-max-length", maxLength); - } - if (maxLengthBytes != null) { - args.put("x-max-length-bytes", maxLengthBytes); - } - if (maxPriority != null) { - args.put("x-max-priority", maxPriority); - } - if (ttl != null) { - args.put("x-message-ttl", ttl); - } - } - private RabbitTemplate buildRabbitTemplate(RabbitProducerProperties properties) { RabbitTemplate rabbitTemplate; if (properties.isBatchingEnabled()) { @@ -476,198 +321,4 @@ public class RabbitMessageChannelBinder return rabbitTemplate; } - /** - * If so requested, declare the DLX/DLQ and bind it. The DLQ is bound to the DLX with a routing key of the original - * queue name because we use default exchange routing by queue name for the original message. - * @param baseQueueName The base name for the queue (including the binder prefix, if any). - * @param routingKey The routing key for the queue. - * @param properties the properties. - */ - private void autoBindDLQ(final String baseQueueName, String routingKey, RabbitCommonProperties properties) { - boolean autoBindDlq = properties.isAutoBindDlq(); - if (this.logger.isDebugEnabled()) { - this.logger.debug("autoBindDLQ=" + autoBindDlq - + " for: " + baseQueueName); - } - if (autoBindDlq) { - String dlqName; - if (properties.getDeadLetterQueueName() == null) { - dlqName = constructDLQName(baseQueueName); - } - else { - dlqName = properties.getDeadLetterQueueName(); - } - Queue dlq = new Queue(dlqName, true, false, false, queueArgs(dlqName, properties, true)); - declareQueue(dlqName, dlq); - String dlxName = deadLetterExchangeName(properties); - final DirectExchange dlx = new DirectExchange(dlxName); - declareExchange(dlxName, dlx); - DirectExchangeRoutingKeyConfigurer bindingBuilder = BindingBuilder.bind(dlq).to(dlx); - Binding dlqBinding; - if (properties.getDeadLetterRoutingKey() == null) { - dlqBinding = bindingBuilder.with(routingKey); - } - else { - dlqBinding = bindingBuilder.with(properties.getDeadLetterRoutingKey()); - } - declareBinding(dlqName, dlqBinding); - if (properties instanceof RabbitConsumerProperties && - ((RabbitConsumerProperties) properties).isRepublishToDlq()) { - /* - * Also bind with the base queue name when republishToDlq is used, which does not know about - * partitioning - */ - declareBinding(dlqName, BindingBuilder.bind(dlq).to(dlx).with(baseQueueName)); - } - } - } - - private void declareQueue(String beanName, Queue queue) { - try { - this.rabbitAdmin.declareQueue(queue); - } - catch (AmqpConnectException e) { - if (this.logger.isDebugEnabled()) { - this.logger.debug("Declaration of queue: " + queue.getName() + " deferred - connection not available"); - } - } - 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); - } - catch (AmqpConnectException e) { - if (this.logger.isDebugEnabled()) { - this.logger.debug( - "Declaration of exchange: " + exchange.getName() + " deferred - connection not available"); - } - } - 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); - } - catch (AmqpConnectException e) { - if (this.logger.isDebugEnabled()) { - this.logger.debug( - "Declaration of binding: " + rootName + ".binding deferred - connection not available"); - } - } - addToAutoDeclareContext(rootName + ".binding", binding); - } - - private String deadLetterExchangeName(RabbitCommonProperties properties) { - if (properties.getDeadLetterExchange() == null) { - return properties.getPrefix() + DEAD_LETTER_EXCHANGE; - } - else { - return properties.getDeadLetterExchange(); - } - } - - private void addToAutoDeclareContext(String name, Object bean) { - synchronized (this.autoDeclareContext) { - if (!this.autoDeclareContext.containsBean(name)) { - this.autoDeclareContext.getBeanFactory().registerSingleton(name, bean); - } - } - } - - private void cleanAutoDeclareContext(String prefix, String name) { - synchronized (this.autoDeclareContext) { - removeSingleton(applyPrefix(prefix, name) + ".binding"); - removeSingleton(applyPrefix(prefix, name)); - String dlq = applyPrefix(prefix, name) + ".dlq"; - removeSingleton(dlq + ".binding"); - removeSingleton(dlq); - } - } - - private void removeSingleton(String name) { - if (this.autoDeclareContext.containsBean(name)) { - ConfigurableListableBeanFactory beanFactory = this.autoDeclareContext.getBeanFactory(); - if (beanFactory instanceof DefaultListableBeanFactory) { - ((DefaultListableBeanFactory) beanFactory).destroySingleton(name); - } - } - } } 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 47864d6c6..d2b4f543e 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,8 +24,10 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.PropertyPlaceholderAutoConfiguration; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.stream.binder.rabbit.RabbitExtendedBindingProperties; import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitBinderConfigurationProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitExtendedBindingProperties; +import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; import org.springframework.cloud.stream.config.codec.kryo.KryoCodecAutoConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -61,7 +63,8 @@ public class RabbitMessageChannelBinderConfiguration { @Bean RabbitMessageChannelBinder rabbitMessageChannelBinder() { - RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(rabbitConnectionFactory, rabbitProperties); + RabbitMessageChannelBinder binder = new RabbitMessageChannelBinder(rabbitConnectionFactory, rabbitProperties, + provisioningProvider()); binder.setCodec(codec); binder.setAdminAddresses(rabbitBinderConfigurationProperties.getAdminAddresses()); binder.setCompressingPostProcessor(gZipPostProcessor()); @@ -82,5 +85,10 @@ public class RabbitMessageChannelBinderConfiguration { gZipPostProcessor.setLevel(rabbitBinderConfigurationProperties.getCompressionLevel()); return gZipPostProcessor; } + + @Bean + RabbitExchangeQueueProvisioner provisioningProvider() { + return new RabbitExchangeQueueProvisioner(rabbitConnectionFactory); + } } diff --git a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java index 4e024bf5c..ece3cb541 100644 --- a/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java +++ b/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderCleanerTests.java @@ -35,6 +35,9 @@ import org.springframework.amqp.rabbit.core.ChannelCallback; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.cloud.stream.binder.AbstractBinder; +import org.springframework.cloud.stream.binder.rabbit.admin.RabbitAdminException; +import org.springframework.cloud.stream.binder.rabbit.admin.RabbitBindingCleaner; +import org.springframework.cloud.stream.binder.rabbit.admin.RabbitManagementUtils; import org.springframework.cloud.stream.binder.test.junit.rabbit.RabbitTestSupport; import org.springframework.web.client.RestTemplate; import org.springframework.web.util.UriComponentsBuilder; 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 bed837197..d1c766086 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 @@ -16,11 +16,6 @@ package org.springframework.cloud.stream.binder.rabbit; -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.Mockito.spy; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - import java.util.HashMap; import java.util.List; import java.util.Map; @@ -28,6 +23,7 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.zip.Deflater; +import com.rabbitmq.http.client.domain.QueueInfo; import org.aopalliance.aop.Advice; import org.apache.commons.logging.Log; import org.junit.Rule; @@ -59,6 +55,9 @@ import org.springframework.cloud.stream.binder.PartitionKeyExtractorStrategy; import org.springframework.cloud.stream.binder.PartitionSelectorStrategy; import org.springframework.cloud.stream.binder.PartitionTestSupport; import org.springframework.cloud.stream.binder.Spy; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; +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; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.context.ApplicationContext; @@ -74,7 +73,10 @@ import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.messaging.support.GenericMessage; -import com.rabbitmq.http.client.domain.QueueInfo; +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; /** * @author Mark Fisher @@ -511,7 +513,7 @@ public class RabbitBinderTests extends consumerBinding.unbind(); - ApplicationContext context = TestUtils.getPropertyValue(binder, "binder.autoDeclareContext", + ApplicationContext context = TestUtils.getPropertyValue(binder, "binder.provisioningProvider.autoDeclareContext", ApplicationContext.class); assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default.binding")).isFalse(); assertThat(context.containsBean(TEST_PREFIX + "dlqtest.default")).isFalse(); @@ -896,7 +898,9 @@ public class RabbitBinderTests extends public void testLateBinding() throws Exception { RabbitTestSupport.RabbitProxy proxy = new RabbitTestSupport.RabbitProxy(); CachingConnectionFactory cf = new CachingConnectionFactory("localhost", proxy.getPort()); - RabbitMessageChannelBinder rabbitBinder = new RabbitMessageChannelBinder(cf, new RabbitProperties()); + + RabbitMessageChannelBinder rabbitBinder = new RabbitMessageChannelBinder(cf, new RabbitProperties(), + new RabbitExchangeQueueProvisioner(cf)); RabbitTestBinder binder = new RabbitTestBinder(cf, rabbitBinder); ExtendedProducerProperties producerProperties = createProducerProperties(); 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 3df34fca3..1e5b758eb 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 @@ -26,6 +26,10 @@ import org.springframework.cloud.stream.binder.AbstractTestBinder; 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.properties.RabbitCommonProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitConsumerProperties; +import org.springframework.cloud.stream.binder.rabbit.properties.RabbitProducerProperties; +import org.springframework.cloud.stream.binder.rabbit.provisioning.RabbitExchangeQueueProvisioner; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.codec.kryo.PojoCodec; import org.springframework.integration.context.IntegrationContextUtils; @@ -51,7 +55,8 @@ public class RabbitTestBinder extends AbstractTestBinder exchanges = new HashSet(); public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitProperties rabbitProperties) { - this(connectionFactory, new RabbitMessageChannelBinder(connectionFactory, rabbitProperties)); + this(connectionFactory, new RabbitMessageChannelBinder(connectionFactory, rabbitProperties, + new RabbitExchangeQueueProvisioner(connectionFactory))); } public RabbitTestBinder(ConnectionFactory connectionFactory, RabbitMessageChannelBinder binder) {