GH-133: Add connection name prefix property

Fixes https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/133

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;
+	}
+
 }

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
 

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;
+	}
+
 }