Fix: Async close /w error from server

Author: iitsoftware

src/main/java/com/swiftmq/amqp/v100/client/Connection.java

@@ -448,10 +448,10 @@ private Session mapSessionToLocalChannel(long incomingWindowSize, long outgoingW
     /**
      * Creates a new Session.
      *
-     * @param incomingWindowSize Incoming Window Size (maxnumber of unsettled incoming transfers)
+     * @param incomingWindowSize Incoming Window Size (max number of unsettled incoming transfers)
      * @param outgoingWindowSize Outgoing Window Size (max number of unsettled outgoing transfers)
      * @return Session
-     * @throws SessionHandshakeException An error occured during handshake
+     * @throws SessionHandshakeException An error occurred during handshake
      * @throws ConnectionClosedException The connection was closed
      */
     public Session createSession(long incomingWindowSize, long outgoingWindowSize) throws SessionHandshakeException, ConnectionClosedException {

src/main/java/com/swiftmq/amqp/v100/client/ConnectionDispatcher.java

@@ -217,12 +217,12 @@ private void checkBothSidesClosed() {
     }
 
     private void notifyWaitingPOs(POObject[] po) {
-        for (int i = 0; i < po.length; i++) {
-            if (po[i] != null) {
-                po[i].setSuccess(false);
-                if (po[i].getException() == null)
-                    po[i].setException("Connection was asynchronously closed");
-                po[i].getSemaphore().notifySingleWaiter();
+        for (POObject poObject : po) {
+            if (poObject != null) {
+                poObject.setSuccess(false);
+                if (poObject.getException() == null)
+                    poObject.setException("Connection was asynchronously closed");
+                poObject.getSemaphore().notifySingleWaiter();
             }
         }
     }
@@ -584,8 +584,15 @@ public void visit(EndFrame frame) {
 
         public void visit(CloseFrame frame) {
             if (pTracer.isEnabled()) pTracer.trace(toString(), ", visit=" + frame);
-            remoteClose = frame;
-            checkBothSidesClosed();
+            // Async remote close due to error
+            if (frame.getError() != null) {
+                if (myConnection.getExceptionListener() != null)
+                    myConnection.getExceptionListener().onException(new ConnectionClosedException(frame.getError().getDescription().getValue()));
+                new Thread(() -> myConnection.cancel()).start();
+            } else {
+                remoteClose = frame;
+                checkBothSidesClosed();
+            }
         }
 
         public void visit(SaslMechanismsFrame frame) {