AMQP-137: Align logger test and appender internals with project norms
This commit is contained in:
Dave Syer
@@ -1,17 +1,14 @@
|
||||
/*
|
||||
* 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.
|
||||
*
|
||||
* 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;
|
||||
@@ -52,7 +49,9 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
* <p>
|
||||
* A fully-configured AmqpAppender, with every option set to their defaults, would look like this:
|
||||
* </p>
|
||||
* <pre><code>
|
||||
*
|
||||
* <pre>
|
||||
* <code>
|
||||
* log4j.appender.amqp=org.springframework.amqp.log4j.AmqpAppender
|
||||
* #-------------------------------
|
||||
* ## Connection settings
|
||||
@@ -69,7 +68,8 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
* log4j.appender.amqp.exchangeName=logs
|
||||
* log4j.appender.amqp.exchangeType=topic
|
||||
* #-------------------------------
|
||||
* ## What Log4J-format pattern to use to create a routing key
|
||||
* ## Log4J-format pattern to use to create a routing key.
|
||||
* ## The application id is available as %X{applicationId}.
|
||||
* #-------------------------------
|
||||
* log4j.appender.amqp.routingKeyPattern=%c.%p
|
||||
* #-------------------------------
|
||||
@@ -92,395 +92,411 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
* #-------------------------------
|
||||
* log4j.appender.amqp.layout=org.apache.log4j.PatternLayout
|
||||
* log4j.appender.amqp.layout.ConversionPattern=%d %p %t [%c] - <%m>%n
|
||||
* </code></pre>
|
||||
*
|
||||
* </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";
|
||||
/**
|
||||
* 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";
|
||||
|
||||
/**
|
||||
* 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";
|
||||
/**
|
||||
* Key name for the logger category name in the message properties
|
||||
*/
|
||||
public static final String CATEGORY_NAME = "categoryName";
|
||||
|
||||
public AmqpAppender() {
|
||||
}
|
||||
/**
|
||||
* Key name for the logger level name in the message properties
|
||||
*/
|
||||
public static final String CATEGORY_LEVEL = "level";
|
||||
|
||||
public String getHost() {
|
||||
return host;
|
||||
}
|
||||
/**
|
||||
* 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";
|
||||
/**
|
||||
* Log4J pattern format to use to generate a routing key.
|
||||
*/
|
||||
private String routingKeyPattern = "%c.%p";
|
||||
/**
|
||||
* Log4J Layout to use to generate routing key.
|
||||
*/
|
||||
private Layout routingKeyLayout = new PatternLayout(routingKeyPattern);
|
||||
/**
|
||||
* Whether or not we've tried to declare this exchange yet.
|
||||
*/
|
||||
private AtomicBoolean exchangeDeclared = new AtomicBoolean(false);
|
||||
/**
|
||||
* How long to wait for a connection to time out.
|
||||
*/
|
||||
private int connectionTimeout = 0;
|
||||
/**
|
||||
* Configuration arbitrary application ID.
|
||||
*/
|
||||
private String applicationId = null;
|
||||
/**
|
||||
* Where LoggingEvents are queued to send.
|
||||
*/
|
||||
private 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 Timer retryTimer = new Timer("log-event-retry-delay", true);
|
||||
/**
|
||||
* RabbitMQ ConnectionFactory.
|
||||
*/
|
||||
private CachingConnectionFactory 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;
|
||||
/**
|
||||
* Used to synchronize access when creating the RabbitMQ ConnectionFactory.
|
||||
*/
|
||||
private final String mutex = "mutex";
|
||||
|
||||
public void setHost(String host) {
|
||||
this.host = host;
|
||||
}
|
||||
public AmqpAppender() {
|
||||
}
|
||||
|
||||
public int getPort() {
|
||||
return port;
|
||||
}
|
||||
public String getHost() {
|
||||
return host;
|
||||
}
|
||||
|
||||
public void setPort(int port) {
|
||||
this.port = port;
|
||||
}
|
||||
public void setHost(String host) {
|
||||
this.host = host;
|
||||
}
|
||||
|
||||
public String getVirtualHost() {
|
||||
return virtualHost;
|
||||
}
|
||||
public int getPort() {
|
||||
return port;
|
||||
}
|
||||
|
||||
public void setVirtualHost(String virtualHost) {
|
||||
this.virtualHost = virtualHost;
|
||||
}
|
||||
public void setPort(int port) {
|
||||
this.port = port;
|
||||
}
|
||||
|
||||
public String getUsername() {
|
||||
return username;
|
||||
}
|
||||
public String getVirtualHost() {
|
||||
return virtualHost;
|
||||
}
|
||||
|
||||
public void setUsername(String username) {
|
||||
this.username = username;
|
||||
}
|
||||
public void setVirtualHost(String virtualHost) {
|
||||
this.virtualHost = virtualHost;
|
||||
}
|
||||
|
||||
public String getPassword() {
|
||||
return password;
|
||||
}
|
||||
public String getUsername() {
|
||||
return username;
|
||||
}
|
||||
|
||||
public void setPassword(String password) {
|
||||
this.password = password;
|
||||
}
|
||||
public void setUsername(String username) {
|
||||
this.username = username;
|
||||
}
|
||||
|
||||
public String getExchangeName() {
|
||||
return exchangeName;
|
||||
}
|
||||
public String getPassword() {
|
||||
return password;
|
||||
}
|
||||
|
||||
public void setExchangeName(String exchangeName) {
|
||||
this.exchangeName = exchangeName;
|
||||
}
|
||||
public void setPassword(String password) {
|
||||
this.password = password;
|
||||
}
|
||||
|
||||
public String getExchangeType() {
|
||||
return exchangeType;
|
||||
}
|
||||
public String getExchangeName() {
|
||||
return exchangeName;
|
||||
}
|
||||
|
||||
public void setExchangeType(String exchangeType) {
|
||||
this.exchangeType = exchangeType;
|
||||
}
|
||||
public void setExchangeName(String exchangeName) {
|
||||
this.exchangeName = exchangeName;
|
||||
}
|
||||
|
||||
public String getRoutingKeyPattern() {
|
||||
return routingKeyPattern;
|
||||
}
|
||||
public String getExchangeType() {
|
||||
return exchangeType;
|
||||
}
|
||||
|
||||
public void setRoutingKeyPattern(String routingKeyPattern) {
|
||||
this.routingKeyPattern = routingKeyPattern;
|
||||
this.routingKeyLayout = new PatternLayout(routingKeyPattern);
|
||||
}
|
||||
public void setExchangeType(String exchangeType) {
|
||||
this.exchangeType = exchangeType;
|
||||
}
|
||||
|
||||
public boolean isDeclareExchange() {
|
||||
return declareExchange;
|
||||
}
|
||||
public String getRoutingKeyPattern() {
|
||||
return routingKeyPattern;
|
||||
}
|
||||
|
||||
public void setDeclareExchange(boolean declareExchange) {
|
||||
this.declareExchange = declareExchange;
|
||||
}
|
||||
public void setRoutingKeyPattern(String routingKeyPattern) {
|
||||
this.routingKeyPattern = routingKeyPattern;
|
||||
this.routingKeyLayout = new PatternLayout(routingKeyPattern);
|
||||
}
|
||||
|
||||
public int getConnectionTimeout() {
|
||||
return connectionTimeout;
|
||||
}
|
||||
public boolean isDeclareExchange() {
|
||||
return declareExchange;
|
||||
}
|
||||
|
||||
public String getContentType() {
|
||||
return contentType;
|
||||
}
|
||||
public void setDeclareExchange(boolean declareExchange) {
|
||||
this.declareExchange = declareExchange;
|
||||
}
|
||||
|
||||
public void setContentType(String contentType) {
|
||||
this.contentType = contentType;
|
||||
}
|
||||
public int getConnectionTimeout() {
|
||||
return connectionTimeout;
|
||||
}
|
||||
|
||||
public String getContentEncoding() {
|
||||
return contentEncoding;
|
||||
}
|
||||
public String getContentType() {
|
||||
return contentType;
|
||||
}
|
||||
|
||||
public void setContentEncoding(String contentEncoding) {
|
||||
this.contentEncoding = contentEncoding;
|
||||
}
|
||||
public void setContentType(String contentType) {
|
||||
this.contentType = contentType;
|
||||
}
|
||||
|
||||
public String getApplicationId() {
|
||||
return applicationId;
|
||||
}
|
||||
public String getContentEncoding() {
|
||||
return contentEncoding;
|
||||
}
|
||||
|
||||
public void setApplicationId(String applicationId) {
|
||||
this.applicationId = applicationId;
|
||||
}
|
||||
public void setContentEncoding(String contentEncoding) {
|
||||
this.contentEncoding = contentEncoding;
|
||||
}
|
||||
|
||||
public int getSenderPoolSize() {
|
||||
return senderPoolSize;
|
||||
}
|
||||
public String getApplicationId() {
|
||||
return applicationId;
|
||||
}
|
||||
|
||||
public void setSenderPoolSize(int senderPoolSize) {
|
||||
this.senderPoolSize = senderPoolSize;
|
||||
}
|
||||
public void setApplicationId(String applicationId) {
|
||||
this.applicationId = applicationId;
|
||||
}
|
||||
|
||||
public int getMaxSenderRetries() {
|
||||
return maxSenderRetries;
|
||||
}
|
||||
public int getSenderPoolSize() {
|
||||
return senderPoolSize;
|
||||
}
|
||||
|
||||
public void setMaxSenderRetries(int maxSenderRetries) {
|
||||
this.maxSenderRetries = maxSenderRetries;
|
||||
}
|
||||
public void setSenderPoolSize(int senderPoolSize) {
|
||||
this.senderPoolSize = senderPoolSize;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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());
|
||||
}
|
||||
}
|
||||
public int getMaxSenderRetries() {
|
||||
return maxSenderRetries;
|
||||
}
|
||||
|
||||
/**
|
||||
* 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);
|
||||
}
|
||||
}
|
||||
public void setMaxSenderRetries(int maxSenderRetries) {
|
||||
this.maxSenderRetries = maxSenderRetries;
|
||||
}
|
||||
|
||||
@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);
|
||||
/**
|
||||
* 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());
|
||||
}
|
||||
}
|
||||
|
||||
startSenders();
|
||||
}
|
||||
}
|
||||
events.add(new Event(event, event.getProperties()));
|
||||
}
|
||||
/**
|
||||
* 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);
|
||||
}
|
||||
}
|
||||
|
||||
public void close() {
|
||||
if (null != senderPool) {
|
||||
senderPool.shutdownNow();
|
||||
senderPool = null;
|
||||
}
|
||||
}
|
||||
@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);
|
||||
|
||||
public boolean requiresLayout() {
|
||||
return true;
|
||||
}
|
||||
startSenders();
|
||||
}
|
||||
}
|
||||
events.add(new Event(event, event.getProperties()));
|
||||
}
|
||||
|
||||
/**
|
||||
* 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();
|
||||
public void close() {
|
||||
if (null != senderPool) {
|
||||
senderPool.shutdownNow();
|
||||
senderPool = null;
|
||||
}
|
||||
}
|
||||
|
||||
String name = logEvent.getLogger().getName();
|
||||
Level level = logEvent.getLevel();
|
||||
public boolean requiresLayout() {
|
||||
return true;
|
||||
}
|
||||
|
||||
MessageProperties amqpProps = new MessageProperties();
|
||||
amqpProps.setContentType(contentType);
|
||||
if (null != contentEncoding) {
|
||||
amqpProps.setContentEncoding(contentEncoding);
|
||||
}
|
||||
amqpProps.setHeader(CATEGORY_NAME, name);
|
||||
amqpProps.setHeader(CATEGORY_LEVEL, level.toString());
|
||||
/**
|
||||
* 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();
|
||||
|
||||
// Set applicationId, if we're using one
|
||||
if (null != applicationId) {
|
||||
amqpProps.setAppId(applicationId);
|
||||
MDC.put(APPLICATION_ID, applicationId);
|
||||
}
|
||||
String name = logEvent.getLogger().getName();
|
||||
Level level = logEvent.getLevel();
|
||||
|
||||
// Set timestamp
|
||||
Calendar tstamp = Calendar.getInstance();
|
||||
tstamp.setTimeInMillis(logEvent.getTimeStamp());
|
||||
amqpProps.setTimestamp(tstamp.getTime());
|
||||
MessageProperties amqpProps = new MessageProperties();
|
||||
amqpProps.setContentType(contentType);
|
||||
if (null != contentEncoding) {
|
||||
amqpProps.setContentEncoding(contentEncoding);
|
||||
}
|
||||
amqpProps.setHeader(CATEGORY_NAME, name);
|
||||
amqpProps.setHeader(CATEGORY_LEVEL, level.toString());
|
||||
|
||||
// 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()));
|
||||
}
|
||||
// Set applicationId, if we're using one
|
||||
if (null != applicationId) {
|
||||
amqpProps.setAppId(applicationId);
|
||||
MDC.put(APPLICATION_ID, applicationId);
|
||||
}
|
||||
|
||||
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));
|
||||
}
|
||||
}
|
||||
// Set timestamp
|
||||
Calendar tstamp = Calendar.getInstance();
|
||||
tstamp.setTimeInMillis(logEvent.getTimeStamp());
|
||||
amqpProps.setTimestamp(tstamp.getTime());
|
||||
|
||||
// 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
// Copy properties in from MDC
|
||||
@SuppressWarnings("rawtypes")
|
||||
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()));
|
||||
}
|
||||
|
||||
/**
|
||||
* 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);
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
||||
public Event(LoggingEvent event, Map properties) {
|
||||
this.event = event;
|
||||
this.properties = properties;
|
||||
}
|
||||
// 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);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public LoggingEvent getEvent() {
|
||||
return event;
|
||||
}
|
||||
/**
|
||||
* Small helper class to encapsulate a LoggingEvent, its MDC properties, and the number of retries.
|
||||
*/
|
||||
@SuppressWarnings("rawtypes")
|
||||
protected class Event {
|
||||
final LoggingEvent event;
|
||||
final Map properties;
|
||||
AtomicInteger retries = new AtomicInteger(0);
|
||||
|
||||
public Map getProperties() {
|
||||
return properties;
|
||||
}
|
||||
public Event(LoggingEvent event, Map properties) {
|
||||
this.event = event;
|
||||
this.properties = properties;
|
||||
}
|
||||
|
||||
public int incrementRetries() {
|
||||
return retries.incrementAndGet();
|
||||
}
|
||||
}
|
||||
public LoggingEvent getEvent() {
|
||||
return event;
|
||||
}
|
||||
|
||||
public Map getProperties() {
|
||||
return properties;
|
||||
}
|
||||
|
||||
public int incrementRetries() {
|
||||
return retries.incrementAndGet();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
|
||||
import org.springframework.amqp.rabbit.test.BrokerRunning;
|
||||
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;
|
||||
import org.springframework.util.Log4jConfigurer;
|
||||
|
||||
/**
|
||||
* @author Jon Brisbin <jbrisbin@vmware.com>
|
||||
*/
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@ContextConfiguration(locations = { "org.springframework.amqp.rabbit.log4j" }, loader = AnnotationConfigContextLoader.class)
|
||||
public class AmqpAppenderIntegrationTests {
|
||||
|
||||
@Rule
|
||||
public BrokerRunning brokerIsRunning = BrokerRunning.isRunning();
|
||||
|
||||
@Autowired
|
||||
private ApplicationContext applicationContext;
|
||||
|
||||
private Logger log;
|
||||
|
||||
private SimpleMessageListenerContainer listenerContainer;
|
||||
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
Log4jConfigurer.initLogging("classpath:log4j-amqp.properties");
|
||||
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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -1,94 +0,0 @@
|
||||
/*
|
||||
* 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);
|
||||
}
|
||||
|
||||
}
|
||||
17
spring-rabbit/src/test/resources/log4j-amqp.properties
Normal file
17
spring-rabbit/src/test/resources/log4j-amqp.properties
Normal file
@@ -0,0 +1,17 @@
|
||||
log4j.rootCategory=INFO, stdout
|
||||
|
||||
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
|
||||
|
||||
@@ -4,14 +4,6 @@ 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.amqp.rabbit=INFO
|
||||
log4j.category.org.springframework.beans.factory=INFO
|
||||
|
||||
|
||||
Reference in New Issue
Block a user