AMQP-428: Add Logback AmqpAppender

JIRA: https://jira.spring.io/browse/AMQP-433, https://jira.spring.io/browse/AMQP-428

Fix `amqp.rabbit.log4j.AmqpAppender` JavaDoc typo

Polishing

Add category abbreviation [o.s.a.r.l.AmqpAppenderIntegrationTests].

Retain full category in AMQP headers.

build.gradle

@@ -78,12 +78,13 @@ subprojects { subproject ->
 		jackson2Version = '2.3.2'
 		junitVersion = '4.11'
 		log4jVersion = '1.2.17'
+		logbackVersion = '1.1.2'
 		mockitoVersion = '1.9.5'
 		rabbitmqVersion = project.hasProperty('rabbitmqVersion') ? project.rabbitmqVersion : '3.3.4'
 
 		springVersion = project.hasProperty('springVersion') ? project.springVersion : '4.1.0.RELEASE'
 
-		springRetryVersion = '1.1.1.RELEASE'
+		springRetryVersion = '1.1.2.RELEASE'
 	}
 
 	eclipse {
@@ -224,6 +225,7 @@ project('spring-rabbit') {
 			exclude group: 'com.sun.jmx', module: 'jmxri'
 		}
 
+		compile ("ch.qos.logback:logback-classic:$logbackVersion", optional)
 	}
 
 	// suppress deprecation warnings (@SuppressWarnings("deprecation") is not enough for javac)

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

@@ -56,7 +56,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
  * <p>
  * A fully-configured AmqpAppender, with every option set to their defaults, would look like this:
  * <pre class="code">
- *   log4j.appender.amqp=org.springframework.amqp.log4j.AmqpAppender
+ *   log4j.appender.amqp=org.springframework.amqp.rabbit.log4j.AmqpAppender
  *   #-------------------------------
  *   ## Connection settings
  *   #-------------------------------
