Added diagnostic messages. Fixed some bugs.

Author: cipriancraciun

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

@@ -2,7 +2,8 @@
 package ch.qos.logback.amqp;
 
 
-import ch.qos.logback.amqp.tools.ExceptionHandler;
+import ch.qos.logback.amqp.tools.Callbacks;
+import ch.qos.logback.classic.Level;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
@@ -14,15 +15,15 @@ public abstract class AmqpAccessor
 {
 	protected AmqpAccessor (
 			final String host, final Integer port, final String virtualHost, final String username, final String password,
-			final ExceptionHandler exceptionHandler)
+			final Callbacks callbacks)
 	{
 		super ();
 		this.host = ((host != null) && !host.isEmpty ()) ? host : "127.0.0.1";
 		this.port = ((port != null) && (port != 0)) ? port : 5672;
 		this.virtualHost = ((virtualHost != null) && !virtualHost.isEmpty ()) ? virtualHost : "/";
 		this.username = ((username != null) && !username.isEmpty ()) ? username : "guest";
 		this.password = ((password != null) && !password.isEmpty ()) ? password : "guest";
-		this.exceptionHandler = exceptionHandler;
+		this.callbacks = callbacks;
 		this.thread = null;
 		this.shutdownHook = null;
 		this.shouldStopLoop = true;
@@ -38,7 +39,7 @@ public final boolean isConnected ()
 		}
 	}
 
