Added filtering support to the AMQP consumer agent. (Updated logback-consumer.xml accordingly to exemplify.)

Author: cipriancraciun

logback-amqp-consumer/pom.xml

@@ -28,6 +28,11 @@
 			<artifactId>amqp-client</artifactId>
 			<version>2.2.0</version>
 		</dependency>
+		<dependency>
+			<groupId>janino</groupId>
+			<artifactId>janino</artifactId>
+			<version>2.5.10</version>
+		</dependency>
 	</dependencies>
 	<build>
 		<plugins>

logback-amqp-consumer/src/main/java/ch/qos/logback/amqp/AmqpConsumerAgent.java

@@ -5,6 +5,8 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
+import ch.qos.logback.core.spi.FilterReply;
+
 import ch.qos.logback.amqp.tools.Callbacks;
 import ch.qos.logback.amqp.tools.DefaultBinarySerializer;
 import ch.qos.logback.amqp.tools.DefaultContextAwareCallbacks;
@@ -13,6 +15,7 @@
 import ch.qos.logback.classic.Logger;
 import ch.qos.logback.classic.spi.ILoggingEvent;
 import ch.qos.logback.core.Context;
+import ch.qos.logback.core.filter.Filter;
 import ch.qos.logback.core.spi.ContextAwareBase;
 import ch.qos.logback.core.spi.LifeCycle;
 import org.slf4j.LoggerFactory;
@@ -32,6 +35,7 @@ public AmqpConsumerAgent ()
 		this.queue = AmqpConsumerAgent.defaultQueue;
 		this.routingKey = AmqpConsumerAgent.defaultRoutingKey;
 		this.serializer = new DefaultBinarySerializer ();
+		this.filter = null;
 		this.thread = null;
 		this.shutdownHook = null;
 		this.shouldStop = true;
@@ -44,6 +48,11 @@ public final String getExchange ()
 		return (this.exchange);
 	}
 
+	public final Filter<ILoggingEvent> getFilter ()
+	{
+		return (this.filter);
+	}
+	
 	public final String getHost ()
 	{
 		return (this.host);
@@ -124,6 +133,15 @@ public final void setExchange (final String exchange)
 		}
 	}
 
+	public final void setFilter (final Filter<ILoggingEvent> filter)
+	{
+		synchronized (this) {
+			if (this.thread != null)
+				throw (new IllegalStateException ("amqp appender is already started"));
+			this.filter = filter;
+		}
+	}
+	
 	public final void setHost (final String host)
 	{
 		synchronized (this) {
@@ -236,8 +254,8 @@ public final void run ()
 			this.thread.start ();
 			this.consumer =
 					new AmqpConsumer (
-							this.host, this.port, this.virtualHost, this.username, this.password,
-							this.exchange, this.queue, this.routingKey, this.callbacks, this.buffer);
+							this.host, this.port, this.virtualHost, this.username, this.password, this.exchange, this.queue,
+							this.routingKey, this.callbacks, this.buffer);
 			this.consumer.start ();
 			this.isStarted = true;
 			this.postStart ();
@@ -299,6 +317,9 @@ private final void loop ()
 				continue;
 			}
 			try {
+				if (this.filter != null)
+					if (this.filter.decide (event) == FilterReply.DENY)
+						continue;
 				this.process (event);
 			} catch (final Throwable exception) {
 				this.addError (
@@ -311,6 +332,7 @@ private final void loop ()
 	private final LinkedBlockingQueue<AmqpMessage> buffer;
 	private AmqpConsumer consumer;
 	private String exchange;
+	private Filter<ILoggingEvent> filter;
 	private String host;
 	private boolean isStarted;
 	private String password;

logback-amqp-consumer/src/test/config/logback-consumer.xml

@@ -13,6 +13,23 @@
 		<exchange>logback</exchange>
 		<queue></queue>
 		<routingKey>#</routingKey>
+		<!--
+		<filter class="ch.qos.logback.core.filter.EvaluatorFilter">
+			<evaluator class="ch.qos.logback.classic.boolex.JaninoEventEvaluator">
+				<expression>
+					final String applicationFilter = (String) System.getProperty ("amqp.consumer.agent.application_filter");
+					if ((applicationFilter == null) || applicationFilter.isEmpty ())
+						return (true);
+					final String applicationValue = (mdc != null) ? (String) mdc.get("application") : null;
+					if ((applicationValue == null) || applicationValue.isEmpty ())
+						return (true);
+					return (applicationFilter.equals (applicationValue));
+				</expression>
+			</evaluator>
+			<onMismatch>DENY</onMismatch>
+			<onMatch>NEUTRAL</onMatch>
+		</filter>
+		-->
 	</amqpConsumerAgent>
 
 	<appender name="sifter" class="ch.qos.logback.classic.sift.SiftingAppender">