Add test cases for partitions and route

Make sure partitions are returned in correct order and non-stream
queues are filtered out.

Author: acogoluegnes

src/test/java/com/rabbitmq/stream/impl/RoutePartitionsTest.java

@@ -18,9 +18,11 @@
 import static java.util.stream.Collectors.toList;
 import static org.assertj.core.api.Assertions.assertThat;
 
+import com.rabbitmq.client.Channel;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast;
+import java.util.Arrays;
 import java.util.List;
 import java.util.UUID;
 import java.util.stream.IntStream;
@@ -114,4 +116,36 @@ void routeReturnsMultipleStreamsIfMultipleBindingsForSameKey() throws Exception
         .hasSize(2)
         .contains(superStream + "-0", superStream + "-1");
   }
+
+  @Test
+  @BrokerVersionAtLeast("3.9.6")
+  void partitionsAndRouteShouldNotReturnNonStreamQueue() throws Exception {
+    declareSuperStreamTopology(connection, superStream, 3);
+    Channel channel = connection.createChannel();
+    String nonStreamQueue = channel.queueDeclare().getQueue();
+    connection.createChannel().queueBind(nonStreamQueue, superStream, "not-a-stream");
+    Client client = cf.get();
+    List<String> streams = client.partitions(superStream);
+    assertThat(streams)
+        .hasSize(partitions)
+        .contains(
+            IntStream.range(0, partitions)
+                .mapToObj(i -> superStream + "-" + i)
+                .toArray(String[]::new));
+    List<String> routes = client.route("not-a-stream", superStream);
+    assertThat(routes).isEmpty();
+  }
+
+  @Test
+  @BrokerVersionAtLeast("3.9.6")
+  void partitionsReturnsCorrectOrder() throws Exception {
+    String[] partitionNames = {"z", "y", "x"};
+    declareSuperStreamTopology(connection, superStream, partitionNames);
+    Client client = cf.get();
+    List<String> streams = client.partitions(superStream);
+    assertThat(streams)
+        .hasSize(partitions)
+        .containsSequence(
+            Arrays.stream(partitionNames).map(p -> superStream + "-" + p).toArray(String[]::new));
+  }
 }

src/test/java/com/rabbitmq/stream/impl/TestUtils.java

@@ -31,6 +31,8 @@
 import com.rabbitmq.stream.impl.Client.StreamMetadata;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
+import io.vavr.Tuple;
+import io.vavr.Tuple2;
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStream;
@@ -45,6 +47,7 @@
 import java.lang.reflect.Method;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
@@ -231,17 +234,28 @@ static void declareSuperStreamTopology(Connection connection, String superStream
         IntStream.range(0, partitions).mapToObj(String::valueOf).toArray(String[]::new));
   }
 
-  static void declareSuperStreamTopology(
-      Connection connection, String superStream, String... routingKeys) throws Exception {
+  static void declareSuperStreamTopology(Connection connection, String superStream, String... rks)
+      throws Exception {
     try (Channel ch = connection.createChannel()) {
       ch.exchangeDeclare(superStream, BuiltinExchangeType.DIRECT, true);
-      for (String routingKey : routingKeys) {
+
+      List<Tuple2<String, Integer>> bindings = new ArrayList<>(rks.length);
+      for (int i = 0; i < rks.length; i++) {
+        bindings.add(Tuple.of(rks[i], i));
+      }
+      // shuffle the order to make sure we get in the correct order from the server
+      Collections.shuffle(bindings);
+
+      for (Tuple2<String, Integer> binding : bindings) {
+        String routingKey = binding._1();
         String partitionName = superStream + "-" + routingKey;
         ch.queueDeclare(
             partitionName, true, false, false, Collections.singletonMap("x-queue-type", "stream"));
-        // TODO consider adding some arguments to the bindings
-        // can be useful to identify a partition, e.g. partition number
-        ch.queueBind(partitionName, superStream, routingKey);
+        ch.queueBind(
+            partitionName,
+            superStream,
+            routingKey,
+            Collections.singletonMap("x-stream-partition-order", binding._2()));
       }
     }
   }