-	public final boolean isStarted ()
+	public final boolean isRunning ()
 	{
 		synchronized (this) {
 			return ((this.thread != null) && (this.thread.isAlive ()));
@@ -50,12 +51,15 @@ public final void start ()
 		synchronized (this) {
 			if (this.thread != null)
 				throw (new IllegalStateException ("amqp accessor is already started"));
+			this.callbacks.handleLogEvent (Level.INFO, null, "amqp accessor starting");
 			this.thread = new Thread (new Runnable () {
 				public final void run ()
 				{
+					AmqpAccessor.this.callbacks.handleLogEvent (Level.INFO, null, "amqp accessor started");
 					AmqpAccessor.this.loop ();
 					if (AmqpAccessor.this.shutdownHook != null)
 						Runtime.getRuntime ().removeShutdownHook (AmqpAccessor.this.shutdownHook);
+					AmqpAccessor.this.callbacks.handleLogEvent (Level.INFO, null, "amqp accessor stopped");
 				}
 			});
 			this.thread.setName (String.format ("%s@%x", this.getClass ().getName (), System.identityHashCode (this)));
@@ -64,9 +68,10 @@ public final void run ()
 				public final void run ()
 				{
 					AmqpAccessor.this.shutdownHook = null;
-					if (AmqpAccessor.this.isStarted ()) {
-						AmqpAccessor.this.stop ();
-						while (AmqpAccessor.this.isStarted ())
+					if (AmqpAccessor.this.isRunning ()) {
+						if (!AmqpAccessor.this.shouldStopLoop)
+							AmqpAccessor.this.stop ();
+						while (AmqpAccessor.this.isRunning ())
 							try {
 								Thread.sleep (AmqpAccessor.waitTimeout);
 							} catch (final InterruptedException exception) {
@@ -86,6 +91,7 @@ public final void stop ()
 		synchronized (this) {
 			if (this.thread == null)
 				throw (new IllegalStateException ("amqp accessor is not started"));
+			this.callbacks.handleLogEvent (Level.INFO, null, "amqp accessor stopping");
 			this.shouldStopLoop = true;
 		}
 	}
@@ -95,6 +101,9 @@ protected final boolean connect ()
 		synchronized (this) {
 			if (this.connection != null)
 				throw (new IllegalStateException ("amqp accessor is already connected"));
+			this.callbacks.handleLogEvent (
+					Level.INFO, null, "amqp accessor connecting to `%s@%s:%s:%s`", this.username, this.host, this.port,
+					this.virtualHost);
 			this.shouldReconnect = true;
 			final ConnectionFactory connectionFactory = new ConnectionFactory ();
 			connectionFactory.setHost (this.host);
@@ -106,16 +115,15 @@ protected final boolean connect ()
 				this.connection = connectionFactory.newConnection ();
 			} catch (final Throwable exception) {
 				this.connection = null;
-				this.exceptionHandler.handleException (
-						"amqp accessor encountered an error while connecting; aborting!", exception);
+				this.callbacks.handleException (exception, "amqp accessor encountered an error while connecting; aborting!");
 			}
 			if (this.connection != null)
 				try {
 					this.channel = this.connection.createChannel ();
 				} catch (final Throwable exception) {
 					this.channel = null;
-					this.exceptionHandler.handleException (
-							"amqp accessor encountered an error while opening a channel; aborting!", exception);
+					this.callbacks.handleException (
+							exception, "amqp accessor encountered an error while opening the channel; aborting!");
 				}
 			if ((this.connection == null) || (this.channel == null)) {
 				if (this.connection != null)
@@ -126,12 +134,15 @@ protected final boolean connect ()
 					public void shutdownCompleted (final ShutdownSignalException exception)
 					{
 						AmqpAccessor.this.shouldReconnect = true;
-						if (!exception.isInitiatedByApplication ())
-							AmqpAccessor.this.exceptionHandler.handleException (
-									"amqp consumer encountered an shutdown error; ignoring!", exception);
+						if (!exception.isInitiatedByApplication ()) {
+							AmqpAccessor.this.callbacks.handleException (
+									exception, "amqp consumer encountered an shutdown error; ignoring!");
+							AmqpAccessor.this.disconnect ();
+						}
 					}
 				});
 				this.shouldReconnect = false;
+				this.callbacks.handleLogEvent (Level.INFO, null, "amqp accessor connected");
 			}
 			return (this.connection != null);
 		}
@@ -141,6 +152,7 @@ protected final void disconnect ()
 	{
 		if (this.connection == null)
 			throw (new IllegalStateException ("amqp accessor is not connected"));
+		this.callbacks.handleLogEvent (Level.INFO, null, "amqp accessor disconnecting");
 		this.shouldReconnect = true;
 		try {
 			try {
@@ -150,8 +162,7 @@ protected final void disconnect ()
 				this.connection.close ();
 			}
 		} catch (final Throwable exception) {
-			this.exceptionHandler.handleException (
-					"amqp accessor encountered an error while disconnecting; ignoring!", exception);
+			this.callbacks.handleException (exception, "amqp accessor encountered an error while disconnecting; ignoring!");
 		} finally {
 			this.connection = null;
 			this.channel = null;
@@ -191,7 +202,7 @@ protected final void sleep ()
 		} catch (final InterruptedException exception) {}
 	}
 
-	protected final ExceptionHandler exceptionHandler;
+	protected final Callbacks callbacks;
 	protected final String host;
 	protected final String password;
 	protected final int port;

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

@@ -6,7 +6,7 @@
 import java.util.concurrent.LinkedBlockingDeque;
 
 import ch.qos.logback.amqp.tools.DefaultBinarySerializer;
-import ch.qos.logback.amqp.tools.ExceptionHandler;
+import ch.qos.logback.amqp.tools.DefaultContextAwareCallbacks;
 import ch.qos.logback.amqp.tools.Serializer;
 import ch.qos.logback.classic.PatternLayout;
 import ch.qos.logback.classic.net.LoggingEventPreSerializationTransformer;
@@ -18,8 +18,6 @@
 
 public final class AmqpAppender
 		extends UnsynchronizedAppenderBase<ILoggingEvent>
-		implements
-			ExceptionHandler
 {
 	public AmqpAppender ()
 	{
@@ -34,11 +32,6 @@ public AmqpAppender ()
 		this.publisher = null;
 	}
 
-	public final void handleException (final String message, final Throwable exception)
-	{
-		this.addError (message, exception);
-	}
-	
 	public final boolean isDrained ()
 	{
 		return (this.buffer.isEmpty ());
@@ -47,7 +40,7 @@ public final boolean isDrained ()
 	public final boolean isRunning ()
 	{
 		final AmqpPublisher publisher = this.publisher;
-		return (((publisher != null) && publisher.isStarted ()) || super.isStarted ());
+		return (((publisher != null) && publisher.isRunning ()) || super.isStarted ());
 	}
 
 	public final void setContext (final Context context)
@@ -59,6 +52,13 @@ public final void setContext (final Context context)
 		this.routingKeyLayout.setContext (context);
 	}
 
+	public final void setExchangePattern (final String pattern)
+	{
+		if (this.isStarted ())
+			throw (new IllegalStateException ("amqp appender is already started"));
+		this.exchangeLayout.setPattern (pattern);
+	}
+	
 	public final void setHost (final String host)
 	{
 		if (this.isStarted ())
@@ -87,13 +87,6 @@ public final void setRoutingKeyPattern (final String pattern)
 		this.routingKeyLayout.setPattern (pattern);
 	}
 
-	public final void setExchangePattern (final String pattern)
-	{
-		if (this.isStarted ())
-			throw (new IllegalStateException ("amqp appender is already started"));
-		this.exchangeLayout.setPattern (pattern);
-	}
-	
 	public final void setUsername (final String username)
 	{
 		if (this.isStarted ())
@@ -117,7 +110,9 @@ public final void start ()
 		this.exchangeLayout.start ();
 		this.routingKeyLayout.start ();
 		this.publisher =
-				new AmqpPublisher (this.host, this.port, this.virtualHost, this.username, this.password, this, this.buffer);
+				new AmqpPublisher (
+						this.host, this.port, this.virtualHost, this.username, this.password,
+						new DefaultContextAwareCallbacks (this), this.buffer);
 		this.publisher.start ();
 		super.start ();
 	}

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

@@ -5,7 +5,8 @@
 import java.util.concurrent.LinkedBlockingDeque;
 import java.util.concurrent.TimeUnit;
 
-import ch.qos.logback.amqp.tools.ExceptionHandler;
+import ch.qos.logback.amqp.tools.Callbacks;
+import ch.qos.logback.classic.Level;
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.Channel;
 
@@ -15,9 +16,9 @@ public final class AmqpPublisher
 {
 	public AmqpPublisher (
 			final String host, final Integer port, final String virtualHost, final String username, final String password,
-			final ExceptionHandler exceptionHandler, final LinkedBlockingDeque<AmqpMessage> source)
+			final Callbacks callbacks, final LinkedBlockingDeque<AmqpMessage> source)
 	{
-		super (host, port, virtualHost, username, password, exceptionHandler);
+		super (host, port, virtualHost, username, password, callbacks);
 		this.source = source;
 	}
 
@@ -35,6 +36,7 @@ protected final void loop ()
 				}
 				this.sleep ();
 			}
+			this.callbacks.handleLogEvent (Level.INFO, null, "amqp publisher showeling outbound messages");
 			while (true) {
 				synchronized (this) {
 					if (this.shouldStopLoop ())
@@ -71,8 +73,8 @@ private final boolean publish (final AmqpMessage message)
 			channel.basicPublish (message.exchange, message.routingKey, false, false, properties, message.content);
 			return (true);
 		} catch (final Throwable exception) {
-			this.exceptionHandler.handleException (
-					"amqp publisher encountered an error while publishing the message; requeueing!", exception);
+			this.callbacks.handleException (
+					exception, "amqp publisher encountered an error while publishing the message; requeueing!");
 			return (false);
 		}
 	}

logback-amqp-appender/src/main/java/ch/qos/logback/amqp/tools/Callbacks.java

@@ -0,0 +1,15 @@
+
+package ch.qos.logback.amqp.tools;
+
+
+import ch.qos.logback.classic.Level;
+
+
+public interface Callbacks
+{
+	public abstract void handleException (
+			final Throwable exception, final String messageFormat, final Object ... messageArguments);
+	
+	public abstract void handleLogEvent (
+			final Level level, final Throwable exception, final String messageFormat, final Object ... messageArguments);
+}

logback-amqp-appender/src/main/java/ch/qos/logback/amqp/tools/DefaultBinarySerializer.java

@@ -22,27 +22,27 @@ public final Serializable deserialize (final byte[] data)
 		return (object);
 	}
 
-	public final byte[] serialize (final Serializable object)
-			throws Throwable
+	public final String getContentEncoding ()
 	{
-		final ByteArrayOutputStream stream = new ByteArrayOutputStream (this.defaultBufferSize);
-		final ObjectOutputStream encoder = new ObjectOutputStream (stream);
-		encoder.writeObject (object);
-		encoder.close ();
-		return (stream.toByteArray ());
+		return (this.contentEncoding);
 	}
 
 	public final String getContentType ()
 	{
 		return (this.contentType);
 	}
 
-	public final String getContentEncoding ()
+	public final byte[] serialize (final Serializable object)
+			throws Throwable
 	{
-		return (this.contentEncoding);
+		final ByteArrayOutputStream stream = new ByteArrayOutputStream (this.defaultBufferSize);
+		final ObjectOutputStream encoder = new ObjectOutputStream (stream);
+		encoder.writeObject (object);
+		encoder.close ();
+		return (stream.toByteArray ());
 	}
 
-	public final int defaultBufferSize = 2048;
-	public final String contentType = "application/x-java-serialized-object";
 	public final String contentEncoding = "binary";
+	public final String contentType = "application/x-java-serialized-object";
+	public final int defaultBufferSize = 2048;
 }

