merge default into bug26008

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

@@ -33,6 +33,7 @@
 
 import com.rabbitmq.client.impl.AMQConnection;
 import com.rabbitmq.client.impl.ConnectionParams;
+import com.rabbitmq.client.impl.DefaultExceptionHandler;
 import com.rabbitmq.client.impl.FrameHandler;
 import com.rabbitmq.client.impl.FrameHandlerFactory;
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
@@ -90,6 +91,7 @@ public class ConnectionFactory implements Cloneable {
     private SaslConfig saslConfig                 = DefaultSaslConfig.PLAIN;
     private ExecutorService sharedExecutor;
     private SocketConfigurator socketConf         = new DefaultSocketConfigurator();
+    private ExceptionHandler exceptionHandler     = null;
 
     private boolean automaticRecovery             = false;
     private boolean topologyRecovery              = true;
@@ -426,6 +428,25 @@ public void setSharedExecutor(ExecutorService executor) {
         this.sharedExecutor = executor;
     }
 
+    /**
+     * Get the exception handler.
+     *
+     * @see ExceptionHandler
+     */
+    public ExceptionHandler getExceptionHandler() {
+        return exceptionHandler;
+    }
+
+    /**
+     * Set the exception handler to use by default for newly created connections.
+     *
+     * @see ExceptionHandler
+     */
+
+    public void setExceptionHandler(ExceptionHandler exceptionHandler) {
+        this.exceptionHandler = exceptionHandler;
+    }
+
     public boolean isSSL(){
         return getSocketFactory() instanceof SSLSocketFactory;
     }
@@ -559,7 +580,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/ExceptionHandler.java renamed to src/com/rabbitmq/client/ExceptionHandler.java

@@ -14,12 +14,7 @@
 //  Copyright (c) 2007-2013 GoPivotal, Inc.  All rights reserved.
 //
 
-package com.rabbitmq.client.impl;
-
-import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.Consumer;
-import com.rabbitmq.client.TopologyRecoveryException;
+package com.rabbitmq.client;
 
 /**
  * Interface to an exception-handling object.

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

@@ -31,6 +31,7 @@
 import com.rabbitmq.client.AMQP;
 import com.rabbitmq.client.AuthenticationFailureException;
 import com.rabbitmq.client.BlockedListener;
+import com.rabbitmq.client.ExceptionHandler;
 import com.rabbitmq.client.Method;
 import com.rabbitmq.client.AlreadyClosedException;
 import com.rabbitmq.client.Channel;
@@ -110,7 +111,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 = new DefaultExceptionHandler();
 
     /** Object used for blocking main application thread when doing all the necessary
      * connection shutdown operations
@@ -214,6 +215,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

@@ -1,5 +1,6 @@
 package com.rabbitmq.client.impl;
 
+import com.rabbitmq.client.ExceptionHandler;
 import com.rabbitmq.client.SaslConfig;
 
 import java.util.Map;
@@ -32,12 +33,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;
@@ -49,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() {
@@ -99,7 +100,7 @@ public void setExceptionHandler(ExceptionHandler exceptionHandler) {
     }
 
     public ConnectionParams exceptionHandler(ExceptionHandler exceptionHandler) {
-        this.exceptionHandler = exceptionHandler;
+        setExceptionHandler(exceptionHandler);
         return this;
     }
 

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

@@ -24,10 +24,11 @@
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.Consumer;
+import com.rabbitmq.client.ExceptionHandler;
 import com.rabbitmq.client.TopologyRecoveryException;
 
 /**
- * Default implementation of {@link ExceptionHandler} used by {@link AMQConnection}.
+ * Default implementation of {@link com.rabbitmq.client.ExceptionHandler} used by {@link AMQConnection}.
  */
 public class DefaultExceptionHandler implements ExceptionHandler {
     public void handleUnexpectedConnectionDriverException(Connection conn, Throwable exception) {

src/com/rabbitmq/client/impl/recovery/AutorecoveringConnection.java

@@ -11,7 +11,7 @@
 import com.rabbitmq.client.ShutdownSignalException;
 import com.rabbitmq.client.TopologyRecoveryException;
 import com.rabbitmq.client.impl.ConnectionParams;
-import com.rabbitmq.client.impl.ExceptionHandler;
+import com.rabbitmq.client.ExceptionHandler;
 import com.rabbitmq.client.impl.FrameHandlerFactory;
 import com.rabbitmq.client.impl.NetworkConnection;
 

test/src/com/rabbitmq/client/test/AMQConnectionTest.java

@@ -35,7 +35,7 @@
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.Consumer;
 import com.rabbitmq.client.impl.AMQConnection;
-import com.rabbitmq.client.impl.ExceptionHandler;
+import com.rabbitmq.client.ExceptionHandler;
 import com.rabbitmq.client.impl.Frame;
 import com.rabbitmq.client.impl.FrameHandler;
 

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);
+    }
+}

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

@@ -76,5 +76,6 @@ public static void add(TestSuite suite) {
         suite.addTestSuite(ConsumerPriorities.class);
         suite.addTestSuite(Policies.class);
         suite.addTestSuite(ConnectionRecovery.class);
+        suite.addTestSuite(ExceptionHandling.class);
     }
 }