Update amqp-client to 5.10.0

Change ShutdownListeners to lambda expressions.
https://github.com/rabbitmq/rabbitmq-java-client/releases/tag/v5.0.0

Instead of setting channel and connection to null in ShutdownListeners,
trust ShutdownNotifierComponent#isOpen() that returns false if there
is a shutdown-cause.

Change-Id: Ia8304b775fb470613b4ae40210a81f6b1ffb55f9

external_plugin_deps.bzl

diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl
index 0af44ef..2827eee 100644
--- a/external_plugin_deps.bzl
+++ b/external_plugin_deps.bzl
@@ -3,6 +3,6 @@
 def external_plugin_deps():
     maven_jar(
         name = "amqp_client",
-        artifact = "com.rabbitmq:amqp-client:4.1.1",
-        sha1 = "256f6c92c55a8d3cfae8d32e1a15713baedab184",
+        artifact = "com.rabbitmq:amqp-client:5.10.0",
+        sha1 = "4de351467a13b8ca4eb7e8023032f9f964a21796",
     )

src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSession.java

diff --git a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSession.java b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSession.java
index c77ed05..1d3c4f0 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSession.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/rabbitmq/session/type/AMQPSession.java
@@ -26,8 +26,6 @@
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
-import com.rabbitmq.client.ShutdownListener;
-import com.rabbitmq.client.ShutdownNotifier;
 import com.rabbitmq.client.ShutdownSignalException;
 import java.io.IOException;
 import java.net.URISyntaxException;
@@ -41,58 +39,11 @@
 public final class AMQPSession implements Session {
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
-  private class ShutdownListenerImpl implements ShutdownListener {
-
-    private final Class<?> clazz;
-
-    <T extends ShutdownNotifier> ShutdownListenerImpl(Class<T> clazz) {
-      this.clazz = clazz;
-    }
-
-    @Override
-    public void shutdownCompleted(ShutdownSignalException cause) {
-      if (cause != null) {
-        Object obj = cause.getReference();
-        if (Channel.class.isInstance(obj)) {
-          Channel.class.cast(obj).removeShutdownListener(this);
-        } else if (Connection.class.isInstance(obj)) {
-          Connection.class.cast(obj).removeShutdownListener(this);
-        }
-        if (clazz.isInstance(obj)) {
-          if (clazz == Channel.class) {
-            Channel ch = Channel.class.cast(obj);
-            if (cause.isInitiatedByApplication()) {
-              logger.atInfo().log(MSG("Channel #%d closed by application."), ch.getChannelNumber());
-            } else {
-              logger.atWarning().log(
-                  MSG("Channel #%dclosed. Cause: %s"), ch.getChannelNumber(), cause.getMessage());
-            }
-            if (ch.equals(AMQPSession.this.channel)) {
-              AMQPSession.this.channel = null;
-            }
-          } else if (clazz == Connection.class) {
-            Connection conn = Connection.class.cast(obj);
-            if (cause.isInitiatedByApplication()) {
-              logger.atInfo().log(MSG("Connection closed by application."));
-            } else {
-              logger.atWarning().log(MSG("Connection closed. Cause: %s"), cause.getMessage());
-            }
-            if (conn.equals(AMQPSession.this.connection)) {
-              AMQPSession.this.connection = null;
-            }
-          }
-        }
-      }
-    }
-  }
-
   private final Properties properties;
   private volatile Connection connection;
   private volatile Channel channel;
 
   private final AtomicInteger failureCount = new AtomicInteger(0);
-  private final ShutdownListener connectionListener = new ShutdownListenerImpl(Connection.class);
-  private final ShutdownListener channelListener = new ShutdownListenerImpl(Channel.class);
 
   public AMQPSession(Properties properties) {
     this.properties = properties;
@@ -104,22 +55,31 @@
 
   @Override
   public boolean isOpen() {
-    if (connection != null) {
+    if (connection != null && connection.isOpen()) {
       return true;
     }
     return false;
   }
 
   private Channel getChannel() {
-    Channel ch = null;
-    if (connection == null) {
+    if (!isOpen()) {
       connect();
     } else {
       try {
-        ch = connection.createChannel();
-        ch.addShutdownListener(channelListener);
+        Channel ch = connection.createChannel();
+        int channelId = ch.getChannelNumber();
+        ch.addShutdownListener(
+            cause -> {
+              if (cause.isInitiatedByApplication()) {
+                logger.atInfo().log(MSG("Channel #%d closed by application."), channelId);
+              } else {
+                logger.atWarning().log(
+                    MSG("Channel #%d closed. Cause: %s"), channelId, cause.getMessage());
+              }
+            });
         failureCount.set(0);
-        logger.atInfo().log(MSG("Channel #%d opened."), ch.getChannelNumber());
+        logger.atInfo().log(MSG("Channel #%d opened."), channelId);
+        return ch;
       } catch (IOException | AlreadyClosedException ex) {
         logger.atSevere().withCause(ex).log(MSG("Failed to open channel."));
         failureCount.incrementAndGet();
@@ -130,12 +90,12 @@
         disconnect();
       }
     }
-    return ch;
+    return null;
   }
 
   @Override
   public boolean connect() {
-    if (connection != null && connection.isOpen()) {
+    if (isOpen()) {
       logger.atInfo().log(MSG("Already connected."));
       return true;
     }
@@ -156,7 +116,14 @@
           factory.setPassword(amqp.password);
         }
         connection = factory.newConnection();
-        connection.addShutdownListener(connectionListener);
+        connection.addShutdownListener(
+            cause -> {
+              if (cause.isInitiatedByApplication()) {
+                logger.atInfo().log(MSG("Connection closed by application."));
+              } else {
+                logger.atWarning().log(MSG("Connection closed. Cause: %s"), cause.getMessage());
+              }
+            });
         logger.atInfo().log(MSG("Connection established."));
         return true;
       }