merge heads

Author: michaelklishin

src/com/rabbitmq/client/AlreadyClosedException.java

@@ -25,8 +25,11 @@ public class AlreadyClosedException extends ShutdownSignalException {
     /** Default for suppressing warnings without version check. */
     private static final long serialVersionUID = 1L;
 
-    public AlreadyClosedException(String s, Object ref)
+    public AlreadyClosedException(ShutdownSignalException sse)
     {
-        super(true, true, s, ref);
+        super(sse.isHardError(),
+              sse.isInitiatedByApplication(),
+              sse.getReason(),
+              sse.getReference());
     }
 }

src/com/rabbitmq/client/ShutdownSignalException.java

@@ -41,7 +41,7 @@ public class ShutdownSignalException extends RuntimeException implements Sensibl
     private final boolean _initiatedByApplication;
 
     /** Possible explanation */
-    private final Object _reason;
+    private final Method _reason;
 
     /** Either Channel or Connection instance, depending on _hardError */
     private final Object _ref;
@@ -50,12 +50,12 @@ public class ShutdownSignalException extends RuntimeException implements Sensibl
      * Construct a ShutdownSignalException from the arguments.
      * @param hardError the relevant hard error
      * @param initiatedByApplication if the shutdown was client-initiated
-     * @param reason Object describing the origin of the exception
+     * @param reason AMQP method describing the exception reason
      * @param ref Reference to Connection or Channel that fired the signal
      */
     public ShutdownSignalException(boolean hardError,
                                    boolean initiatedByApplication,
-                                   Object reason, Object ref)
+                                   Method reason, Object ref)
     {
         super((initiatedByApplication
                ? ("clean " + (hardError ? "connection" : "channel") + " shutdown")
@@ -77,8 +77,8 @@ public ShutdownSignalException(boolean hardError,
      */
     public boolean isInitiatedByApplication() { return _initiatedByApplication; }
 
-    /** @return the reason object, if any */
-    public Object getReason() { return _reason; }
+    /** @return the reason, if any */
+    public Method getReason() { return _reason; }
 
     /** @return Reference to Connection or Channel object that fired the signal **/
     public Object getReference() { return _ref; }

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

@@ -187,7 +187,7 @@ public void ensureIsOpen()
         throws AlreadyClosedException
     {
         if (!isOpen()) {
-            throw new AlreadyClosedException("Attempt to use closed channel", this);
+            throw new AlreadyClosedException(getCloseReason());
         }
     }
 
@@ -262,7 +262,7 @@ public void processShutdownSignal(ShutdownSignalException signal,
             synchronized (_channelMutex) {
                 if (!setShutdownCauseIfOpen(signal)) {
                     if (!ignoreClosed)
-                        throw new AlreadyClosedException("Attempt to use closed channel", this);
+                        throw new AlreadyClosedException(getCloseReason());
                 }
 
                 _channelMutex.notifyAll();

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

@@ -164,7 +164,7 @@ private final void ensureIsOpen()
         throws AlreadyClosedException
     {
         if (!isOpen()) {
-            throw new AlreadyClosedException("Attempt to use closed connection", this);
+            throw new AlreadyClosedException(getCloseReason());
         }
     }
 
@@ -312,14 +312,11 @@ public void start()
                         response = sm.handleChallenge(challenge, this.username, this.password);
                     }
                 } catch (ShutdownSignalException e) {
-                    Object shutdownReason = e.getReason();
-                    if (shutdownReason instanceof AMQCommand) {
-                        Method shutdownMethod = ((AMQCommand) shutdownReason).getMethod();
-                        if (shutdownMethod instanceof AMQP.Connection.Close) {
-                            AMQP.Connection.Close shutdownClose =  (AMQP.Connection.Close) shutdownMethod;
-                            if (shutdownClose.getReplyCode() == AMQP.ACCESS_REFUSED) {
-                                throw new AuthenticationFailureException(shutdownClose.getReplyText());
-                            }
+                    Method shutdownMethod = e.getReason();
+                    if (shutdownMethod instanceof AMQP.Connection.Close) {
+                        AMQP.Connection.Close shutdownClose =  (AMQP.Connection.Close) shutdownMethod;
+                        if (shutdownClose.getReplyCode() == AMQP.ACCESS_REFUSED) {
+                            throw new AuthenticationFailureException(shutdownClose.getReplyText());
                         }
                     }
                     throw new PossibleAuthenticationFailureException(e);
@@ -534,11 +531,11 @@ private class MainLoop extends Thread {
                 }
             } catch (EOFException ex) {
                 if (!_brokerInitiatedShutdown)
-                    shutdown(ex, false, ex, true);
+                    shutdown(null, false, ex, true);
             } catch (Throwable ex) {
                 _exceptionHandler.handleUnexpectedConnectionDriverException(AMQConnection.this,
                                                                             ex);
-                shutdown(ex, false, ex, true);
+                shutdown(null, false, ex, true);
             } finally {
                 // Finally, shut down our underlying data connection.
                 _frameHandler.close();
@@ -631,7 +628,7 @@ public boolean processControlCommand(Command c) throws IOException
     }
 
     public void handleConnectionClose(Command closeCommand) {
-        ShutdownSignalException sse = shutdown(closeCommand, false, null, _inConnectionNegotiation);
+        ShutdownSignalException sse = shutdown(closeCommand.getMethod(), false, null, _inConnectionNegotiation);
         try {
             _channel0.quiescingTransmit(new AMQP.Connection.CloseOk.Builder().build());
         } catch (IOException _) { } // ignore
@@ -664,13 +661,13 @@ public SocketCloseWait(ShutdownSignalException sse) {
      * built from the argument, and stops this connection from accepting further work from the
      * application. {@link com.rabbitmq.client.ShutdownListener ShutdownListener}s for the
      * connection are notified when the main loop terminates.
-     * @param reason object being shutdown
+     * @param reason description of reason for the exception
      * @param initiatedByApplication true if caused by a client command
      * @param cause trigger exception which caused shutdown
      * @param notifyRpc true if outstanding rpc should be informed of shutdown
      * @return a shutdown signal built using the given arguments
      */
-    public ShutdownSignalException shutdown(Object reason,
+    public ShutdownSignalException shutdown(Method reason,
                          boolean initiatedByApplication,
                          Throwable cause,
                          boolean notifyRpc)
@@ -680,7 +677,7 @@ public ShutdownSignalException shutdown(Object reason,
         return sse;
     }
 
-    private ShutdownSignalException startShutdown(Object reason,
+    private ShutdownSignalException startShutdown(Method reason,
                          boolean initiatedByApplication,
                          Throwable cause,
                          boolean notifyRpc)
@@ -690,7 +687,7 @@ private ShutdownSignalException startShutdown(Object reason,
         sse.initCause(cause);
         if (!setShutdownCauseIfOpen(sse)) {
             if (initiatedByApplication)
-                throw new AlreadyClosedException("Attempt to use closed connection", this);
+                throw new AlreadyClosedException(getCloseReason());
         }
 
         // stop any heartbeating
@@ -813,8 +810,11 @@ public AMQCommand transformReply(AMQCommand command) {
               _channel0.quiescingTransmit(reason);
             }
         } catch (TimeoutException tte) {
-            if (!abort)
-                throw new ShutdownSignalException(true, true, tte, this);
+            if (!abort) {
+                ShutdownSignalException sse = new ShutdownSignalException(true, true, null, this);
+                sse.initCause(cause);
+                throw sse;
+            }
         } catch (ShutdownSignalException sse) {
             if (!abort)
                 throw sse;

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

@@ -477,7 +477,7 @@ private void callConfirmListeners(@SuppressWarnings("unused") Command command, B
     private void asyncShutdown(Command command) throws IOException {
         ShutdownSignalException signal = new ShutdownSignalException(false,
                                                                      false,
-                                                                     command,
+                                                                     command.getMethod(),
                                                                      this);
         synchronized (_channelMutex) {
             try {

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

@@ -129,20 +129,8 @@ public void checkShutdownSignal(int expectedCode, IOException ioe) {
         checkShutdownSignal(expectedCode, sse);
     }
 
-    public void checkShutdownSignal(int expectedCode, AlreadyClosedException ace) {
-        ShutdownNotifierComponent snc = (ShutdownNotifierComponent) ace.getReference();
-        ShutdownSignalException sse = snc.getCloseReason();
-        checkShutdownSignal(expectedCode, sse);
-    }
-
     public void checkShutdownSignal(int expectedCode, ShutdownSignalException sse) {
-        Object reason = sse.getReason();
-        Method method;
-        if (reason instanceof Command) {
-            method = ((Command) reason).getMethod();
-        } else {
-            method = (Method) reason;
-        }
+        Method method = sse.getReason();
         channel = null;
         if (sse.isHardError()) {
             connection = null;

test/src/com/rabbitmq/examples/perf/MulticastParams.java

@@ -237,9 +237,8 @@ private static boolean exists(Connection connection, Checker checker) throws IOE
         }
         catch (IOException e) {
             ShutdownSignalException sse = (ShutdownSignalException) e.getCause();
-            Command closeCommand = (Command) sse.getReason();
             if (!sse.isHardError()) {
-                AMQP.Channel.Close closeMethod = (AMQP.Channel.Close) closeCommand.getMethod();
+                AMQP.Channel.Close closeMethod = (AMQP.Channel.Close) sse.getReason();
                 if (closeMethod.getReplyCode() == AMQP.NOT_FOUND) {
                     return false;
                 }