From 26f27909b88542487078b65aec2acabfc28b1a43 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Tue, 6 Mar 2018 18:00:22 -0500 Subject: [PATCH] GH-133: Add connection name prefix property Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/133 --- .../RabbitBinderConfigurationProperties.java | 14 +++++++++++++- .../src/main/asciidoc/overview.adoc | 5 +++++ ...abbitMessageChannelBinderConfiguration.java | 18 ++++++++++++++++++ 3 files changed, 36 insertions(+), 1 deletion(-) diff --git a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBinderConfigurationProperties.java b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBinderConfigurationProperties.java index bb3797638..b4beedabe 100644 --- a/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBinderConfigurationProperties.java +++ b/spring-cloud-stream-binder-rabbit-core/src/main/java/org/springframework/cloud/stream/binder/rabbit/properties/RabbitBinderConfigurationProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2016 the original author or authors. + * Copyright 2015-2018 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. @@ -20,6 +20,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties; /** * @author David Turanski + * @author Gary Russell */ @ConfigurationProperties(prefix = "spring.cloud.stream.rabbit.binder") public class RabbitBinderConfigurationProperties { @@ -30,6 +31,8 @@ public class RabbitBinderConfigurationProperties { private int compressionLevel; + private String connectionNamePrefix; + public String[] getAdminAddresses() { return adminAddresses; } @@ -68,4 +71,13 @@ public class RabbitBinderConfigurationProperties { public void setCompressionLevel(int compressionLevel) { this.compressionLevel = compressionLevel; } + + public String getConnectionNamePrefix() { + return this.connectionNamePrefix; + } + + public void setConnectionNamePrefix(String connectionNamePrefix) { + this.connectionNamePrefix = connectionNamePrefix; + } + } 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 1b046f9e1..121c42faf 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 @@ -98,6 +98,11 @@ spring.cloud.stream.rabbit.binder.compressionLevel:: See `java.util.zip.Deflater`. + Default: `1` (BEST_LEVEL). +spring.cloud.stream.binder.connection-name-prefix:: + A connection name prefix used to name the connection(s) created by this binder. +The name will be this prefix followed by `#n`, where n increments each time a new connection is opened. ++ +Defauklt: none (Spring AMQP default). === RabbitMQ Consumer Properties 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 e3ead9dbd..e7cc6bafa 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 @@ -16,12 +16,18 @@ package org.springframework.cloud.stream.binder.rabbit.config; +import java.util.concurrent.atomic.AtomicInteger; + import org.springframework.amqp.core.MessagePostProcessor; +import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; +import org.springframework.amqp.rabbit.connection.ConnectionNameStrategy; import org.springframework.amqp.support.postprocessor.DelegatingDecompressingPostProcessor; import org.springframework.amqp.support.postprocessor.GZipPostProcessor; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; +import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.context.PropertyPlaceholderAutoConfiguration; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.cloud.stream.binder.rabbit.RabbitMessageChannelBinder; @@ -91,4 +97,16 @@ public class RabbitMessageChannelBinderConfiguration { return new RabbitExchangeQueueProvisioner(this.rabbitConnectionFactory); } + @Bean + @ConditionalOnMissingBean(ConnectionNameStrategy.class) + @ConditionalOnProperty("spring.cloud.stream.rabbit.binder.connection-name-prefix") + public ConnectionNameStrategy connectionNamer(CachingConnectionFactory cf) { + final AtomicInteger nameIncrementer = new AtomicInteger(); + ConnectionNameStrategy namer = f -> this.rabbitBinderConfigurationProperties.getConnectionNamePrefix() + + "#" + nameIncrementer.getAndIncrement(); + // TODO: this can be removed when Boot 2.0.1 wires it in + cf.setConnectionNameStrategy(namer); + return namer; + } + }