AMQP-137: Added AMQP Log4J appender to spring-rabbit

spring-rabbit/pom.xml

@@ -1,5 +1,5 @@
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
 	<modelVersion>4.0.0</modelVersion>
 	<parent>
 		<groupId>org.springframework.amqp</groupId>
@@ -57,7 +57,8 @@
 		<dependency>
 			<groupId>log4j</groupId>
 			<artifactId>log4j</artifactId>
-			<scope>test</scope>
+			<scope>compile</scope>
+			<optional>true</optional>
 		</dependency>
 
 		<dependency>

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

@@ -0,0 +1,486 @@
+/*
+ * Copyright (c) 2011 by the original author(s).
+ *
+ * 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.log4j;
+
+import java.util.Calendar;
+import java.util.Map;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.log4j.AppenderSkeleton;
+import org.apache.log4j.Layout;
+import org.apache.log4j.Level;
+import org.apache.log4j.MDC;
+import org.apache.log4j.PatternLayout;
+import org.apache.log4j.spi.ErrorCode;
+import org.apache.log4j.spi.LocationInfo;
+import org.apache.log4j.spi.LoggingEvent;
+import org.apache.log4j.spi.ThrowableInformation;
+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.MessageProperties;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+
+/**
+ * A Log4J 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:
+ * </p>
+ * <pre><code>
+ *   log4j.appender.amqp=org.springframework.amqp.log4j.AmqpAppender
+ *   #-------------------------------
+ *   ## Connection settings
+ *   #-------------------------------
+ *   log4j.appender.amqp.host=localhost
+ *   log4j.appender.amqp.port=5672
+ *   log4j.appender.amqp.username=guest
+ *   log4j.appender.amqp.password=guest
+ *   log4j.appender.amqp.virtualHost=/
+ *   log4j.appender.amqp.connectionTimeout=0
+ *   #-------------------------------
+ *   ## Exchange name and type
+ *   #-------------------------------
+ *   log4j.appender.amqp.exchangeName=logs
+ *   log4j.appender.amqp.exchangeType=topic
+ *   #-------------------------------
+ *   ## What Log4J-format pattern to use to create a routing key
+ *   #-------------------------------
+ *   log4j.appender.amqp.routingKeyPattern=%c.%p
+ *   #-------------------------------
+ *   ## Whether or not to declare this configured exchange
+ *   #-------------------------------
+ *   log4j.appender.amqp.declareExchange=false
+ *   #-------------------------------
+ *   ## Message properties
+ *   #-------------------------------
+ *   log4j.appender.amqp.contentType=text/plain
+ *   log4j.appender.amqp.contentEncoding=null
+ *   #-------------------------------
+ *   ## Sender configuration
+ *   #-------------------------------
+ *   log4j.appender.amqp.senderPoolSize=2
+ *   log4j.appender.amqp.maxSenderRetries=30
+ *   log4j.appender.amqp.applicationId=null
+ *   #-------------------------------
+ *   ## Standard Log4J stuff
+ *   #-------------------------------
+ *   log4j.appender.amqp.layout=org.apache.log4j.PatternLayout
+ *   log4j.appender.amqp.layout.ConversionPattern=%d %p %t [%c] - <%m>%n
+ * </code></pre>
+ *
+ * @author Jon Brisbin <jbrisbin@vmware.com>
+ */
+public class AmqpAppender extends AppenderSkeleton {
+
+  protected static final String APPLICATION_ID = "applicationId";
+  protected static final String CATEGORY_NAME = "categoryName";
+  protected static final String CATEGORY_LEVEL = "level";
+
+  /**
+   * Name of the exchange to publish log events to.
+   */
+  protected String exchangeName = "logs";
+  /**
+   * Type of the exchange to publish log events to.
+   */
+  protected String exchangeType = "topic";
+  /**
+   * Log4J pattern format to use to generate a routing key.
+   */
+  protected String routingKeyPattern = "%c.%p";
+  /**
+   * Log4J Layout to use to generate routing key.
+   */
+  protected Layout routingKeyLayout = new PatternLayout(routingKeyPattern);
+  /**
+   * Whether or not we've tried to declare this exchange yet.
+   */
+  protected AtomicBoolean exchangeDeclared = new AtomicBoolean(false);
+  /**
+   * How long to wait for a connection to time out.
+   */
+  protected int connectionTimeout = 0;
+  /**
+   * Configuration arbitrary application ID.
+   */
+  protected String applicationId = null;
+  /**
+   * Where LoggingEvents are queued to send.
+   */
+  protected LinkedBlockingQueue<Event> events = new LinkedBlockingQueue<Event>();
+  /**
+   * The pool of senders.
+   */
+  protected ExecutorService senderPool = null;
+  /**
+   * How many senders to use at once. Use more senders if you have lots of log output going through this appender.
+   */
+  protected int senderPoolSize = 2;
+  /**
+   * How many times to retry sending a message if the broker is unavailable or there is some other error.
+   */
+  protected int maxSenderRetries = 30;
+  /**
+   * Retries are delayed like: N ^ log(N), where N is the retry number.
+   */
+  protected Timer retryTimer = new Timer("log-event-retry-delay", true);
+  /**
+   * RabbitMQ ConnectionFactory.
+   */
+  protected CachingConnectionFactory connectionFactory;
+  /**
+   * RabbitMQ host to connect to.
+   */
+  protected String host = "localhost";
+  /**
+   * RabbitMQ virtual host to connect to.
+   */
+  protected String virtualHost = "/";
+  /**
+   * RabbitMQ port to connect to.
+   */
+  protected int port = 5672;
+  /**
+   * RabbitMQ user to connect as.
+   */
+  protected String username = "guest";
+  /**
+   * RabbitMQ password for this user.
+   */
+  protected String password = "guest";
+  /**
+   * Default content-type of log messages.
+   */
+  protected String contentType = "text/plain";
+  /**
+   * Default content-encoding of log messages.
+   */
+  protected String contentEncoding = null;
+  /**
+   * Whether or not to try and declare the configured exchange when this appender starts.
+   */
+  protected boolean declareExchange = false;
+  /**
+   * Used to synchronize access when creating the RabbitMQ ConnectionFactory.
+   */
+  protected final String mutex = "mutex";
+
+  public AmqpAppender() {
+  }
+
+  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 routingKeyPattern;
+  }
+
+  public void setRoutingKeyPattern(String routingKeyPattern) {
+    this.routingKeyPattern = routingKeyPattern;
+    this.routingKeyLayout = new PatternLayout(routingKeyPattern);
+  }
+
+  public boolean isDeclareExchange() {
+    return declareExchange;
+  }
+
+  public void setDeclareExchange(boolean declareExchange) {
+    this.declareExchange = declareExchange;
+  }
+
+  public int getConnectionTimeout() {
+    return connectionTimeout;
+  }
+
+  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;
+  }
+
+  /**
+   * Submit the required number of senders into the pool.
+   */
+  protected void startSenders() {
+    senderPool = Executors.newCachedThreadPool();
+    for (int i = 0; i < senderPoolSize; i++) {
+      senderPool.submit(new EventSender());
+    }
+  }
+
+  /**
+   * Maybe declare the exchange.
+   */
+  protected void maybeDeclareExchange() {
+    RabbitAdmin admin = new RabbitAdmin(connectionFactory);
+    if (declareExchange) {
+      Exchange x;
+      if ("topic".equals(exchangeType)) {
+        x = new TopicExchange(exchangeName, true, false);
+      } else if ("direct".equals(exchangeType)) {
+        x = new DirectExchange(exchangeName, true, false);
+      } else if ("fanout".equals(exchangeType)) {
+        x = new FanoutExchange(exchangeName, true, false);
+      } else if ("headers".equals(exchangeType)) {
+        x = new HeadersExchange(exchangeType, true, false);
+      } else {
+        x = new TopicExchange(exchangeName);
+      }
+      //admin.deleteExchange(exchangeName);
+      admin.declareExchange(x);
+    }
+  }
+
+  @Override
+  public void append(LoggingEvent event) {
+    if (null == senderPool) {
+      synchronized (mutex) {
+        connectionFactory = new CachingConnectionFactory();
+        connectionFactory.setHost(host);
+        connectionFactory.setPort(port);
+        connectionFactory.setUsername(username);
+        connectionFactory.setPassword(password);
+        connectionFactory.setVirtualHost(virtualHost);
+        maybeDeclareExchange();
+        exchangeDeclared.set(true);
+
+        startSenders();
+      }
+    }
+    events.add(new Event(event, event.getProperties()));
+  }
+
+  public void close() {
+    if (null != senderPool) {
+      senderPool.shutdownNow();
+      senderPool = null;
+    }
+  }
+
+  public boolean requiresLayout() {
+    return true;
+  }
+
+  /**
+   * Helper class to actually send LoggingEvents asynchronously.
+   */
+  protected class EventSender implements Runnable {
+    public void run() {
+      try {
+        RabbitTemplate rabbitTmpl = new RabbitTemplate(connectionFactory);
+        while (true) {
+          final Event event = events.take();
+          LoggingEvent logEvent = event.getEvent();
+
+          String name = logEvent.getLogger().getName();
+          Level level = logEvent.getLevel();
+
+          MessageProperties amqpProps = new MessageProperties();
+          amqpProps.setContentType(contentType);
+          if (null != contentEncoding) {
+            amqpProps.setContentEncoding(contentEncoding);
+          }
+          amqpProps.setHeader(CATEGORY_NAME, name);
+          amqpProps.setHeader(CATEGORY_LEVEL, level.toString());
+
+          // Set applicationId, if we're using one
+          if (null != applicationId) {
+            amqpProps.setAppId(applicationId);
+            MDC.put(APPLICATION_ID, applicationId);
+          }
+
+          // Set timestamp
+          Calendar tstamp = Calendar.getInstance();
+          tstamp.setTimeInMillis(logEvent.getTimeStamp());
+          amqpProps.setTimestamp(tstamp.getTime());
+
+          // Copy properties in from MDC
+          Map props = event.getProperties();
+          for (Object key : event.getProperties().entrySet()) {
+            amqpProps.setHeader(key.toString(), props.get(key));
+          }
+          LocationInfo locInfo = logEvent.getLocationInformation();
+          if (!"?".equals(locInfo.getClassName())) {
+            amqpProps.setHeader("location", String.format("%s.%s()[%s]", locInfo.getClassName(), locInfo.getMethodName(), locInfo.getLineNumber()));
+          }
+
+          StringBuffer msgBody = new StringBuffer(String.format("%s%n", logEvent.getRenderedMessage()));
+          if (null != logEvent.getThrowableInformation()) {
+            ThrowableInformation tinfo = logEvent.getThrowableInformation();
+            for (String line : tinfo.getThrowableStrRep()) {
+              msgBody.append(String.format("%s%n", line));
+            }
+          }
+
+          // Send a message
+          String routingKey = routingKeyLayout.format(logEvent);
+          try {
+            rabbitTmpl.send(exchangeName, routingKey, new Message(msgBody.toString().getBytes(), amqpProps));
+          } 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 {
+              errorHandler.error("Could not send log message " + logEvent.getRenderedMessage() + " after " + maxSenderRetries + " retries",
+                  e,
+                  ErrorCode.WRITE_FAILURE,
+                  logEvent);
+            }
+          } finally {
+            if (null != applicationId) {
+              MDC.remove(APPLICATION_ID);
+            }
+          }
+        }
+      } 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 LoggingEvent event;
+    final Map properties;
+    AtomicInteger retries = new AtomicInteger(0);
+
+    public Event(LoggingEvent event, Map properties) {
+      this.event = event;
+      this.properties = properties;
+    }
+
+    public LoggingEvent getEvent() {
+      return event;
+    }
+
+    public Map getProperties() {
+      return properties;
+    }
+
+    public int incrementRetries() {
+      return retries.incrementAndGet();
+    }
+  }
+
+}

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