logback-amqp-appender/src/main/java/ch/qos/logback/amqp/tools/DefaultContextAwareCallbacks.java

@@ -0,0 +1,58 @@
+
+package ch.qos.logback.amqp.tools;
+
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.core.spi.ContextAware;
+
+
+public final class DefaultContextAwareCallbacks
+		implements
+			Callbacks
+{
+	public DefaultContextAwareCallbacks (final ContextAware delegate)
+	{
+		super ();
+		this.delegate = delegate;
+	}
+	
+	public final void handleException (
+			final Throwable exception, final String messageFormat, final Object ... messageArguments)
+	{
+		this.handleLogEvent (Level.ERROR, exception, messageFormat, messageArguments);
+	}
+	
+	public final void handleLogEvent (
+			final Level level, final Throwable exception, final String messageFormat, final Object ... messageArguments)
+	{
+		final String message = String.format (messageFormat, messageArguments);
+		switch (level.levelInt) {
+			case Level.ERROR_INT :
+				if (exception != null)
+					this.delegate.addError (message, exception);
+				else
+					this.delegate.addError (message);
+				break;
+			case Level.WARN_INT :
+				if (exception != null)
+					this.delegate.addWarn (message, exception);
+				else
+					this.delegate.addWarn (message);
+				break;
+			case Level.INFO_INT :
+				if (exception != null)
+					this.delegate.addInfo (message, exception);
+				else
+					this.delegate.addInfo (message);
+				break;
+			default:
+				if (exception != null)
+					this.delegate.addInfo (message, exception);
+				else
+					this.delegate.addInfo (message);
+				break;
+		}
+	}
+	
+	private final ContextAware delegate;
+}