AMQP-605: Expose RabbitMQ ClientProperties

JIRA: https://jira.spring.io/browse/AMQP-605

`AMQPAppender`s, `RabbitConnectionFactoryBean`; also add a getter
to the `AbstractConnectionFactory` to provide access to the underlying
factory.

Fix Incorrect "new" `@Since`

Polishing - PR Comments

Additional polishing according the latest PR comments

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/AbstractConnectionFactory.java

@@ -77,6 +77,16 @@ public abstract class AbstractConnectionFactory implements ConnectionFactory, Di
 		this.rabbitConnectionFactory = rabbitConnectionFactory;
 	}
 
+	/**
+	 * Return a reference to the underlying Rabbit Connection factory.
+	 * @return the connection factory.
+	 * @since 1.5.6
+	 */
+	public com.rabbitmq.client.ConnectionFactory getRabbitConnectionFactory() {
+		return this.rabbitConnectionFactory;
+	}
+
+
 	public void setUsername(String username) {
 		this.rabbitConnectionFactory.setUsername(username);
 	}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/connection/RabbitConnectionFactoryBean.java

@@ -320,11 +320,12 @@ public class RabbitConnectionFactoryBean extends AbstractFactoryBean<ConnectionF
 	}
 
 	/**
+	 * Add custom client properties.
 	 * @param clientProperties the client properties.
 	 * @see com.rabbitmq.client.ConnectionFactory#setClientProperties(java.util.Map)
 	 */
 	public void setClientProperties(Map<String, Object> clientProperties) {
-		this.connectionFactory.setClientProperties(clientProperties);
+		this.connectionFactory.getClientProperties().putAll(clientProperties);
 	}
 
 	/**

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j/AmqpAppender.java

@@ -52,6 +52,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.DeclareExchangeConnectionListener;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.support.LogAppenderUtils;
 
 /**
  * A Log4J appender that publishes logging events to an AMQP Exchange.
@@ -189,6 +190,12 @@ public class AmqpAppender extends AppenderSkeleton {
 	 */
 	private AbstractConnectionFactory connectionFactory;
 