@@ -0,0 +1,93 @@
+/*
+ * Copyright (c) 2011 by the original author(s).
+ *
+ * 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.log4j;
+
+import org.springframework.amqp.core.AcknowledgeMode;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.rabbit.connection.SingleConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitAdmin;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.beans.factory.config.BeanDefinition;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Scope;
+
+/**
+ * @author Jon Brisbin <jbrisbin@vmware.com>
+ */
+@Configuration
+public class AmqpAppenderConfiguration {
+
+  static {
+    //DOMConfigurator.configure(AmqpAppenderTests.class.getResource("/log4j.xml"));
+  }
+
+  static final String QUEUE = "amqp.appender.test";
+  static final String EXCHANGE = "logs";
+  static final String ROUTING_KEY = "AmqpAppenderTest.#";
+
+  @Bean
+  public SingleConnectionFactory connectionFactory() {
+    return new SingleConnectionFactory();
+  }
+
+  @Bean
+  public TopicExchange testExchange() {
+    return new TopicExchange(EXCHANGE, true, false);
+  }
+
+  @Bean
+  public Queue testQueue() {
+    return new Queue(QUEUE);
+  }
+
+  @Bean
+  public Binding testBinding() {
+    return new Binding(testQueue(), testExchange(), ROUTING_KEY);
+  }
+
+  @Bean
+  public RabbitAdmin rabbitAdmin() {
+    return new RabbitAdmin(connectionFactory());
+  }
+
+  @Bean
+  @Scope(BeanDefinition.SCOPE_PROTOTYPE)
+  public SimpleMessageListenerContainer listenerContainer() {
+    SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory());
+    Queue q = testQueue();
+
+    RabbitAdmin admin = rabbitAdmin();
+    admin.declareQueue(q);
+    admin.declareBinding(testBinding());
+
+    container.setQueues(q);
+    //container.setMessageListener(testListener(4));
+    container.setAutoStartup(false);
+    container.setAcknowledgeMode(AcknowledgeMode.AUTO);
+
+    return container;
+  }
+
+  @Bean
+  @Scope(BeanDefinition.SCOPE_PROTOTYPE)
+  public TestListener testListener(int count) {
+    return new TestListener(count);
+  }
+}

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

