Throw exception if set_token called on RabbitMQ < 4.1

Author: acogoluegnes

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

@@ -18,6 +18,8 @@
 package com.rabbitmq.client.amqp.impl;
 
 import static com.rabbitmq.client.amqp.Resource.State.*;
+import static com.rabbitmq.client.amqp.impl.Utils.supportFilterExpressions;
+import static com.rabbitmq.client.amqp.impl.Utils.supportSetToken;
 
 import com.rabbitmq.client.amqp.*;
 import com.rabbitmq.client.amqp.ObservationCollector;
@@ -72,7 +74,7 @@ final class AmqpConnection extends ResourceBase implements Connection {
   private final ConnectionSettings.AffinityStrategy affinityStrategy;
   private final String name;
   private final Lock instanceLock = new ReentrantLock();
-  private final boolean filterExpressionsSupported;
+  private final boolean filterExpressionsSupported, setTokenSupported;
   private volatile ExecutorService dispatchingExecutorService;
 
   AmqpConnection(AmqpConnectionBuilder builder) {
@@ -128,8 +130,9 @@ final class AmqpConnection extends ResourceBase implements Connection {
             ConnectionUtils.NO_RETRY_STRATEGY,
             this.name());
     this.sync(ncw);
-    this.filterExpressionsSupported =
-        Utils.supportFilterExpressions(brokerVersion(this.nativeConnection));
+    String brokerVesion = brokerVersion(this.nativeConnection);
+    this.filterExpressionsSupported = supportFilterExpressions(brokerVesion);
+    this.setTokenSupported = supportSetToken(brokerVesion);
     LOGGER.debug("Opened connection '{}' on node '{}'.", this.name(), this.connectionNodename());
     this.state(OPEN);
     this.environment.metricsCollector().openConnection();
@@ -708,6 +711,10 @@ boolean filterExpressionsSupported() {
     return this.filterExpressionsSupported;
   }
 
+  boolean setTokenSupported() {
+    return this.setTokenSupported;
+  }
+
   long id() {
     return this.id;
   }

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

@@ -172,6 +172,9 @@ public UnbindSpecification unbind() {
   }
 
   void setToken(String token) {
+    if (!this.connection.setTokenSupported()) {
+      throw new UnsupportedOperationException("Token renewal requires at least RabbitMQ 4.1.0");
+    }
     checkAvailable();
     UUID requestId = messageId();
     try {

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

@@ -233,6 +233,10 @@ static boolean supportFilterExpressions(String brokerVersion) {
     return is4_1_OrMore(brokerVersion);
   }
 
+  static boolean supportSetToken(String brokerVersion) {
+    return is4_1_OrMore(brokerVersion);
+  }
+
   static final class ObservationConnectionInfo implements ObservationCollector.ConnectionInfo {
 
     private final String address;