Clean state in AutorecoveringChannel#abort

Fixes #661
rabbitmq · Oct 19, 2020 · 4d6847f · 4d6847f
1 parent 04a5e5c
commit 4d6847f
Showing 1 changed file with 24 additions and 15 deletions.
diff --git a/src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringChannel.java b/src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringChannel.java
@@ -17,6 +17,7 @@
 
 import com.rabbitmq.client.*;
 import com.rabbitmq.client.impl.AMQCommand;
+import com.rabbitmq.client.impl.recovery.Utils.IoTimeoutExceptionRunnable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -69,33 +70,41 @@ public Channel getDelegate() {
 
     @Override
     public void close() throws IOException, TimeoutException {
-        try {
-            delegate.close();
-        } finally {
-            for (String consumerTag : consumerTags) {
-                this.connection.deleteRecordedConsumer(consumerTag);
-            }
-            this.connection.unregisterChannel(this);
-        }
+        executeAndClean(() -> delegate.close());
     }
 
     @Override
     public void close(int closeCode, String closeMessage) throws IOException, TimeoutException {
-        try {
-          delegate.close(closeCode, closeMessage);
-        } finally {
-          this.connection.unregisterChannel(this);
-        }
+        executeAndClean(() -> delegate.close(closeCode, closeMessage));
     }
 
     @Override
     public void abort() throws IOException {
-        delegate.abort();
+        try {
+            executeAndClean(() -> delegate.abort());
+        } catch (TimeoutException e) {
+            // abort() ignores exceptions
+        }
     }
 
     @Override
     public void abort(int closeCode, String closeMessage) throws IOException {
-        delegate.abort(closeCode, closeMessage);
+        try {
+            executeAndClean(() -> delegate.abort(closeCode, closeMessage));
+        } catch (TimeoutException e) {
+            // abort() ignores exceptions
+        }
+    }
+
+    private void executeAndClean(IoTimeoutExceptionRunnable callback) throws IOException, TimeoutException {
+        try {
+            callback.run();
+        } finally {
+            for (String consumerTag : consumerTags) {
+                this.connection.deleteRecordedConsumer(consumerTag);
+            }
+            this.connection.unregisterChannel(this);
+        }
     }
 
     @Override