@@ -0,0 +1,94 @@
+/*
+ * Copyright (c) 2011 by the original author(s).
+ *
+ * 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.log4j;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.log4j.Logger;
+import org.apache.log4j.MDC;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationContext;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * @author Jon Brisbin <jbrisbin@vmware.com>
+ */
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration(
+    locations = {
+        "org.springframework.amqp.rabbit.log4j"
+    },
+    loader = AnnotationConfigContextLoader.class
+)
+public class AmqpAppenderTests {
+
+  @Autowired
+  ApplicationContext applicationContext;
+
+  Logger log;
+  SimpleMessageListenerContainer listenerContainer;
+
+  @Before
+  public void setUp() {
+    log = Logger.getLogger(getClass());
+    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();
+
+    Logger log = Logger.getLogger(getClass());
+
+    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"));
+
+    testListener.getLatch().await(5, TimeUnit.SECONDS);
+  }
+
+  @Test
+  public void testAppenderWithProps() throws InterruptedException {
+    TestListener testListener = (TestListener) applicationContext.getBean("testListener", 4);
+    listenerContainer.setMessageListener(testListener);
+    listenerContainer.start();
+
+    MDC.put("someproperty", "property.value");
+    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("someproperty");
+
+    testListener.getLatch().await(5, TimeUnit.SECONDS);
+  }
+
+}

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

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2011 by the original author(s).
+ *
+ * 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.log4j;
+
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.test.context.ContextLoader;
+
+/**
+ * @author Jon Brisbin <jbrisbin@vmware.com>
+ */
+public class AnnotationConfigContextLoader implements ContextLoader {
+  public String[] processLocations(Class<?> clazz, String... locations) {
+    return locations;
+  }
+
+  public ApplicationContext loadContext(String... locations) throws Exception {
+    return new AnnotationConfigApplicationContext(locations);
+  }
+}

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

