AMQP-282 Log4j Appender - Configurable Charset

A charset is used when converting the message body
String to a byte[] to send to AMQP.

For backwards compatibility, the appender uses the
system charset by default, but now allows the charset
name to be specified in the log4j configuration...

log4j.appender.amqp.charset=UTF-8

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011-2012 by the original author(s).
+ * Copyright (c) 2011-2013 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
@@ -13,6 +13,7 @@
 
 package org.springframework.amqp.rabbit.log4j;
 
+import java.io.UnsupportedEncodingException;
 import java.util.Calendar;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -91,6 +92,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
  *   log4j.appender.amqp.contentType=text/plain
  *   log4j.appender.amqp.contentEncoding=null
  *   log4j.appender.amqp.generateId=false
+ *   log4j.appender.amqp.charset=null
  *   #-------------------------------
  *   ## Sender configuration
  *   #-------------------------------
@@ -235,6 +237,13 @@ public class AmqpAppender extends AppenderSkeleton {
 	 */
 	private final String mutex = "mutex";
 
+	/**
+	 * 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 boolean autoDelete = false;
@@ -384,6 +393,14 @@ public class AmqpAppender extends AppenderSkeleton {
 		this.generateId = generateId;
 	}
 
+	public String getCharset() {
+		return charset;
+	}
+
+	public void setCharset(String charset) {
+		this.charset = charset;
+	}
+
 	/**
 	 * Submit the required number of senders into the pool.
 	 */
@@ -522,8 +539,17 @@ public class AmqpAppender extends AppenderSkeleton {
 
 					// Send a message
 					try {
-						rabbitTemplate.send(exchangeName, routingKey, new Message(msgBody.toString().getBytes(),
-								amqpProps));
+						Message message = null;
+						if (AmqpAppender.this.charset != null) {
+							try {
+								message = new Message(msgBody.toString().getBytes(AmqpAppender.this.charset), amqpProps);
+							}
+							catch (UnsupportedEncodingException e) {/* fall back to default */}
+						}
+						if (message == null) {
+							message = new Message(msgBody.toString().getBytes(), amqpProps);
+						}
+						rabbitTemplate.send(exchangeName, routingKey, message);
 					}
 					catch (AmqpException e) {
 						int retries = event.incrementRetries();

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

@@ -106,6 +106,25 @@ public class AmqpAppenderIntegrationTests {
 		assertEquals(propertyValue, messageProperties.getHeaders().get(propertyName));
 	}
 
+	@Test
+	public void testCharset() throws InterruptedException {
+		Logger packageLogger = Logger.getLogger("org.springframework.amqp.rabbit.log4j");
+		AmqpAppender appender = (AmqpAppender) packageLogger.getAppender("amqp");
+		assertEquals("UTF-8", appender.getCharset());
+
+		TestListener testListener = (TestListener) applicationContext.getBean("testListener", 1);
+		listenerContainer.setMessageListener(testListener);
+		listenerContainer.start();
+
+		String foo = new String("\u0fff"); // UTF-8 -> 0xe0bfbf
+		log.info(foo);
+		assertTrue(testListener.getLatch().await(5, TimeUnit.SECONDS));
+		byte[] body = testListener.getMessage().getBody();
+		assertEquals(0xe0, body[body.length-5] & 0xff);
+		assertEquals(0xbf, body[body.length-4] & 0xff);
+		assertEquals(0xbf, body[body.length-3] & 0xff);
+	}
+
 	/*
 	 * When running as main(); should shutdown cleanly.
 	 */

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

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2011-2012 by the original author(s).
+ * Copyright (c) 2011-2013 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.
@@ -28,9 +28,9 @@ import org.springframework.amqp.core.MessageProperties;
  */
 public class TestListener implements MessageListener {
 
-  private CountDownLatch latch;
+  private final CountDownLatch latch;
 
-  private MessageProperties messageProperties;
+  private volatile Message message;
 
   public TestListener(int count) {
     latch = new CountDownLatch(count);
@@ -40,21 +40,28 @@ public class TestListener implements MessageListener {
     return latch;
   }
 
+  public Message getMessage() {
+	return message;
+  }
+
   public Object getId() {
-	if (this.messageProperties == null) {
+	if (this.message == null || this.getMessageProperties() == null) {
 		throw new IllegalStateException("No MessageProperties received");
 	}
-	return this.messageProperties.getMessageId();
+	return this.message.getMessageProperties().getMessageId();
   }
 
   public MessageProperties getMessageProperties() {
-	return this.messageProperties;
+    if (this.message == null) {
+      throw new IllegalStateException("No Message received");
+    }
+    return this.message.getMessageProperties();
   }
 
-public void onMessage(Message message) {
+  public void onMessage(Message message) {
     System.out.println("MESSAGE: " + message);
     System.out.println("BODY: " + new String(message.getBody()));
-    this.messageProperties = message.getMessageProperties();
+    this.message = message;
     latch.countDown();
   }
 

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

@@ -10,6 +10,7 @@ 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.appender.amqp.generateId=true
+log4j.appender.amqp.charset=UTF-8
 
 log4j.category.org.springframework.amqp.rabbit.log4j=DEBUG, amqp