Added support for custom routing key inside AMQP consumer.

Author: cipriancraciun

logback-amqp-appender/src/test/java/ch/qos/logback/amqp/tests/AmqpAppenderTests.java

@@ -37,7 +37,11 @@ public final void testAppender ()
 		for (int index = 0; index < AmqpAppenderTests.messageCount; index++) {
 			MDC.put (DefaultMutator.applicationKey, String.format ("app-%d", index % 3 + 1));
 			MDC.put (DefaultMutator.componentKey, String.format ("comp-%d", index % 2 + 1));
-			testLogger.error (UUID.randomUUID ().toString ());
+			final String message = UUID.randomUUID ().toString ();
+			if (index % 4 != 0)
+				testLogger.error (message);
+			else
+				testLogger.error (message, new Throwable (message));
 		}
 
 		realLogger.debug ("waiting for message draining (i.e. until their all sent)");

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

@@ -18,12 +18,13 @@ public final class AmqpConsumer
 {
 	public AmqpConsumer (
 			final String host, final Integer port, final String virtualHost, final String username, final String password,
-			final String[] exchanges, final String queue, final Callbacks callbacks,
+			final String exchange, final String queue, final String routingKey, final Callbacks callbacks,
 			final LinkedBlockingQueue<AmqpMessage> sink)
 	{
 		super (host, port, virtualHost, username, password, callbacks);
-		this.exchanges = exchanges;
+		this.exchange = exchange;
 		this.queue = queue;
+		this.routingKey = routingKey;
 		this.queue1 = null;
 		this.sink = sink;
 	}
@@ -98,14 +99,14 @@ private final void consume (final Envelope envelope, final BasicProperties prope
 	private final boolean declare ()
 	{
 		final Channel channel = this.getChannel ();
-		for (final String exchange : this.exchanges) {
-			this.callbacks.handleLogEvent (Level.INFO, null, "amqp consumer declaring the exchange `%s`", exchange);
+		{
+			this.callbacks.handleLogEvent (Level.INFO, null, "amqp consumer declaring the exchange `%s`", this.exchange);
 			try {
-				channel.exchangeDeclare (exchange, "topic", true, false, null);
+				channel.exchangeDeclare (this.exchange, "topic", true, false, null);
 			} catch (final Throwable exception) {
 				this.callbacks.handleException (
 						exception, "amqp consumer encountered an error while declaring the exchange `%s`; aborting!",
-						exchange);
+						this.exchange);
 				return (false);
 			}
 		}
@@ -129,16 +130,16 @@ private final boolean declare ()
 				return (false);
 			}
 		}
-		for (final String exchange : this.exchanges) {
+		{
 			this.callbacks.handleLogEvent (
-					Level.INFO, null, "amqp consumer binding the queue `%s` to exchange `%s`", this.queue1, exchange);
+					Level.INFO, null, "amqp consumer binding the queue `%s` to exchange `%s` with routing key `%s`", this.queue1, this.exchange, this.routingKey);
 			try {
-				channel.queueBind (this.queue1, exchange, "#", null);
+				channel.queueBind (this.queue1, this.exchange, this.routingKey, null);
 			} catch (final Throwable exception) {
 				this.callbacks.handleException (
 						exception,
-						"amqp consumer encountered an error while binding the queue `%s` to exchange `%s`; aborting!",
-						this.queue1, exchange);
+						"amqp consumer encountered an error while binding the queue `%s` to exchange `%s` with routing key `%s`; aborting!",
+						this.queue1, this.exchange, this.routingKey);
 				return (false);
 			}
 		}
@@ -159,8 +160,9 @@ private final boolean register ()
 		}
 	}
 
-	private final String[] exchanges;
+	private final String exchange;
 	private final String queue;
+	private final String routingKey;
 	private String queue1;
 	private final LinkedBlockingQueue<AmqpMessage> sink;
 

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

@@ -237,7 +237,7 @@ public final void run ()
 			this.consumer =
 					new AmqpConsumer (
 							this.host, this.port, this.virtualHost, this.username, this.password,
-							new String[] {this.exchange}, this.queue, this.callbacks, this.buffer);
+							this.exchange, this.queue, this.routingKey, this.callbacks, this.buffer);
 			this.consumer.start ();
 			this.isStarted = true;
 			this.postStart ();

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

@@ -12,6 +12,7 @@
 		<password>guest</password>
 		<exchange>logback</exchange>
 		<queue></queue>
+		<routingKey>#</routingKey>
 	</amqpConsumerAgent>
 
 	<appender name="console" class="ch.qos.logback.core.ConsoleAppender">