Make exception handler configurable via ConnectionFactory

Author: michaelklishin

src/com/rabbitmq/client/ConnectionFactory.java

@@ -33,6 +33,8 @@
 
 import com.rabbitmq.client.impl.AMQConnection;
 import com.rabbitmq.client.impl.ConnectionParams;
+import com.rabbitmq.client.impl.DefaultExceptionHandler;
+import com.rabbitmq.client.impl.ExceptionHandler;
 import com.rabbitmq.client.impl.FrameHandler;
 import com.rabbitmq.client.impl.FrameHandlerFactory;
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
@@ -90,6 +92,7 @@ public class ConnectionFactory implements Cloneable {
     private SaslConfig saslConfig                 = DefaultSaslConfig.PLAIN;
     private ExecutorService sharedExecutor;
     private SocketConfigurator socketConf         = new DefaultSocketConfigurator();
+    private ExceptionHandler exceptionHandler     = new DefaultExceptionHandler();
 
     private boolean automaticRecovery             = false;
     private boolean topologyRecovery              = true;
@@ -426,6 +429,25 @@ public void setSharedExecutor(ExecutorService executor) {
         this.sharedExecutor = executor;
     }
 
+    /**
+     * Get the exception handler.
+     *
+     * @see com.rabbitmq.client.impl.ExceptionHandler
+     */
+    public ExceptionHandler getExceptionHandler() {
+        return exceptionHandler;
+    }
+
+    /**
+     * Set the exception handler to use by default for newly created connections.
+     *
+     * @see com.rabbitmq.client.impl.ExceptionHandler
+     */
+
+    public void setExceptionHandler(ExceptionHandler exceptionHandler) {
+        this.exceptionHandler = exceptionHandler;
+    }
+
     public boolean isSSL(){
         return getSocketFactory() instanceof SSLSocketFactory;
     }
@@ -557,7 +579,7 @@ public Connection newConnection(ExecutorService executor, Address[] addrs)
     public ConnectionParams params(ExecutorService executor) {
         return new ConnectionParams(username, password, executor, virtualHost, getClientProperties(),
                                     requestedFrameMax, requestedChannelMax, requestedHeartbeat, saslConfig,
-                                    networkRecoveryInterval, topologyRecovery);
+                                    networkRecoveryInterval, topologyRecovery, exceptionHandler);
     }
 
     /**

src/com/rabbitmq/client/impl/AMQConnection.java

@@ -110,7 +110,7 @@ public static final Map<String, Object> defaultClientProperties() {
     private volatile boolean _running = false;
 
     /** Handler for (uncaught) exceptions that crop up in the {@link MainLoop}. */
-    private final ExceptionHandler _exceptionHandler;
+    private ExceptionHandler _exceptionHandler;
 
     /** Object used for blocking main application thread when doing all the necessary
      * connection shutdown operations
@@ -214,6 +214,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
         this.saslConfig = params.getSaslConfig();
 
         this._workService  = new ConsumerWorkService(params.getExecutor());
+        this._exceptionHandler = params.getExceptionHandler();
         this._channelManager = null;
 
         this._heartbeatSender = new HeartbeatSender(frameHandler);

src/com/rabbitmq/client/impl/ConnectionParams.java

@@ -32,12 +32,13 @@ public class ConnectionParams {
      * @param saslConfig sasl configuration hook
      * @param networkRecoveryInterval interval used when recovering from network failure
      * @param topologyRecovery should topology (queues, exchanges, bindings, consumers) recovery be performed?
+     * @param exceptionHandler
      */
     public ConnectionParams(String username, String password, ExecutorService executor,
                             String virtualHost, Map<String, Object> clientProperties,
                             int requestedFrameMax, int requestedChannelMax, int requestedHeartbeat,
                             SaslConfig saslConfig, int networkRecoveryInterval,
-                            boolean topologyRecovery) {
+                            boolean topologyRecovery, ExceptionHandler exceptionHandler) {
         this.username = username;
         this.password = password;
         this.executor = executor;