Don't make ConnectionParams mutable WRT exception handler, it's not mutable for anything else. That means we can skip the final safety check in the AMQConnection ctor.

Author: Simon MacMullen

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

@@ -205,14 +205,15 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
         this.password = params.getPassword();
         this._frameHandler = frameHandler;
         this._virtualHost = params.getVirtualHost();
+        this._exceptionHandler = params.getExceptionHandler();
+
         this._clientProperties = new HashMap<String, Object>(params.getClientProperties());
         this.requestedFrameMax = params.getRequestedFrameMax();
         this.requestedChannelMax = params.getRequestedChannelMax();
         this.requestedHeartbeat = params.getRequestedHeartbeat();
         this.saslConfig = params.getSaslConfig();
 
         this._workService  = new ConsumerWorkService(params.getExecutor());
-        this._exceptionHandler = initExceptionHandler(params);
         this._channelManager = null;
 
         this._heartbeatSender = new HeartbeatSender(frameHandler);
@@ -221,17 +222,6 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler)
         this._inConnectionNegotiation = true; // we start out waiting for the first protocol response
     }
 
-  private ExceptionHandler initExceptionHandler(ConnectionParams params) {
-    ExceptionHandler result = params.getExceptionHandler();
-    // null exception handler causes I/O thread to fail,
-    // so be extra defensive here
-    if(result == null) {
-      return new DefaultExceptionHandler();
-    } else {
-      return result;
-    }
-  }
-
   /**
      * Start up the connection, including the MainLoop thread.
      * Sends the protocol

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

@@ -51,7 +51,6 @@ public ConnectionParams(String username, String password, ExecutorService execut
         this.saslConfig = saslConfig;
         this.networkRecoveryInterval = networkRecoveryInterval;
         this.topologyRecovery = topologyRecovery;
-
         this.exceptionHandler = exceptionHandler;
     }
 
@@ -95,15 +94,6 @@ public ExceptionHandler getExceptionHandler() {
         return exceptionHandler;
     }
 
-    public void setExceptionHandler(ExceptionHandler exceptionHandler) {
-        this.exceptionHandler = exceptionHandler;
-    }
-
-    public ConnectionParams exceptionHandler(ExceptionHandler exceptionHandler) {
-        setExceptionHandler(exceptionHandler);
-        return this;
-    }
-
     public int getNetworkRecoveryInterval() {
         return networkRecoveryInterval;
     }

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

@@ -59,6 +59,7 @@ public static TestSuite suite() {
     /** The mock frame handler used to test connection behaviour. */
     private MockFrameHandler _mockFrameHandler;
     private ConnectionFactory factory;
