Mark direct reply-to supported for RabbitMQ 4.2+

Author: acogoluegnes

.github/workflows/test-pr.yml

@@ -23,8 +23,6 @@ jobs:
           cache: 'maven'
       - name: Start broker
         run: ci/start-broker.sh
-        env:
-          RABBITMQ_IMAGE: pivotalrabbitmq/rabbitmq:pr-14474-otp28
       - name: Start toxiproxy
         run: ci/start-toxiproxy.sh
       - name: Display Java version

src/main/java/com/rabbitmq/client/amqp/impl/AmqpConnection.java

@@ -872,9 +872,7 @@ boolean setTokenSupported() {
   }
 
   boolean directReplyToSupported() {
-    // TODO use flag when version for direct reply-to is known
-    return true;
-    //    return this.directReplyToSupported;
+    return this.directReplyToSupported;
   }
 
   long id() {

src/main/java/com/rabbitmq/client/amqp/impl/AmqpRpcClient.java

@@ -83,6 +83,7 @@ final class AmqpRpcClient implements RpcClient {
       this.correlationIdExtractor = builder.correlationIdExtractor();
     }
     AmqpConsumerBuilder consumerBuilder = (AmqpConsumerBuilder) this.connection.consumerBuilder();
+    LOGGER.debug("Using direct reply-to: {}", this.connection.directReplyToSupported());
     this.consumer =
         (AmqpConsumer)
             consumerBuilder

src/main/java/com/rabbitmq/client/amqp/impl/Utils.java

@@ -239,10 +239,6 @@ static boolean is4_2_OrMore(String brokerVersion) {
     return atLeastVersion(brokerVersion, "4.2.0");
   }
 
-  static boolean is4_3_OrMore(String brokerVersion) {
-    return atLeastVersion(brokerVersion, "4.3.0");
-  }
-
   private static boolean atLeastVersion(String brokerVersion, String expectedVersion) {
     try {
       return versionCompare(currentVersion(brokerVersion), expectedVersion) >= 0;
@@ -265,7 +261,7 @@ static boolean supportSqlFilterExpressions(String brokerVersion) {
   }
 
   static boolean supportDirectReplyTo(String brokerVersion) {
-    return is4_3_OrMore(brokerVersion);
+    return is4_2_OrMore(brokerVersion);
   }
 
   static final class ObservationConnectionInfo implements ObservationCollector.ConnectionInfo {