Initialize exception handler in ConnectionFactory

Author: michaelklishin

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     = null;
+    private ExceptionHandler exceptionHandler     = new DefaultExceptionHandler();
 
     private boolean automaticRecovery             = false;
     private boolean topologyRecovery              = true;
@@ -444,6 +444,9 @@ public ExceptionHandler getExceptionHandler() {
      */
 
     public void setExceptionHandler(ExceptionHandler exceptionHandler) {
+        if (exceptionHandler == null) {
+          throw new IllegalArgumentException("exception handler cannot be null!");
+        }
         this.exceptionHandler = exceptionHandler;
     }
 

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

@@ -111,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 ExceptionHandler _exceptionHandler = new DefaultExceptionHandler();
+    private ExceptionHandler _exceptionHandler;
 
     /** Object used for blocking main application thread when doing all the necessary
      * connection shutdown operations
@@ -207,15 +207,14 @@ 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 = params.getExceptionHandler();
+        this._exceptionHandler = initExceptionHandler(params);
         this._channelManager = null;
 
         this._heartbeatSender = new HeartbeatSender(frameHandler);
@@ -224,7 +223,18 @@ 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
      * version negotiation header, and runs through

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

@@ -46,6 +46,16 @@ public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProp
         wait(latch);
     }
 
+    public void testNullExceptionHandler() {
+      ConnectionFactory cf = new ConnectionFactory();
+      try {
+        cf.setExceptionHandler(null);
+        fail("expected setExceptionHandler to throw");
+      } catch (IllegalArgumentException iae) {
+        // expected
+      }
+    }
+
     private void wait(CountDownLatch latch) throws InterruptedException {
         latch.await(30, TimeUnit.MINUTES);
     }