Record recoverable entities in RPC channel methods

[#162201529]

Fixes #425

(cherry picked from commit f560108)

Conflicts:
	src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringChannel.java
(cherry picked from commit 62d0584)

Author: acogoluegnes

pom.xml

@@ -75,14 +75,15 @@
     <groovy.all.version>2.4.8</groovy.all.version>
     <keytool.maven.plugin.version>1.5</keytool.maven.plugin.version>
     <build.helper.maven-plugin.version>1.12</build.helper.maven-plugin.version>
-    <maven.compiler.plugin.version>3.5.1</maven.compiler.plugin.version>
+    <maven.compiler.plugin.version>3.8.0</maven.compiler.plugin.version>
     <maven.surefire.plugin.version>2.19.1</maven.surefire.plugin.version>
     <maven.failsafe.plugin.version>2.19.1</maven.failsafe.plugin.version>
     <maven.gpg.plugin.version>1.6</maven.gpg.plugin.version>
     <maven.jar.plugin.version>3.0.2</maven.jar.plugin.version>
     <maven.bundle.plugin.version>2.3.7</maven.bundle.plugin.version>
     <maven.packagecloud.wagon.version>0.0.6</maven.packagecloud.wagon.version>
     <checksum.maven.plugin.version>1.8</checksum.maven.plugin.version>
+    <animal-sniffer-maven-plugin.version>1.16</animal-sniffer-maven-plugin.version>
 
     <!--
     These groovy scripts are used later in this POM file to generate
@@ -872,13 +873,51 @@
         <configuration>
           <source>1.6</source>
           <target>1.6</target>
+          <testSource>1.8</testSource>
+          <testTarget>1.8</testTarget>
           <compilerArgs>
             <arg>-Xlint:deprecation</arg>
             <arg>-Xlint:unchecked</arg>
           </compilerArgs>
         </configuration>
       </plugin>
 
+      <plugin>
+        <groupId>org.codehaus.mojo</groupId>
+        <artifactId>animal-sniffer-maven-plugin</artifactId>
+        <version>${animal-sniffer-maven-plugin.version}</version>
+        <executions>
+          <execution>
+            <id>signature-check-in-verify</id>
+            <phase>verify</phase>
+            <goals>
+              <goal>check</goal>
+            </goals>
+          </execution>
+          <execution>
+            <id>signature-check-before-packaging</id>
+            <phase>package</phase>
+            <goals>
+              <goal>check</goal>
+            </goals>
+          </execution>
+        </executions>
+        <configuration>
+          <ignores>
+            <!--
+              To avoid failure because of call to SSLParameters#setEndpointIdentificationAlgorithm (Java 7)
+              It is mentioned in the Javadoc that the class making this call requires Java 7.
+            -->
+            <ignore>javax.net.ssl.SSLParameters</ignore>
+          </ignores>
+          <signature>
+            <groupId>org.codehaus.mojo.signature</groupId>
+            <artifactId>java16</artifactId>
+            <version>1.0</version>
+          </signature>
+        </configuration>
+      </plugin>
+
       <!--
       Disable the unit tests plugin because we only have integration
       tests.

src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringChannel.java

@@ -16,7 +16,9 @@
 package com.rabbitmq.client.impl.recovery;
 
 import com.rabbitmq.client.*;
-import com.rabbitmq.client.RecoverableChannel;
+import com.rabbitmq.client.impl.AMQCommand;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.util.*;
@@ -30,6 +32,9 @@
  * @since 3.3.0
  */
 public class AutorecoveringChannel implements RecoverableChannel {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(AutorecoveringChannel.class);
+
     private volatile RecoveryAwareChannelN delegate;
     private volatile AutorecoveringConnection connection;
     private final List<ShutdownListener> shutdownHooks  = new CopyOnWriteArrayList<ShutdownListener>();
@@ -235,12 +240,7 @@ public AMQP.Exchange.DeclareOk exchangeDeclare(String exchange, BuiltinExchangeT
     @Override
     public AMQP.Exchange.DeclareOk exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments) throws IOException {
         final AMQP.Exchange.DeclareOk ok = delegate.exchangeDeclare(exchange, type, durable, autoDelete, internal, arguments);
-        RecordedExchange x = new RecordedExchange(this, exchange).
-          type(type).
-          durable(durable).
-          autoDelete(autoDelete).
-          arguments(arguments);
-        recordExchange(exchange, x);
+        recordExchange(ok, exchange, type, durable, autoDelete, arguments);
         return ok;
     }
 
@@ -331,15 +331,7 @@ public AMQP.Queue.DeclareOk queueDeclare() throws IOException {
     @Override
     public AMQP.Queue.DeclareOk queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) throws IOException {
         final AMQP.Queue.DeclareOk ok = delegate.queueDeclare(queue, durable, exclusive, autoDelete, arguments);
-        RecordedQueue q = new RecordedQueue(this, ok.getQueue()).
-            durable(durable).
-            exclusive(exclusive).
-            autoDelete(autoDelete).
-            arguments(arguments);
-        if (queue.equals(RecordedQueue.EMPTY_STRING)) {
-            q.serverNamed(true);
-        }
-        recordQueue(ok, q);
+        recordQueue(ok, queue, durable, exclusive, autoDelete, arguments);
         return ok;
     }
 
@@ -356,7 +348,6 @@ public void queueDeclareNoWait(String queue,
             arguments(arguments);
         delegate.queueDeclareNoWait(queue, durable, exclusive, autoDelete, arguments);
         recordQueue(queue, meta);
-
     }
 
     @Override
@@ -546,7 +537,10 @@ public void asyncRpc(Method method) throws IOException {
 
     @Override
     public Command rpc(Method method) throws IOException {
-        return delegate.rpc(method);
+        recordOnRpcRequest(method);
+        AMQCommand response = delegate.rpc(method);
+        recordOnRpcResponse(response.getMethod(), method);
+        return response;
     }
 
     /**
@@ -680,6 +674,18 @@ private boolean deleteRecordedExchangeBinding(String destination, String source,
         return this.connection.deleteRecordedExchangeBinding(this, destination, source, routingKey, arguments);
     }
 
+    private void recordQueue(AMQP.Queue.DeclareOk ok, String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
+        RecordedQueue q = new RecordedQueue(this, ok.getQueue()).
+                durable(durable).
+                exclusive(exclusive).
+                autoDelete(autoDelete).
+                arguments(arguments);
+        if (queue.equals(RecordedQueue.EMPTY_STRING)) {
+            q.serverNamed(true);
+        }
+        recordQueue(ok, q);
+    }
+
     private void recordQueue(AMQP.Queue.DeclareOk ok, RecordedQueue q) {
         this.connection.recordQueue(ok, q);
     }
@@ -692,6 +698,15 @@ private void deleteRecordedQueue(String queue) {
         this.connection.deleteRecordedQueue(queue);
     }
 
+    private void recordExchange(AMQP.Exchange.DeclareOk ok, String exchange, String type, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
+        RecordedExchange x = new RecordedExchange(this, exchange).
+                type(type).
+                durable(durable).
+                autoDelete(autoDelete).
+                arguments(arguments);
+        recordExchange(exchange, x);
+    }
+
     private void recordExchange(String exchange, RecordedExchange x) {
         this.connection.recordExchange(exchange, x);
     }
@@ -736,6 +751,74 @@ void updateConsumerTag(String tag, String newTag) {
         }
     }
 
+    private void recordOnRpcRequest(Method method) {
+        if (method instanceof AMQP.Queue.Delete) {
+            deleteRecordedQueue(((AMQP.Queue.Delete) method).getQueue());
+        } else if (method instanceof AMQP.Exchange.Delete) {
+            deleteRecordedExchange(((AMQP.Exchange.Delete) method).getExchange());
+        } else if (method instanceof AMQP.Queue.Unbind) {
+            AMQP.Queue.Unbind unbind = (AMQP.Queue.Unbind) method;
+            deleteRecordedQueueBinding(
+                    unbind.getQueue(), unbind.getExchange(),
+                    unbind.getRoutingKey(), unbind.getArguments()
+            );
+            this.maybeDeleteRecordedAutoDeleteExchange(unbind.getExchange());
+        } else if (method instanceof AMQP.Exchange.Unbind) {
+            AMQP.Exchange.Unbind unbind = (AMQP.Exchange.Unbind) method;
+            deleteRecordedExchangeBinding(
+                    unbind.getDestination(), unbind.getSource(),
+                    unbind.getRoutingKey(), unbind.getArguments()
+            );
+            this.maybeDeleteRecordedAutoDeleteExchange(unbind.getSource());
+        }
+    }
+
+    private void recordOnRpcResponse(Method response, Method request) {
+        if (response instanceof AMQP.Queue.DeclareOk) {
+            if (request instanceof AMQP.Queue.Declare) {
+                AMQP.Queue.DeclareOk ok = (AMQP.Queue.DeclareOk) response;
+                AMQP.Queue.Declare declare = (AMQP.Queue.Declare) request;
+                recordQueue(
+                        ok, declare.getQueue(),
+                        declare.getDurable(), declare.getExclusive(), declare.getAutoDelete(),
+                        declare.getArguments()
+                );
+            } else {
+                LOGGER.warn("RPC response {} and RPC request {} not compatible, topology not recorded.",
+                        response.getClass(), request.getClass());
+            }
+        } else if (response instanceof AMQP.Exchange.DeclareOk) {
+            if (request instanceof AMQP.Exchange.Declare) {
+                AMQP.Exchange.DeclareOk ok = (AMQP.Exchange.DeclareOk) response;
+                AMQP.Exchange.Declare declare = (AMQP.Exchange.Declare) request;
+                recordExchange(
+                        ok, declare.getExchange(), declare.getType(),
+                        declare.getDurable(), declare.getAutoDelete(),
+                        declare.getArguments()
+                );
+            } else {
+                LOGGER.warn("RPC response {} and RPC request {} not compatible, topology not recorded.",
+                        response.getClass(), request.getClass());
+            }
+        } else if (response instanceof AMQP.Queue.BindOk) {
+            if (request instanceof AMQP.Queue.Bind) {
+                AMQP.Queue.Bind bind = (AMQP.Queue.Bind) request;
+                recordQueueBinding(bind.getQueue(), bind.getExchange(), bind.getRoutingKey(), bind.getArguments());
+            } else {
+                LOGGER.warn("RPC response {} and RPC request {} not compatible, topology not recorded.",
+                        response.getClass(), request.getClass());
+            }
+        } else if (response instanceof AMQP.Exchange.BindOk) {
+            if (request instanceof AMQP.Exchange.Bind) {
+                AMQP.Exchange.Bind bind = (AMQP.Exchange.Bind) request;
+                recordExchangeBinding(bind.getDestination(), bind.getSource(), bind.getRoutingKey(), bind.getArguments());
+            } else {
+                LOGGER.warn("RPC response {} and RPC request {} not compatible, topology not recorded.",
+                        response.getClass(), request.getClass());
+            }
+        }
+    }
+
     @Override
     public String toString() {
         return this.delegate.toString();

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

@@ -63,7 +63,8 @@
     AddressTest.class,
     DefaultRetryHandlerTest.class,
     NioDeadlockOnConnectionClosing.class,
-    GeneratedClassesTest.class
+    GeneratedClassesTest.class,
+    RpcTopologyRecordingTest.class
 })
 public class ClientTests {