@@ -0,0 +1,45 @@
+/*
+ * Copyright (c) 2011 by the original author(s).
+ *
+ * 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.log4j;
+
+import java.util.concurrent.CountDownLatch;
+
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageListener;
+
+/**
+ * @author Jon Brisbin <jbrisbin@vmware.com>
+ */
+public class TestListener implements MessageListener {
+
+  private CountDownLatch latch;
+
+  public TestListener(int count) {
+    latch = new CountDownLatch(count);
+  }
+
+  public CountDownLatch getLatch() {
+    return latch;
+  }
+
+  public void onMessage(Message message) {
+    System.out.println("MESSAGE: " + message);
+    System.out.println("BODY: " + new String(message.getBody()));
+    latch.countDown();
+  }
+
+}

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

@@ -4,6 +4,14 @@ log4j.appender.stdout=org.apache.log4j.ConsoleAppender
 log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
 log4j.appender.stdout.layout.ConversionPattern=%d %p %t [%c] - <%m>%n
 
+log4j.appender.amqp=org.springframework.amqp.rabbit.log4j.AmqpAppender
+log4j.appender.amqp.applicationId=AmqpAppenderTest
+log4j.appender.amqp.routingKeyPattern=%X{applicationId}.%c.%p
+log4j.appender.amqp.layout=org.apache.log4j.PatternLayout
+log4j.appender.amqp.layout.ConversionPattern=%d %p %t [%c] - <%m>%n
+
+log4j.category.org.springframework.amqp.rabbit.log4j=DEBUG, amqp
+
 log4j.category.org.springframework.amqp.rabbit=DEBUG
 log4j.category.org.springframework.beans.factory=INFO