diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java index f4cf247d3..e84bc3676 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitCommonProperties.java @@ -30,7 +30,7 @@ public abstract class RabbitCommonProperties { public static final String DEAD_LETTER_EXCHANGE = "DLX"; /** - * type of exchange to declare (if necessary, and declareExchange is true). + * type of exchange to declare (if necessary, and declareExchange is true) */ private String exchangeType = ExchangeTypes.TOPIC; @@ -39,6 +39,16 @@ public abstract class RabbitCommonProperties { */ private boolean declareExchange = true; + /** + * whether to declare the exchange as durable + */ + private boolean exchangeDurable = true; + + /** + * whether to declare the exchange as auto-delete + */ + private boolean exchangeAutoDelete = false; + /** * whether a delayed message exchange should be used */ @@ -155,6 +165,22 @@ public abstract class RabbitCommonProperties { this.declareExchange = declareExchange; } + public boolean isExchangeDurable() { + return this.exchangeDurable; + } + + public void setExchangeDurable(boolean exchangeDurable) { + this.exchangeDurable = exchangeDurable; + } + + public boolean isExchangeAutoDelete() { + return this.exchangeAutoDelete; + } + + public void setExchangeAutoDelete(boolean exchangeAutoDelete) { + this.exchangeAutoDelete = exchangeAutoDelete; + } + public boolean isDelayedExchange() { return this.delayedExchange; } diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java index d0e66d9d5..5d22aa7f5 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java @@ -391,6 +391,12 @@ public class RabbitExchangeQueueProvisioner implements ProvisioningProvider` will be appended. ++ +Default: `#`. 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. + @@ -158,11 +163,14 @@ durableSubscription:: 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. +exchangeAutoDelete:: + If `declareExchange` is true, whether the exchange should be auto-delete (removed after the last queue is removed). + -Default: `#`. +Default: `true`. +exchangeDurable:: + If `declareExchange` is true, whether the exchange should be durable (survives broker restart). ++ +Default: `true`. exchangeType:: The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations. + @@ -251,6 +259,12 @@ batchBufferLimit:: Default: `10000`. batchTimeout:: Default: `5000`. +bindingRoutingKey:: + 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 and then only to those groups. ++ +Default: `#`. 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. @@ -327,12 +341,14 @@ dlqTtl:: Only applies if `requiredGroups` are provided and then only to those groups. + Default: `no limit` -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 and then only to those groups. +exchangeAutoDelete:: + If `declareExchange` is true, whether the exchange should be auto-delete (removed after the last queue is removed). + -Default: `#`. +Default: `true`. +exchangeDurable:: + If `declareExchange` is true, whether the exchange should be durable (survives broker restart). ++ +Default: `true`. exchangeType:: The exchange type; `direct`, `fanout` or `topic` for non-partitioned destinations; `direct` or `topic` for partitioned destinations. + 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 86fda69de..925c6c859 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 @@ -271,6 +271,8 @@ public class RabbitBinderTests extends exchange = rmt.getExchange("propsUser2"); } assertThat(exchange).isInstanceOf(DirectExchange.class); + assertThat(exchange.isDurable()).isEqualTo(true); + assertThat(exchange.isAutoDelete()).isEqualTo(false); } @Test @@ -279,6 +281,8 @@ public class RabbitBinderTests extends ExtendedConsumerProperties properties = createConsumerProperties(); RabbitConsumerProperties extProps = properties.getExtension(); extProps.setExchangeType(ExchangeTypes.DIRECT); + extProps.setExchangeDurable(false); + extProps.setExchangeAutoDelete(true); extProps.setBindingRoutingKey("foo"); extProps.setExpires(30_000); extProps.setMaxLength(10_000); @@ -324,6 +328,8 @@ public class RabbitBinderTests extends exchange = rmt.getExchange("propsUser3"); } assertThat(exchange).isInstanceOf(DirectExchange.class); + assertThat(exchange.isDurable()).isEqualTo(false); + assertThat(exchange.isAutoDelete()).isEqualTo(true); // Queue queue = rmt.getQueue("propsUser3"); AMQP-698 QueueInfo queue = rmt.getClient().getQueue("/", "propsUser3.infra"); 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 1e5b758eb..82a952f11 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 @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-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. @@ -101,7 +101,7 @@ public class RabbitTestBinder extends AbstractTestBinder binder = binderFactory.getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor @@ -107,7 +107,7 @@ public class RabbitBinderModuleTests { "--spring.cloud.stream.rabbit.bindings.input.consumer.transacted=true", "--spring.cloud.stream.rabbit.bindings.output.producer.transacted=true"); BinderFactory binderFactory = context.getBean(BinderFactory.class); - Binder binder = binderFactory.getBinder(null, MessageChannel.class); + Binder binder = binderFactory.getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); BindingService bindingService = context.getBean(BindingService.class); DirectFieldAccessor channelBindingServiceAccessor = new DirectFieldAccessor(bindingService); @@ -144,7 +144,7 @@ public class RabbitBinderModuleTests { context = new SpringApplication(SimpleProcessor.class, ConnectionFactoryConfiguration.class) .run("--server.port=0"); BinderFactory binderFactory = context.getBean(BinderFactory.class); - Binder binder = binderFactory.getBinder(null, MessageChannel.class); + Binder binder = binderFactory.getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor @@ -174,7 +174,7 @@ public class RabbitBinderModuleTests { params.add("--server.port=0"); context = SpringApplication.run(SimpleProcessor.class, params.toArray(new String[params.size()])); BinderFactory binderFactory = context.getBean(BinderFactory.class); - Binder binder = binderFactory.getBinder(null, MessageChannel.class); + Binder binder = binderFactory.getBinder(null, MessageChannel.class); assertThat(binder).isInstanceOf(RabbitMessageChannelBinder.class); DirectFieldAccessor binderFieldAccessor = new DirectFieldAccessor(binder); ConnectionFactory binderConnectionFactory = (ConnectionFactory) binderFieldAccessor