@@ -251,9 +251,6 @@ public class AmqpAppender extends AppenderSkeleton {
 
 	private final AtomicBoolean initializing = new AtomicBoolean();
 
-	public AmqpAppender() {
-	}
-
 	public String getHost() {
 		return host;
 	}

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

@@ -0,0 +1,586 @@
+/*
+ * Copyright 2014 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.logback;
+
+import java.io.UnsupportedEncodingException;
+import java.util.Calendar;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.Exchange;
+import org.springframework.amqp.core.FanoutExchange;
+import org.springframework.amqp.core.HeadersExchange;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageDeliveryMode;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.PatternLayout;
+import ch.qos.logback.classic.pattern.TargetLengthBasedClassNameAbbreviator;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.classic.spi.LoggingEvent;
+import ch.qos.logback.core.AppenderBase;
+import ch.qos.logback.core.Layout;
+
+/**
+ * A Lockback appender that publishes logging events to an AMQP Exchange.
+ * <p>
+ * A fully-configured AmqpAppender, with every option set to their defaults, would look like this:
+ * <pre class="code">
+ * {@code
+ * <appender name="AMQP" class="org.springframework.amqp.rabbit.logback.AmqpAppender">
+ *     <layout>
+ *        <pattern><![CDATA[ %d %p %t [%c] - <%m>%n ]]></pattern>
+ *     </layout>
+ *     <!-- <abbreviation>36</abbreviation --> <!-- no category abbreviation by default -->
+ *     <applicationId>AmqpAppenderTest</applicationId>
+ *     <routingKeyPattern>%property{applicationId}.%c.%p</routingKeyPattern>
+ *     <generateId>true</generateId>
+ *     <charset>UTF-8</charset>
+ *     <durable>false</durable>
+ *     <deliveryMode>NON_PERSISTENT</deliveryMode>
+ * </appender>
+ * }
+ * </pre>
+ *
+ * @author Artem Bilan
+ * @since 1.4
+ */
+public class AmqpAppender extends AppenderBase<ILoggingEvent> {
+
+	/**
+	 * Key name for the application id (if there is one set via the appender config) in the message properties.
+	 */
+	public static final String APPLICATION_ID = "applicationId";
+
+	/**
+	 * Key name for the logger category name in the message properties
+	 */
+	public static final String CATEGORY_NAME = "categoryName";
+
+	/**
+	 * Key name for the logger level name in the message properties
+	 */
+	public static final String CATEGORY_LEVEL = "level";
+
+	/**
+	 * Name of the exchange to publish log events to.
+	 */
+	private String exchangeName = "logs";
+
+	/**
+	 * Type of the exchange to publish log events to.
+	 */
+	private String exchangeType = "topic";
+
+	private final PatternLayout locationLayout = new PatternLayout();
+
+	{
+		this.locationLayout.setPattern("%nopex%class|%method|%line");
+	}
+
+	/**
+	 * Logback Layout to use to generate routing key.
+	 */
+	private final PatternLayout routingKeyLayout = new PatternLayout();
+
+	{
+		this.routingKeyLayout.setPattern("%nopex%c.%p");
+	}
+
+	/**
+	 * Configuration arbitrary application ID.
+	 */
+	private String applicationId = null;
+
+	/**
+	 * Where LoggingEvents are queued to send.
+	 */
+	private final LinkedBlockingQueue<Event> events = new LinkedBlockingQueue<Event>();
+
+	/**
+	 * The pool of senders.
+	 */
+	private ExecutorService senderPool = null;
+
+	/**
+	 * How many senders to use at once. Use more senders if you have lots of log output going through this appender.
+	 */
+	private int senderPoolSize = 2;
+
+	/**
+	 * How many times to retry sending a message if the broker is unavailable or there is some other error.
+	 */
+	private int maxSenderRetries = 30;
+
+	/**
+	 * Retries are delayed like: N ^ log(N), where N is the retry number.
+	 */
+	private final Timer retryTimer = new Timer("log-event-retry-delay", true);
+
+	/**
+	 * RabbitMQ ConnectionFactory.
+	 */
+	private AbstractConnectionFactory connectionFactory;
+
+	/**
+	 * RabbitMQ host to connect to.
+	 */
+	private String host = "localhost";
+
+	/**
+	 * RabbitMQ virtual host to connect to.
+	 */
+	private String virtualHost = "/";
+
+	/**
+	 * RabbitMQ port to connect to.
+	 */
+	private int port = 5672;
+
+	/**
+	 * RabbitMQ user to connect as.
+	 */
+	private String username = "guest";
+
+	/**
+	 * RabbitMQ password for this user.
+	 */
+	private String password = "guest";
+
+	/**
+	 * Default content-type of log messages.
+	 */
+	private String contentType = "text/plain";
+
+	/**
+	 * Default content-encoding of log messages.
+	 */
+	private String contentEncoding = null;
+
+	/**
+	 * Whether or not to try and declare the configured exchange when this appender starts.
+	 */
+	private boolean declareExchange = false;
+
+	/**
+	 * 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
+	 * the system charset.
+	 */
+	private String charset;
+
+	private boolean durable = true;
+
+	private MessageDeliveryMode deliveryMode = MessageDeliveryMode.PERSISTENT;
+
+	private boolean autoDelete = false;
+
+	/**
+	 * Used to determine whether {@link MessageProperties#setMessageId(String)} is set.
+	 */
+	private boolean generateId = false;
+
+	private Layout<ILoggingEvent> layout;
+
+	private TargetLengthBasedClassNameAbbreviator abbreviator;
+
+	public void setRoutingKeyPattern(String routingKeyPattern) {
+		this.routingKeyLayout.setPattern("%nopex" + routingKeyPattern);
+	}
+
+	public String getHost() {
+		return host;
+	}
+
+	public void setHost(String host) {
+		this.host = host;
+	}
+
+	public int getPort() {
+		return port;
+	}
+
+	public void setPort(int port) {
+		this.port = port;
+	}
+
+	public String getVirtualHost() {
+		return virtualHost;
+	}
+
+	public void setVirtualHost(String virtualHost) {
+		this.virtualHost = virtualHost;
+	}
+
+	public String getUsername() {
+		return username;
+	}
+
+	public void setUsername(String username) {
+		this.username = username;
+	}
+
+	public String getPassword() {
+		return password;
+	}
+
+	public void setPassword(String password) {
+		this.password = password;
+	}
+
+	public String getExchangeName() {
+		return exchangeName;
+	}
+
+	public void setExchangeName(String exchangeName) {
+		this.exchangeName = exchangeName;
+	}
+
+	public String getExchangeType() {
+		return exchangeType;
+	}
+
+	public void setExchangeType(String exchangeType) {
+		this.exchangeType = exchangeType;
+	}
+
+	public String getRoutingKeyPattern() {
+		return this.routingKeyLayout.getPattern();
+	}
+
+	public boolean isDeclareExchange() {
+		return declareExchange;
+	}
+
+	public void setDeclareExchange(boolean declareExchange) {
+		this.declareExchange = declareExchange;
+	}
+
+	public String getContentType() {
+		return contentType;
+	}
+
+	public void setContentType(String contentType) {
+		this.contentType = contentType;
+	}
+
+	public String getContentEncoding() {
+		return contentEncoding;
+	}
+
+	public void setContentEncoding(String contentEncoding) {
+		this.contentEncoding = contentEncoding;
+	}
+
+	public String getApplicationId() {
+		return applicationId;
+	}
+
+	public void setApplicationId(String applicationId) {
+		this.applicationId = applicationId;
+	}
+
+	public int getSenderPoolSize() {
+		return senderPoolSize;
+	}
+
+	public void setSenderPoolSize(int senderPoolSize) {
+		this.senderPoolSize = senderPoolSize;
+	}
+
+	public int getMaxSenderRetries() {
+		return maxSenderRetries;
+	}
+
+	public void setMaxSenderRetries(int maxSenderRetries) {
+		this.maxSenderRetries = maxSenderRetries;
+	}
+
+	public boolean isDurable() {
+		return durable;
+	}
+
+	public void setDurable(boolean durable) {
+		this.durable = durable;
+	}
+
+	public String getDeliveryMode() {
+		return this.deliveryMode.toString();
+	}
+
+	public void setDeliveryMode(String deliveryMode) {
+		this.deliveryMode = MessageDeliveryMode.valueOf(deliveryMode);
+	}
+
+	public boolean isAutoDelete() {
+		return autoDelete;
+	}
+
+	public void setAutoDelete(boolean autoDelete) {
+		this.autoDelete = autoDelete;
+	}
+
+	public boolean isGenerateId() {
+		return generateId;
+	}
+
+	public void setGenerateId(boolean generateId) {
+		this.generateId = generateId;
+	}
+
+	public String getCharset() {
+		return charset;
+	}
+
+	public void setCharset(String charset) {
+		this.charset = charset;
+	}
+
+	public void setLayout(Layout<ILoggingEvent> layout) {
+		this.layout = layout;
+	}
+
+	public void setAbbreviation(int len) {
+		this.abbreviator = new TargetLengthBasedClassNameAbbreviator(len);
+	}
+
+	@Override
+	public void start() {
+		super.start();
+		this.routingKeyLayout.setContext(getContext());
+		this.routingKeyLayout.start();
+		this.locationLayout.setContext(getContext());
+		this.locationLayout.start();
+		this.connectionFactory = new CachingConnectionFactory();
+		this.connectionFactory.setHost(this.host);
+		this.connectionFactory.setPort(this.port);
+		this.connectionFactory.setUsername(this.username);
+		this.connectionFactory.setPassword(this.password);
+		this.connectionFactory.setVirtualHost(this.virtualHost);
+		maybeDeclareExchange();
+		this.senderPool = Executors.newCachedThreadPool();
+		for (int i = 0; i < this.senderPoolSize; i++) {
+			this.senderPool.submit(new EventSender());
+		}
+	}
+
+	@Override
+	public void stop() {
+		super.stop();
+		if (null != this.senderPool) {
+			this.senderPool.shutdownNow();
+			this.senderPool = null;
+		}
+		if (null != this.connectionFactory) {
+			this.connectionFactory.destroy();
+		}
+		this.retryTimer.cancel();
+		this.routingKeyLayout.stop();
+	}
+
+	@Override
+	protected void append(ILoggingEvent event) {
+		this.events.add(new Event(event));
+	}
+
+	/**
+	 * Maybe declare the exchange.
+	 */
+	protected void maybeDeclareExchange() {
+		RabbitAdmin admin = new RabbitAdmin(this.connectionFactory);
+		if (this.declareExchange) {
+			Exchange x;
+			if ("topic".equals(this.exchangeType)) {
+				x = new TopicExchange(this.exchangeName, this.durable, this.autoDelete);
+			}
+			else if ("direct".equals(this.exchangeType)) {
+				x = new DirectExchange(this.exchangeName, this.durable, this.autoDelete);
+			}
+			else if ("fanout".equals(this.exchangeType)) {
+				x = new FanoutExchange(this.exchangeName, this.durable, this.autoDelete);
+			}
+			else if ("headers".equals(this.exchangeType)) {
+				x = new HeadersExchange(this.exchangeType, this.durable, this.autoDelete);
+			}
+			else {
+				x = new TopicExchange(this.exchangeName, this.durable, this.autoDelete);
+			}
+			admin.declareExchange(x);
+		}
+	}
+
+	/**
+	 * Subclasses may modify the final message before sending.
+	 * @param message The message.
+	 * @param event The event.
+	 * @return The modified message.
+	 * @since 1.4
+	 */
+	public Message postProcessMessageBeforeSend(Message message, Event event) {
+		return message;
+	}
+
+	/**
+	 * Helper class to actually send LoggingEvents asynchronously.
+	 */
+	protected class EventSender implements Runnable {
+		@Override
+		public void run() {
+			try {
+				RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
+				while (true) {
+					final Event event = events.take();
+					ILoggingEvent logEvent = event.getEvent();
+
+					String name = logEvent.getLoggerName();
+					Level level = logEvent.getLevel();
+
+					MessageProperties amqpProps = new MessageProperties();
+					amqpProps.setDeliveryMode(deliveryMode);
+					amqpProps.setContentType(contentType);
+					if (null != contentEncoding) {
+						amqpProps.setContentEncoding(contentEncoding);
+					}
+					amqpProps.setHeader(CATEGORY_NAME, name);
+					amqpProps.setHeader(CATEGORY_LEVEL, level.toString());
+					if (generateId) {
+						amqpProps.setMessageId(UUID.randomUUID().toString());
+					}
+
+					// Set timestamp
+					Calendar tstamp = Calendar.getInstance();
+					tstamp.setTimeInMillis(logEvent.getTimeStamp());
+					amqpProps.setTimestamp(tstamp.getTime());
+
+					// Copy properties in from MDC
+					Map<String, String> props = event.getProperties();
+					Set<Entry<String, String>> entrySet = props.entrySet();
+					for (Entry<String, String> entry : entrySet) {
+						amqpProps.setHeader(entry.getKey(), entry.getValue());
+					}
+					String[] location = locationLayout.doLayout(logEvent).split("\\|");
+					if (!"?".equals(location[0])) {
+						amqpProps.setHeader(
+								"location",
+								String.format("%s.%s()[%s]", location[0], location[1], location[2]));
+					}
+					String msgBody;
+					String routingKey;
+					// Set applicationId, if we're using one
+					if (null != applicationId) {
+						amqpProps.setAppId(applicationId);
+						logEvent.getLoggerContextVO().getPropertyMap().put(APPLICATION_ID, applicationId);
+					}
+
+					routingKey = routingKeyLayout.doLayout(logEvent);
+
+					if (abbreviator != null && logEvent instanceof LoggingEvent) {
+						((LoggingEvent) logEvent).setLoggerName(abbreviator.abbreviate(name));
+						msgBody = layout.doLayout(logEvent);
+						((LoggingEvent) logEvent).setLoggerName(name);
+					}
+					else {
+						msgBody = layout.doLayout(logEvent);
+					}
+
+					// Send a message
+					try {
+						Message message = null;
+						if (AmqpAppender.this.charset != null) {
+							try {
+								message = new Message(msgBody.getBytes(AmqpAppender.this.charset), amqpProps);
+							}
+							catch (UnsupportedEncodingException e) {
+								message = new Message(msgBody.getBytes(), amqpProps);
+							}
+						}
+
+						message = postProcessMessageBeforeSend(message, event);
+						rabbitTemplate.send(exchangeName, routingKey, message);
+					}
+					catch (AmqpException e) {
+						int retries = event.incrementRetries();
+						if (retries < maxSenderRetries) {
+							// Schedule a retry based on the number of times I've tried to re-send this
+							retryTimer.schedule(new TimerTask() {
+								@Override
+								public void run() {
+									events.add(event);
+								}
+							}, (long) (Math.pow(retries, Math.log(retries)) * 1000));
+						}
+						else {
+							addError("Could not send log message " + logEvent.getMessage()
+									+ " after " + maxSenderRetries + " retries", e);
+						}
+					}
+				}
+			}
+			catch (Throwable t) {
+				throw new RuntimeException(t.getMessage(), t);
+			}
+		}
+	}
+
+	/**
+	 * Small helper class to encapsulate a LoggingEvent, its MDC properties, and the number of retries.
+	 */
+	protected class Event {
+
+		final ILoggingEvent event;
+
+		final Map<String, String> properties;
+
+		final AtomicInteger retries = new AtomicInteger(0);
+
+		public Event(ILoggingEvent event) {
+			this.event = event;
+			this.properties = this.event.getMDCPropertyMap();
+			this.event.getCallerData();
+		}
+
+		public ILoggingEvent getEvent() {
+			return event;
+		}
+
+		public Map<String, String> getProperties() {
+			return properties;
+		}
+
+		public int incrementRetries() {
+			return retries.incrementAndGet();
+		}
+
+	}
+
+}

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

@@ -0,0 +1,4 @@
+/**
+ * Provides classes supporting Logback appenders.
+ */
+package org.springframework.amqp.rabbit.logback;

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

@@ -196,7 +196,7 @@ public class AmqpAppenderIntegrationTests {
 		listenerContainer.setMessageListener(testListener);
 		listenerContainer.start();
 
-		String foo = new String("\u0fff"); // UTF-8 -> 0xe0bfbf
+		String foo = "\u0fff"; // UTF-8 -> 0xe0bfbf
 		log.info(foo);
 		assertTrue(testListener.getLatch().await(5, TimeUnit.SECONDS));
 		byte[] body = testListener.getMessage().getBody();

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

@@ -0,0 +1,137 @@
+/*
+ * Copyright 2014 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.logback;
+
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.startsWith;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThat;
+import static org.junit.Assert.assertTrue;
+
+import java.util.concurrent.TimeUnit;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.log4j.AmqpAppenderConfiguration;
+import org.springframework.amqp.rabbit.log4j.TestListener;
+import org.springframework.amqp.rabbit.test.BrokerRunning;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import ch.qos.logback.classic.Logger;
+
+/**
+ * @author Artem Bilan
+ * @since 1.4
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration(classes = AmqpAppenderConfiguration.class)
+@DirtiesContext
+public class AmqpAppenderIntegrationTests {
+
+	/* logback will automatically find lockback-test.xml */
+	private final static Logger log = (Logger) LoggerFactory.getLogger(AmqpAppenderIntegrationTests.class);
+
+	@Rule
+	public BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
+
+	@Autowired
+	private ApplicationContext applicationContext;
+
+	private SimpleMessageListenerContainer listenerContainer;
+
+	@Before
+	public void setUp() throws Exception {
+		listenerContainer = applicationContext.getBean(SimpleMessageListenerContainer.class);
+	}
+
+	@After
+	public void tearDown() {
+		listenerContainer.shutdown();
+	}
+
+	@Test
+	public void testAppender() throws InterruptedException {
+		TestListener testListener = (TestListener) applicationContext.getBean("testListener", 4);
+		listenerContainer.setMessageListener(testListener);
+		listenerContainer.start();
+
+		log.debug("This is a DEBUG message");
+		log.info("This is an INFO message");
+		log.warn("This is a WARN message");
+		log.error("This is an ERROR message", new RuntimeException("Test exception"));
+
+		assertTrue(testListener.getLatch().await(5, TimeUnit.SECONDS));
+		assertNotNull(testListener.getId());
+	}
+
+	@Test
+	public void testAppenderWithProps() throws InterruptedException {
+		TestListener testListener = (TestListener) applicationContext.getBean("testListener", 4);
+		listenerContainer.setMessageListener(testListener);
+		listenerContainer.start();
+
+		String propertyName = "someproperty";
+		String propertyValue = "property.value";
+		MDC.put(propertyName, propertyValue);
+		log.debug("This is a DEBUG message with properties");
+		log.info("This is an INFO message with properties");
+		log.warn("This is a WARN message with properties");
+		log.error("This is an ERROR message with properties", new RuntimeException("Test exception"));
+		MDC.remove(propertyName);
+
+		assertTrue(testListener.getLatch().await(5, TimeUnit.SECONDS));
+		MessageProperties messageProperties = testListener.getMessageProperties();
+		assertNotNull(messageProperties);
+		assertNotNull(messageProperties.getHeaders().get(propertyName));
+		assertEquals(propertyValue, messageProperties.getHeaders().get(propertyName));
+		Object location = messageProperties.getHeaders().get("location");
+		assertNotNull(location);
+		assertThat(location, instanceOf(String.class));
+		assertThat((String) location,
+				startsWith("org.springframework.amqp.rabbit.logback.AmqpAppenderIntegrationTests.testAppenderWithProps()"));
+	}
+
+	@Test
+	public void testCharset() throws InterruptedException {
+		TestListener testListener = (TestListener) applicationContext.getBean("testListener", 1);
+		listenerContainer.setMessageListener(testListener);
+		listenerContainer.start();
+
+		String foo = "\u0fff"; // UTF-8 -> 0xe0bfbf
+		log.info(foo);
+		assertTrue(testListener.getLatch().await(5, TimeUnit.SECONDS));
+		byte[] body = testListener.getMessage().getBody();
+		int lineSeparatorExtraBytes = System.getProperty("line.separator").getBytes().length - 1;
+		assertEquals(0xe0, body[body.length - 5 - lineSeparatorExtraBytes] & 0xff);
+		assertEquals(0xbf, body[body.length - 4 - lineSeparatorExtraBytes] & 0xff);
+		assertEquals(0xbf, body[body.length - 3 - lineSeparatorExtraBytes] & 0xff);
+	}
+
+}

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

@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+
+	<appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+		<encoder>
+			<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+		</encoder>
+	</appender>
+
+	<appender name="AMQP" class="org.springframework.amqp.rabbit.logback.AmqpAppender">
+		<layout>
+			<pattern><![CDATA[ %d %p %t [%c] - <%m>%n ]]></pattern>
+		</layout>
+		<abbreviation>36</abbreviation>
+		<applicationId>AmqpAppenderTest</applicationId>
+		<routingKeyPattern>%property{applicationId}.%c.%p</routingKeyPattern>
+		<generateId>true</generateId>
+		<charset>UTF-8</charset>
+		<durable>false</durable>
+		<deliveryMode>NON_PERSISTENT</deliveryMode>
+	</appender>
+
+	<logger name="org.springframework.amqp.rabbit.logback" level="DEBUG" additivity="false">
+		<appender-ref ref="AMQP"/>
+	</logger>
+
+	<root level="INFO">
+		<appender-ref ref="CONSOLE"/>
+	</root>
+
+</configuration>

src/reference/docbook/whats-new.xml

@@ -66,6 +66,12 @@
 		</section>
 		<section>
 			<title>Log Appender</title>
+			<para>
+				The Logback <classname>org.springframework.amqp.rabbit.logback.AmqpAppender</classname>
+				has been introduced. It provides similar options like
+				<classname>org.springframework.amqp.rabbit.log4j.AmqpAppender</classname>.
+				For more info see JavaDocs of these classes.
+			</para>
 			<para>
 				The Log4j <classname>AmqpAppender</classname> now supports the
 				<code>deliveryMode</code> property (<code>PERSISTENT</code> or