Propagate exception handler to connections

Author: michaelklishin

src/com/rabbitmq/client/Connection.java

@@ -233,4 +233,6 @@ public interface Connection extends ShutdownNotifier { // rename to AMQPConnecti
      * Remove all {@link BlockedListener}s.
      */
     void clearBlockedListeners();
+
+    ExceptionHandler getExceptionHandler();
 }

src/com/rabbitmq/client/ConnectionFactory.java

@@ -91,7 +91,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 ExceptionHandler exceptionHandler     = null;
 
     private boolean automaticRecovery             = false;
     private boolean topologyRecovery              = true;

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

@@ -51,9 +51,8 @@ public ConnectionParams(String username, String password, ExecutorService execut
         this.saslConfig = saslConfig;
         this.networkRecoveryInterval = networkRecoveryInterval;
         this.topologyRecovery = topologyRecovery;
-        
 
-        this.exceptionHandler = new DefaultExceptionHandler();
+        this.exceptionHandler = exceptionHandler;
     }
 
     public String getUsername() {
@@ -101,7 +100,7 @@ public void setExceptionHandler(ExceptionHandler exceptionHandler) {
     }
 
     public ConnectionParams exceptionHandler(ExceptionHandler exceptionHandler) {
-        this.exceptionHandler = exceptionHandler;
+        setExceptionHandler(exceptionHandler);
         return this;
     }
 

test/src/com/rabbitmq/client/test/functional/ExceptionHandling.java

@@ -0,0 +1,52 @@
+package com.rabbitmq.client.test.functional;
+
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.Envelope;
+import com.rabbitmq.client.ExceptionHandler;
+import com.rabbitmq.client.impl.DefaultExceptionHandler;
+import junit.framework.TestCase;
+
+import java.io.IOException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+public class ExceptionHandling extends TestCase {
+    private ConnectionFactory newConnectionFactory(ExceptionHandler eh) {
+        ConnectionFactory cf = new ConnectionFactory();
+        cf.setExceptionHandler(eh);
+        return cf;
+    }
+
+    public void testHandleConsumerException() throws IOException, InterruptedException {
+        final CountDownLatch latch = new CountDownLatch(1);
+        final DefaultExceptionHandler eh = new DefaultExceptionHandler() {
+            @Override
+            public void handleConsumerException(Channel channel, Throwable exception, Consumer consumer, String consumerTag, String methodName) {
+                latch.countDown();
+            }
+        };
+        ConnectionFactory cf = newConnectionFactory(eh);
+        assertEquals(cf.getExceptionHandler(), eh);
+        Connection conn = cf.newConnection();
+        assertEquals(conn.getExceptionHandler(), eh);
+        Channel ch = conn.createChannel();
+        String q = ch.queueDeclare().getQueue();
+        ch.basicConsume(q, new DefaultConsumer(ch) {
+            @Override
+            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
+                throw new RuntimeException("oops");
+            }
+        });
+        ch.basicPublish("", q, null, "".getBytes());
+        wait(latch);
+    }
+
+    private void wait(CountDownLatch latch) throws InterruptedException {
+        latch.await(30, TimeUnit.MINUTES);
+    }
+}