+	/**
+	 * Additional client connection properties added to the rabbit connection, with the form
+	 * {@code key:value[,key:value]...}.
+	 */
+	private String clientConnectionProperties;
+
 	/**
 	 * A comma-delimited list of broker addresses: host:port[,host:port]*.
 	 * @since 1.5.6
@@ -413,6 +420,16 @@ public class AmqpAppender extends AppenderSkeleton {
 		this.charset = charset;
 	}
 
+	/**
+	 * Set additional client connection properties to be added to the rabbit connection,
+	 * with the form {@code key:value[,key:value]...}.
+	 * @param clientConnectionProperties the properties.
+	 * @since 1.5.6
+	 */
+	public void setClientConnectionProperties(String clientConnectionProperties) {
+		this.clientConnectionProperties = clientConnectionProperties;
+	}
+
 	@Override
 	public void activateOptions() {
 		this.routingKeyLayout = new PatternLayout(this.routingKeyPattern
@@ -426,10 +443,21 @@ public class AmqpAppender extends AppenderSkeleton {
 		this.connectionFactory.setUsername(this.username);
 		this.connectionFactory.setPassword(this.password);
 		this.connectionFactory.setVirtualHost(this.virtualHost);
+		LogAppenderUtils.updateClientConnectionProperties(this.connectionFactory, this.clientConnectionProperties);
+		updateConnectionClientProperties(this.connectionFactory.getRabbitConnectionFactory().getClientProperties());
 		setUpExchangeDeclaration();
 		startSenders();
 	}
 
+	/**
+	 * Subclasses can override this method to add properties to the connection client
+	 * properties.
+	 * @param clientProperties the client properties.
+	 * @since 1.5.6
+	 */
+	protected void updateConnectionClientProperties(Map<String, Object> clientProperties) {
+	}
+
 	/**
 	 * @deprecated - use {@link #setUpExchangeDeclaration()}
 	 */

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/log4j2/AmqpAppender.java

@@ -58,6 +58,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.DeclareExchangeConnectionListener;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.support.LogAppenderUtils;
 
 /**
  * A Log4j 2 appender that publishes logging events to an AMQP Exchange.
@@ -138,6 +139,7 @@ public class AmqpAppender extends AbstractAppender {
 			@PluginAttribute("autoDelete") boolean autoDelete,
 			@PluginAttribute("contentType") String contentType,
 			@PluginAttribute("contentEncoding") String contentEncoding,
+			@PluginAttribute("clientConnectionProperties") String clientConnectionProperties,
 			@PluginAttribute("charset") String charset) {
 		if (name == null) {
 			LogFactory.getLog("log4j2AppenderErrors").error("No name for AmqpAppender");
@@ -166,6 +168,7 @@ public class AmqpAppender extends AbstractAppender {
 		manager.autoDelete = autoDelete;
 		manager.contentType = contentType;
 		manager.contentEncoding = contentEncoding;
+		manager.clientConnectionProperties = clientConnectionProperties;
 		manager.charset = charset;
 		AmqpAppender appender = new AmqpAppender(name, filter, theLayout, ignoreExceptions, manager);
 		manager.activateOptions();
@@ -193,7 +196,6 @@ public class AmqpAppender extends AbstractAppender {
 	 * @param message The message.
 	 * @param event The event.
 	 * @return The modified message.
-	 * @since 1.4
 	 */
 	public Message postProcessMessageBeforeSend(Message message, Event event) {
 		return message;
@@ -385,7 +387,6 @@ public class AmqpAppender extends AbstractAppender {
 
 		/**
 		 * A comma-delimited list of broker addresses: host:port[,host:port]*.
-		 * @since 1.6
 		 */
 		private String addresses;
 
@@ -424,6 +425,12 @@ public class AmqpAppender extends AbstractAppender {
 		 */
 		private boolean declareExchange = false;
 
+		/**
+		 * Additional client connection properties to be added to the rabbit connection,
+		 * with the form {@code key:value[,key:value]...}.
+		 */
+		private String clientConnectionProperties;
+
 		/**
 		 * charset to use when converting String to byte[], default null (system default charset used).
 		 * If the charset is unsupported on the current platform, we fall back to using
@@ -452,7 +459,6 @@ public class AmqpAppender extends AbstractAppender {
 		 */
 		private final Timer retryTimer = new Timer("log-event-retry-delay", true);
 
-
 		protected AmqpManager(String name) {
 			super(name);
 		}
@@ -470,6 +476,10 @@ public class AmqpAppender extends AbstractAppender {
 			this.connectionFactory.setUsername(this.username);
 			this.connectionFactory.setPassword(this.password);
 			this.connectionFactory.setVirtualHost(this.virtualHost);
+			if (this.clientConnectionProperties != null) {
+				LogAppenderUtils.updateClientConnectionProperties(this.connectionFactory,
+						this.clientConnectionProperties);
+			}
 			setUpExchangeDeclaration();
 			this.senderPool = Executors.newCachedThreadPool();
 		}

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/logback/AmqpAppender.java

@@ -43,6 +43,7 @@ import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.core.DeclareExchangeConnectionListener;
 import org.springframework.amqp.rabbit.core.RabbitAdmin;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.amqp.rabbit.support.LogAppenderUtils;
 
 import ch.qos.logback.classic.Level;
 import ch.qos.logback.classic.PatternLayout;
@@ -154,6 +155,12 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
 	 */
 	private AbstractConnectionFactory connectionFactory;
 
+	/**
+	 * Additional client connection properties added to the rabbit connection, with the form
+	 * {@code key:value[,key:value]...}.
+	 */
+	private String clientConnectionProperties;
+
 	/**
 	 * A comma-delimited list of broker addresses: host:port[,host:port]*
 	 * @since 1.5.6
@@ -390,6 +397,16 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
 		this.abbreviator = new TargetLengthBasedClassNameAbbreviator(len);
 	}
 
+	/**
+	 * Set additional client connection properties to be added to the rabbit connection,
+	 * with the form {@code key:value[,key:value]...}.
+	 * @param clientConnectionProperties the properties.
+	 * @since 1.5.6
+	 */
+	public void setClientConnectionProperties(String clientConnectionProperties) {
+		this.clientConnectionProperties = clientConnectionProperties;
+	}
+
 	@Override
 	public void start() {
 		super.start();
@@ -408,6 +425,8 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
 		this.connectionFactory.setUsername(this.username);
 		this.connectionFactory.setPassword(this.password);
 		this.connectionFactory.setVirtualHost(this.virtualHost);
+		LogAppenderUtils.updateClientConnectionProperties(this.connectionFactory, this.clientConnectionProperties);
+		updateConnectionClientProperties(this.connectionFactory.getRabbitConnectionFactory().getClientProperties());
 		setUpExchangeDeclaration();
 		this.senderPool = Executors.newCachedThreadPool();
 		for (int i = 0; i < this.senderPoolSize; i++) {
@@ -415,6 +434,15 @@ public class AmqpAppender extends AppenderBase<ILoggingEvent> {
 		}
 	}
 
+	/**
+	 * Subclasses can override this method to add properties to the connection client
+	 * properties.
+	 * @param clientProperties the client properties.
+	 * @since 1.5.6
+	 */
+	protected void updateConnectionClientProperties(Map<String, Object> clientProperties) {
+	}
+
 	@Override
 	public void stop() {
 		super.stop();

spring-rabbit/src/main/java/org/springframework/amqp/rabbit/support/LogAppenderUtils.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2016 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.amqp.rabbit.support;
+
+import java.util.Map;
+
+import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
+
+/**
+ * Utility methods for log appenders.
+ *
+ * @author Gary Russell
+ * @since 1.5.6
+ *
+ */
+public final class LogAppenderUtils {
+
+	private LogAppenderUtils() {
+		// empty
+	}
+
+	/**
+	 * Parse the properties {@code key:value[,key:value]...} and add them to the
+	 * connection factory client properties.
+	 * @param connectionFactory the connection factory.
+	 * @param clientConnectionProperties the properties.
+	 */
+	public static void updateClientConnectionProperties(AbstractConnectionFactory connectionFactory,
+			String clientConnectionProperties) {
+		if (clientConnectionProperties != null) {
+			String[] props = clientConnectionProperties.split(",");
+			if (props.length > 0) {
+				Map<String, Object> clientProps = connectionFactory.getRabbitConnectionFactory()
+						.getClientProperties();
+				for (String prop : props) {
+					String[] aProp = prop.split(":");
+					if (aProp.length == 2) {
+						clientProps.put(aProp[0].trim(), aProp[1].trim());
+					}
+				}
+			}
+		}
+	}
+
+}

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/CachingConnectionFactoryIntegrationTests.java

@@ -93,6 +93,7 @@ public class CachingConnectionFactoryIntegrationTests {
 		connectionFactory = new CachingConnectionFactory();
 		connectionFactory.setHost("localhost");
 		connectionFactory.setPort(BrokerTestUtils.getPort());
+		connectionFactory.getRabbitConnectionFactory().getClientProperties().put("foo", "bar");
 	}
 
 	@After
@@ -100,6 +101,7 @@ public class CachingConnectionFactoryIntegrationTests {
 		if (!this.connectionFactory.getVirtualHost().equals("non-existent")) {
 			new RabbitAdmin(this.connectionFactory).deleteQueue(CF_INTEGRATION_TEST_QUEUE);
 		}
+		assertEquals("bar", connectionFactory.getRabbitConnectionFactory().getClientProperties().get("foo"));
 		connectionFactory.destroy();
 	}
 

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/connection/SSLConnectionTests.java

@@ -16,9 +16,12 @@
 
 package org.springframework.amqp.rabbit.connection;
 
+import static org.junit.Assert.assertEquals;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
 
+import java.util.Collections;
+
 import org.junit.Ignore;
 import org.junit.Test;
 
@@ -43,8 +46,10 @@ public class SSLConnectionTests {
 		RabbitConnectionFactoryBean fb = new RabbitConnectionFactoryBean();
 		fb.setUseSSL(true);
 		fb.setSslPropertiesLocation(new ClassPathResource("ssl.properties"));
+		fb.setClientProperties(Collections.<String, Object>singletonMap("foo", "bar"));
 		fb.afterPropertiesSet();
 		ConnectionFactory cf = fb.getObject();
+		assertEquals("bar", cf.getClientProperties().get("foo"));
 		Connection conn = cf.newConnection();
 		Channel chan = conn.createChannel();
 		chan.close();

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/log4j/AmqpAppenderIntegrationTests.java

@@ -25,6 +25,7 @@ import static org.junit.Assert.assertTrue;
 
 import java.io.ByteArrayInputStream;
 import java.util.Collection;
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import javax.xml.parsers.DocumentBuilderFactory;
@@ -207,6 +208,13 @@ public class AmqpAppenderIntegrationTests {
 			this.foo = foo;
 		}
 
+		@Override
+		protected void updateConnectionClientProperties(Map<String, Object> clientProperties) {
+			assertEquals("bar", clientProperties.get("foo"));
+			assertEquals("qux", clientProperties.get("baz"));
+			clientProperties.put("foo", this.foo.toUpperCase());
+		}
+
 	}
 
 }

spring-rabbit/src/test/java/org/springframework/amqp/rabbit/logback/AmqpAppenderIntegrationTests.java

@@ -23,6 +23,7 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.assertTrue;
 
+import java.util.Map;
 import java.util.concurrent.TimeUnit;
 
 import org.junit.After;
@@ -33,6 +34,7 @@ import org.junit.runner.RunWith;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
 
+import org.springframework.amqp.core.Message;
 import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
 import org.springframework.amqp.rabbit.log4j.AmqpAppenderConfiguration;
@@ -116,6 +118,7 @@ public class AmqpAppenderIntegrationTests {
 		assertThat(location, instanceOf(String.class));
 		assertThat((String) location,
 				startsWith("org.springframework.amqp.rabbit.logback.AmqpAppenderIntegrationTests.testAppenderWithProps()"));
+		assertEquals("bar", messageProperties.getHeaders().get("foo"));
 	}
 
 	@Test
@@ -134,4 +137,31 @@ public class AmqpAppenderIntegrationTests {
 		assertEquals(0xbf, body[body.length - 3 - lineSeparatorExtraBytes] & 0xff);
 	}
 
+	public static class EnhancedAppender extends AmqpAppender {
+
+		private String foo;
+
+		@Override
+		public Message postProcessMessageBeforeSend(Message message, Event event) {
+			message.getMessageProperties().setHeader("foo", this.foo);
+			return message;
+		}
+
+		public String getFoo() {
+			return this.foo;
+		}
+
+		public void setFoo(String foo) {
+			this.foo = foo;
+		}
+
+		@Override
+		protected void updateConnectionClientProperties(Map<String, Object> clientProperties) {
+			assertEquals("bar", clientProperties.get("foo"));
+			assertEquals("qux", clientProperties.get("baz"));
+			clientProperties.put("foo", this.foo.toUpperCase());
+		}
+
+	}
+
 }

spring-rabbit/src/test/resources/log4j-amqp.properties

@@ -15,6 +15,7 @@ log4j.appender.amqp.charset=UTF-8
 log4j.appender.amqp.durable=false
 log4j.appender.amqp.deliveryMode=NON_PERSISTENT
 log4j.appender.amqp.declareExchange=true
+log4j.appender.amqp.clientConnectionProperties=foo:bar,baz:qux
 
 
 log4j.appender.amqp.foo=bar

spring-rabbit/src/test/resources/log4j2.xml

@@ -11,6 +11,7 @@
 			applicationId="testAppId" routingKeyPattern="%X{applicationId}.%c.%p"
 			contentType="text/plain" contentEncoding="UTF-8" generateId="true" deliveryMode="NON_PERSISTENT"
 			charset="UTF-8"
+			clientConnectionProperties="foo:bar,baz:qux"
 			senderPoolSize="3" maxSenderRetries="5">
 		</RabbitMQ>
 	</Appenders>

spring-rabbit/src/test/resources/logback-test.xml

@@ -7,7 +7,7 @@
 		</encoder>
 	</appender>
 
-	<appender name="AMQP" class="org.springframework.amqp.rabbit.logback.AmqpAppender">
+	<appender name="AMQP" class="org.springframework.amqp.rabbit.logback.AmqpAppenderIntegrationTests$EnhancedAppender">
 		<layout>
 			<pattern><![CDATA[ %d %p %t [%c] - <%m>%n ]]></pattern>
 		</layout>
@@ -20,6 +20,8 @@
 		<durable>false</durable>
 		<deliveryMode>NON_PERSISTENT</deliveryMode>
 		<declareExchange>true</declareExchange>
+		<clientConnectionProperties>foo:bar,baz:qux</clientConnectionProperties>
+		<foo>bar</foo>
 	</appender>
 
 	<logger name="org.springframework.amqp.rabbit.logback" level="DEBUG" additivity="false">

src/reference/asciidoc/amqp.adoc

@@ -342,7 +342,7 @@ For convenience, a factory bean is provided to assist in configuring the connect
       id="connectionFactory" connection-factory="rabbitConnectionFactory"/>
 ----
 
-===== Configuring SSL
+===== RabbitConnectionFactoryBean and Configuring SSL
 
 Starting with _version 1.4_, a convenient `RabbitConnectionFactoryBean` is provided to enable convenient configuration of SSL properties on the underlying client connection factory, using dependency injection.
 Other setters simply delegate to the underlying factory.
@@ -635,6 +635,19 @@ The `cacheMode` property (`CHANNEL` or `CONNECTION` is also included).
 .JVisualVM Example
 image::images/cacheStats.png[align="center"]
 
+[[custom-client-props]]
+==== Adding Custom Client Connection Properties
+
+The `CachingConnectionFactory` now allows you to access the underlying connection factory to allow, for example,
+setting custom client properties:
+
+[source, java]
+----
+connectionFactory.getRabbitConnectionFactory().getClientProperties().put("foo", "bar");
+----
+
+These properties appear in the RabbitMQ Admin UI when viewing the connection.
+
 [[amqp-template]]
 ==== AmqpTemplate
 

src/reference/asciidoc/logging.adoc

@@ -105,6 +105,10 @@ If the charset is unsupported on the current platform, we fall back to using the
 | false
 | Used to determine whether the `messageId` property is set to a unique value.
 
+| clientConnectionProperties
+| null
+| A comma-delimited list of `key:value` pairs for custom client properties to the RabbitMQ connection.
+
 |===
 
 ==== Log4j Appender
@@ -181,3 +185,74 @@ public class MyEnhancedAppender extends AmqpAppender {
 
 }
 ----
+
+==== Customizing the Client Properties
+
+===== Simple String Properties
+
+Each appender supports adding client properties to the RabbitMQ connection.
+
+.log4j
+[source, text]
+----
+log4j.appender.amqp.clientConnectionProperties=foo:bar,baz:qux
+----
+
+.logback
+[source, xml]
+----
+<appender name="AMQP" ...>
+    ...
+    <clientConnectionProperties>foo:bar,baz:qux</clientConnectionProperties>
+    ...
+</appender>
+----
+
+.log4j2
+[source, xml]
+----
+<Appenders>
+    ...
+    <RabbitMQ name="rabbitmq"
+        ...
+        clientConnectionProperties="foo:bar,baz:qux"
+        ...
+    </RabbitMQ>
+</Appenders>
+----
+
+The properties are a comma-delimited list of `key:value` pairs; keys and values cannot contain commas or colons.
+
+These properties appear on the RabbitMQ Admin UI when viewing the connection.
+
+===== Advanced Technique for Log4j and Logback
+
+With the log4j and logback appenders, the appenders can be subclassed, allowing you to modify the client connection
+properties before the connection is established:
+
+.Customizing the Client Connection Properties
+[source, java]
+----
+public class MyEnhancedAppender extends AmqpAppender {
+
+    private String foo;
+
+    @Override
+    protected void updateConnectionClientProperties(Map<String, Object> clientProperties) {
+        clientProperties.put("foo", this.foo);
+    }
+
+    public void setFoo(String foo) {
+        this.foo = foo;
+    }
+
+}
+----
+
+For log4j2, add `log4j.appender.amqp.foo=bar` to log4j.properties to set the property.
+For logback, add `<foo>bar</foo>` to logback.xml.
+
+Of course, for simple String properties like this example, the previous technique can be used; subclasses allow
+richer properties (such as adding a `Map` or numeric property).
+
+With log4j2, subclasses are not supported, due to the way log4j2 uses static factory methods.

src/reference/asciidoc/whats-new.adoc

@@ -147,11 +147,24 @@ See <<async-annotation-driven>> for more information.
 Spring AMQP now has first class support for the RabbitMQ Delayed Message Exchange plugin.
 See <<delayed-message-exchange>> for more information.
 
-===== CachingConnectionFactory Cache Statistics
+===== CachingConnectionFactory Changes
+
+====== CachingConnectionFactory Cache Statistics
 
 The `CachingConnectionFactory` now provides cache properties at runtime and over JMX.
 See <<runtime-cache-properties>> for more information.
 
+====== Access the Underlying RabbitMQ Connection Factory
+
+A new getter has been added to provide access to the underlying factory.
+This can be used, for example, to add custom connection properties.
+See <<custom-client-props>> for more information.
+
+===== RabbitConnectionFactoryBean
+
+The factory bean now exposes a property to add client connection properties to connections made by the resulting
+factory.
+
 ===== Java Deserialization
 
 A "white list" of allowable classes can now be configured when using Java deserialization.
@@ -167,8 +180,15 @@ See <<async-annotation-conversion>> and <<json-message-converter>> for more info
 
 ===== Logging Appenders
 
+====== Log4j2
+
 A log4j2 appender has been added, and the appenders can now be configured with an `addresses` property to connect
 to a broker cluster.
+
+====== Client Connection Properties
+
+You can now add custom client connection properties to RabbitMQ connections.
+
 See <<logging>> for more information.
 
 ==== Earlier Releases