Added support for anonymous consumer queues (just specify an empty string as the queue). (Minor bug fix.)

Author: cipriancraciun

logback-amqp-appender/src/main/java/ch/qos/logback/amqp/AmqpPublisher.java

@@ -29,8 +29,6 @@ protected final void loop ()
 				synchronized (this) {
 					if (this.shouldStopLoop ())
 						break loop;
-					if (!this.shouldReconnect ())
-						break;
 					if (this.reconnect ())
 						break;
 				}

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

@@ -4,13 +4,11 @@
 
 import java.util.UUID;
 
-import ch.qos.logback.amqp.tools.DefaultMutator;
-
-import org.slf4j.MDC;
-
 import ch.qos.logback.amqp.AmqpAppender;
+import ch.qos.logback.amqp.tools.DefaultMutator;
 import ch.qos.logback.classic.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
 
 import org.junit.Assert;
 import org.junit.Test;

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

@@ -24,6 +24,7 @@ public AmqpConsumer (
 		super (host, port, virtualHost, username, password, callbacks);
 		this.exchanges = exchanges;
 		this.queue = queue;
+		this.queue1 = null;
 		this.sink = sink;
 	}
 
@@ -34,9 +35,7 @@ protected final void loop ()
 				synchronized (this) {
 					if (this.shouldStopLoop ())
 						break loop;
-					if (!this.shouldReconnect ())
-						break;
-					if (this.connect ())
+					if (this.reconnect ())
 						break;
 				}
 				this.sleep ();
@@ -111,27 +110,38 @@ private final boolean declare ()
 			}
 		}
 		{
-			this.callbacks.handleLogEvent (Level.INFO, null, "amqp consumer declaring the queue `%s`", this.queue);
+			final String queue;
+			final boolean unique;
+			if ((this.queue == null) || this.queue.isEmpty ()) {
+				if (this.queue1 != null)
+					queue = this.queue1;
+				else
+					queue = "";
+				unique = true;
+			} else {
+				queue = this.queue;
+				unique = false;
+			}
+			this.callbacks.handleLogEvent (Level.INFO, null, "amqp consumer declaring the queue `%s`", queue);
 			try {
-				channel.queueDeclare (this.queue, true, false, false, null);
+				this.queue1 = channel.queueDeclare (queue, true, unique, unique, null).getQueue ();
 			} catch (final Throwable exception) {
-				this.callbacks
-						.handleException (
-								exception, "amqp consumer encountered an error while declaring the queue `%s`; aborting!",
-								this.queue);
+				this.queue1 = null;
+				this.callbacks.handleException (
+						exception, "amqp consumer encountered an error while declaring the queue `%s`; aborting!", queue);
 				return (false);
 			}
 		}
 		for (final String exchange : this.exchanges) {
 			this.callbacks.handleLogEvent (
-					Level.INFO, null, "amqp consumer binding the queue `%s` to exchange `%s`", this.queue, exchange);
+					Level.INFO, null, "amqp consumer binding the queue `%s` to exchange `%s`", this.queue1, exchange);
 			try {
-				channel.queueBind (this.queue, exchange, "#", null);
+				channel.queueBind (this.queue1, exchange, "#", null);
 			} catch (final Throwable exception) {
 				this.callbacks.handleException (
 						exception,
 						"amqp consumer encountered an error while binding the queue `%s` to exchange `%s`; aborting!",
-						this.queue, exchange);
+						this.queue1, exchange);
 				return (false);
 			}
 		}
@@ -143,7 +153,7 @@ private final boolean register ()
 		this.callbacks.handleLogEvent (Level.INFO, null, "amqp consumer registering the consumer");
 		final Channel channel = this.getChannel ();
 		try {
-			channel.basicConsume (this.queue, true, this.queue, true, true, null, new ConsumerCallback ());
+			channel.basicConsume (this.queue1, true, this.queue1, true, true, null, new ConsumerCallback ());
 			return (true);
 		} catch (final Throwable exception) {
 			this.callbacks.handleException (
@@ -154,6 +164,7 @@ private final boolean register ()
 
 	private final String[] exchanges;
 	private final String queue;
+	private String queue1;
 	private final LinkedBlockingQueue<AmqpMessage> sink;
 
 	private final class ConsumerCallback

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

@@ -234,6 +234,6 @@ private final void loop ()
 	private String virtualHost;
 
 	public static final String defaultExchange = "logback";
-	public static final String defaultQueue = "logback.agent";
+	public static final String defaultQueue = "";
 	public static final int waitTimeout = 1000;
 }

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

@@ -11,7 +11,7 @@
 		<username>guest</username>
 		<password>guest</password>
 		<exchange>logback</exchange>
-		<queue>logback.agent</queue>
+		<queue></queue>
 	</amqpConsumerAgent>
 
 	<appender name="console" class="ch.qos.logback.core.ConsoleAppender">