SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-212: Configurable anonymous queue name prefix

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/212

Enable configuration of the prefix part of anonymous auto-delete queues.
This commit is contained in:
Gary Russell
2019-04-09 11:34:06 -04:00
committed by Oleg Zhurakousky
parent 442f04dadf
commit 31f02523d7
4 changed files with 49 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
docs/src/main/asciidoc/overview.adoc
View File

@@ -122,6 +122,12 @@ acknowledgeMode::
The acknowledge mode.
+
Default: `AUTO`.
anonymousGroupPrefix::
When the binding has no `group` property, an anonymous, auto-delete queue is bound to the destination exchange.
The default naming stragegy for such queues results in a queue named `anonymous.<base64 representation of a UUID>`.
Set this property to change the prefix to something other than the default.
+
Default: `anonymous.`.
autoBindDlq::
Whether to automatically declare the DLQ and bind it to the binder DLX.
+

13
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitConsumerProperties.java
View File

@@ -120,6 +120,11 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
*/
private ContainerType containerType = ContainerType.SIMPLE;
/**
* Prefix for anonymous queue names (when no group is provided).
*/
private String anonymousGroupPrefix = "anonymous.";
public boolean isTransacted() {
return transacted;
}
@@ -286,4 +291,12 @@ public class RabbitConsumerProperties extends RabbitCommonProperties {
this.containerType = containerType;
}
public String getAnonymousGroupPrefix() {
return this.anonymousGroupPrefix;
}
public void setAnonymousGroupPrefix(String anonymousGroupPrefix) {
this.anonymousGroupPrefix = anonymousGroupPrefix;
}
}

14
spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/provisioning/RabbitExchangeQueueProvisioner.java
View File

@@ -66,8 +66,6 @@ public class RabbitExchangeQueueProvisioner
ProvisioningProvider<ExtendedConsumerProperties<RabbitConsumerProperties>, ExtendedProducerProperties<RabbitProducerProperties>> {
// @checkstyle:on
private static final Base64UrlNamingStrategy ANONYMOUS_GROUP_NAME_GENERATOR = new Base64UrlNamingStrategy(
"anonymous.");
/**
* The delimiter between a group and index when constructing a binder
@@ -163,15 +161,23 @@ public class RabbitExchangeQueueProvisioner
private ConsumerDestination doProvisionConsumerDestination(String name, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> properties) {
boolean anonymous = !StringUtils.hasText(group);
Base64UrlNamingStrategy anonQueueNameGenerator = null;
if (anonymous) {
anonQueueNameGenerator = new Base64UrlNamingStrategy(
properties.getExtension().getAnonymousGroupPrefix() == null
? ""
: properties.getExtension().getAnonymousGroupPrefix());
}
String baseQueueName;
if (properties.getExtension().isQueueNameGroupOnly()) {
baseQueueName = anonymous ? ANONYMOUS_GROUP_NAME_GENERATOR.generateName()
baseQueueName = anonymous ? anonQueueNameGenerator.generateName()
: group;
}
else {
baseQueueName = groupedName(name,
anonymous ? ANONYMOUS_GROUP_NAME_GENERATOR.generateName() : group);
anonymous ? anonQueueNameGenerator.generateName() : group);
}
if (this.logger.isInfoEnabled()) {
this.logger.info("declaring queue for inbound: " + baseQueueName

20
spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -461,6 +461,26 @@ public class RabbitBinderTests extends
assertThat(container.isRunning()).isFalse();
}
@Test
public void testAnonWithBuiltInExchangeCustomPrefix() throws Exception {
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setDeclareExchange(false);
properties.getExtension().setQueueNameGroupOnly(true);
properties.getExtension().setAnonymousGroupPrefix("customPrefix.");
Binding<MessageChannel> consumerBinding = binder.bindConsumer("amq.topic", null,
createBindableChannel("input", new BindingProperties()), properties);
Lifecycle endpoint = extractEndpoint(consumerBinding);
SimpleMessageListenerContainer container = TestUtils.getPropertyValue(endpoint,
"messageListenerContainer", SimpleMessageListenerContainer.class);
String queueName = container.getQueueNames()[0];
assertThat(queueName).startsWith("customPrefix.");
assertThat(container.isRunning()).isTrue();
consumerBinding.unbind();
assertThat(container.isRunning()).isFalse();
}
@SuppressWarnings("deprecation")
@Test
public void testConsumerPropertiesWithUserInfrastructureCustomExchangeAndRK()