Fix upgrade rabbitmq version cause BuiltinExchangeType has 4 new types.

Author: Technoboy-

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/AmqpExchange.java

@@ -34,10 +34,20 @@ public interface AmqpExchange {
      *
      */
     enum Type{
-        Direct,
-        Fanout,
-        Topic,
-        Headers;
+        Direct("direct"),
+        Fanout("fanout"),
+        Topic("topic"),
+        Headers("headers"),
+        ConsistentHash("x-consistent-hash"),
+        ModulusHash("x-modulus-hash"),
+        LocalRandom("x-local-random"),
+        Random("x-random");
+
+        private final String type;
+
+        Type(String type) {
+            this.type = type;
+        }
 
         public static Type value(String type) {
             if (type == null || type.length() == 0) {
@@ -53,11 +63,24 @@ public static Type value(String type) {
                     return Topic;
                 case "headers":
                     return Headers;
+                case "x-consistent-hash":
+                    return ConsistentHash;
+                case "x-modulus-hash":
+                    return ModulusHash;
+                case "x-local-random":
+                    return LocalRandom;
+                case "x-random":
+                    return Random;
                 default:
                     return null;
             }
         }
 
+        @Override
+        public String toString() {
+            return type;
+        }
+
     }
 
     /**

amqp-impl/src/main/java/io/streamnative/pulsar/handlers/amqp/ExchangeMessageRouter.java

@@ -301,6 +301,7 @@ public static ExchangeMessageRouter getInstance(PersistentExchange exchange, Exe
             case Direct -> new DirectExchangeMessageRouter(exchange, routeExecutor);
             case Topic -> new TopicExchangeMessageRouter(exchange, routeExecutor);
             case Headers -> new HeadersExchangeMessageRouter(exchange, routeExecutor);
+            default -> null;
         };
     }
 

tests/src/test/java/io/streamnative/pulsar/handlers/amqp/rabbitmq/functional/ExchangeDeclareTest.java

@@ -16,6 +16,7 @@
 
 package io.streamnative.pulsar.handlers.amqp.rabbitmq.functional;
 
+import static org.junit.Assert.assertEquals;
 import com.rabbitmq.client.BuiltinExchangeType;
 import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
@@ -116,11 +117,9 @@ public void exchangeDeclaredWithEnumerationEquivalentOnRecoverableConnection()
 
     private void doTestExchangeDeclaredWithEnumerationEquivalent(Channel channel)
             throws IOException, InterruptedException {
-        for (BuiltinExchangeType exchangeType : new BuiltinExchangeType[]{
-                BuiltinExchangeType.DIRECT,
-                BuiltinExchangeType.FANOUT,
-                BuiltinExchangeType.TOPIC,
-                BuiltinExchangeType.HEADERS}) {
+        assertEquals("There are 8 standard exchange types",
+                8, BuiltinExchangeType.values().length);
+        for (BuiltinExchangeType exchangeType : BuiltinExchangeType.values()) {
             channel.exchangeDeclare(NAME, exchangeType);
             verifyEquivalent(NAME, exchangeType.getType(), false, false, null);
             deleteExchangeWithRetry();