+    private MyExceptionHandler exceptionHandler;
 
     /** Setup the environment for this test
      * @see junit.framework.TestCase#setUp()
@@ -68,6 +69,8 @@ public static TestSuite suite() {
         super.setUp();
         _mockFrameHandler = new MockFrameHandler();
         factory = new ConnectionFactory();
+        exceptionHandler = new MyExceptionHandler();
+        factory.setExceptionHandler(exceptionHandler);
     }
 
     /** Tear down the environment for this test
@@ -86,19 +89,17 @@ public static TestSuite suite() {
     public void testConnectionSendsSingleHeaderAndTimesOut() {
         IOException exception = new SocketTimeoutException();
         _mockFrameHandler.setExceptionOnReadingFrames(exception);
-        MyExceptionHandler handler = new MyExceptionHandler();
         assertEquals(0, _mockFrameHandler.countHeadersSent());
         try {
             ConnectionParams params = factory.params(Executors.newFixedThreadPool(1));
-            params.setExceptionHandler(handler);
             new AMQConnection(params, _mockFrameHandler).start();
             fail("Connection should have thrown exception");
         } catch(IOException signal) {
            // As expected
         }
         assertEquals(1, _mockFrameHandler.countHeadersSent());
         // _connection.close(0, CLOSE_MESSAGE);
-        List<Throwable> exceptionList = handler.getHandledExceptions();
+        List<Throwable> exceptionList = exceptionHandler.getHandledExceptions();
         assertEquals(Collections.<Throwable>singletonList(exception), exceptionList);
     }
 
@@ -124,7 +125,6 @@ public void testConnectionHangInNegotiation() {
         assertEquals(0, this._mockFrameHandler.countHeadersSent());
         try {
             ConnectionParams params = factory.params(Executors.newFixedThreadPool(1));
-            params.setExceptionHandler(handler);
             new AMQConnection(params, this._mockFrameHandler).start();
             fail("Connection should have thrown exception");
         } catch(IOException signal) {

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

@@ -36,10 +36,38 @@
 import com.rabbitmq.client.impl.SocketFrameHandler;
 
 public class CloseInMainLoop extends BrokerTestCase{
+    private final CountDownLatch closeLatch = new CountDownLatch(1);
+
+    private ConnectionFactory specialConnectionFactory() {
+        ConnectionFactory f = new ConnectionFactory();
+        f.setExceptionHandler(new DefaultExceptionHandler(){
+            @Override
+            public void handleConsumerException(Channel channel,
+                                                Throwable exception,
+                                                Consumer consumer,
+                                                String consumerTag,
+                                                String methodName) {
+                try {
+                    // TODO: change this to call 4-parameter close and make 6-parm one private
+                    ((AMQConnection) channel.getConnection())
+                            .close(AMQP.INTERNAL_ERROR,
+                                    "Internal error in Consumer " + consumerTag,
+                                    false,
+                                    exception,
+                                    -1,
+                                    false);
+                } catch (Throwable e) {
+                    // Man, this clearly isn't our day.
+                    // TODO: Log the nested failure
+                } finally {
+                    closeLatch.countDown();
+                }
+            }
+        });
+        return f;
+    }
 
-  private final CountDownLatch closeLatch = new CountDownLatch(1);
-
-  class SpecialConnection extends AMQConnection{
+    class SpecialConnection extends AMQConnection{
     private AtomicBoolean validShutdown = new AtomicBoolean(false);
 
     public boolean hadValidShutdown(){
@@ -48,45 +76,16 @@ public boolean hadValidShutdown(){
     }
 
     public SpecialConnection() throws Exception {
-        this(new ConnectionFactory());
-    }
-
-    private SpecialConnection(ConnectionFactory factory) throws Exception{
-        super(factory.params(Executors.newFixedThreadPool(1)).exceptionHandler(
-                             new DefaultExceptionHandler(){
-                                 @Override
-                                 public void handleConsumerException(Channel channel,
-                                                                     Throwable exception,
-                                                                     Consumer consumer,
-                                                                     String consumerTag,
-                                                                     String methodName) {
-                                     try {
-                                         // TODO: change this to call 4-parameter close and make 6-parm one private
-                                       ((AMQConnection) channel.getConnection())
-                                           .close(AMQP.INTERNAL_ERROR,
-                                                   "Internal error in Consumer " + consumerTag,
-                                                   false,
-                                                   exception,
-                                                   -1,
-                                                   false);
-                                      } catch (Throwable e) {
-                                         // Man, this clearly isn't our day.
-                                         // TODO: Log the nested failure
-                                     } finally {
-                                         closeLatch.countDown();
-                                     }
-                                 }
-                             }
-        ), new SocketFrameHandler(SocketFactory.getDefault().createSocket("localhost", AMQP.PROTOCOL.PORT)));
+        super(specialConnectionFactory().params(Executors.newFixedThreadPool(1)),
+              new SocketFrameHandler(SocketFactory.getDefault().createSocket("localhost", AMQP.PROTOCOL.PORT)));
         this.start();
-      }
+    }
 
     @Override
     public boolean processControlCommand(Command c) throws IOException{
       if(c.getMethod() instanceof AMQP.Connection.CloseOk) validShutdown.set(true);
       return super.processControlCommand(c);
     }
-
   }
 
   public void testCloseOKNormallyReceived() throws Exception{