Fail message if no route is found

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/Constants.java

@@ -39,6 +39,7 @@ public final class Constants {
   public static final short CODE_PRODUCER_NOT_AVAILABLE = 10_002;
   public static final short CODE_PRODUCER_CLOSED = 10_003;
   public static final short CODE_PUBLISH_CONFIRM_TIMEOUT = 10_004;
+  public static final short CODE_NO_ROUTE_FOUND = 10_005;
 
   public static final short COMMAND_DECLARE_PUBLISHER = 1;
   public static final short COMMAND_PUBLISH = 2;

src/main/java/com/rabbitmq/stream/impl/SuperStreamProducer.java

@@ -15,6 +15,8 @@
 
 import com.rabbitmq.stream.Codec;
 import com.rabbitmq.stream.ConfirmationHandler;
+import com.rabbitmq.stream.ConfirmationStatus;
+import com.rabbitmq.stream.Constants;
 import com.rabbitmq.stream.Message;
 import com.rabbitmq.stream.MessageBuilder;
 import com.rabbitmq.stream.Producer;
@@ -92,15 +94,20 @@ public void send(Message message, ConfirmationHandler confirmationHandler) {
     // TODO handle when the stream is not found (no partition found for the message)
     // and call the confirmation handler with a failure
     List<String> streams = this.routingStrategy.route(message, superStreamMetadata);
-    for (String stream : streams) {
-      Producer producer =
-          producers.computeIfAbsent(
-              stream,
-              stream1 -> {
-                Producer p = producerBuilder.duplicate().stream(stream1).build();
-                return p;
-              });
-      producer.send(message, confirmationHandler);
+    if (streams.isEmpty()) {
+      confirmationHandler.handle(
+          new ConfirmationStatus(message, false, Constants.CODE_NO_ROUTE_FOUND));
+    } else {
+      for (String stream : streams) {
+        Producer producer =
+            producers.computeIfAbsent(
+                stream,
+                stream1 -> {
+                  Producer p = producerBuilder.duplicate().stream(stream1).build();
+                  return p;
+                });
+        producer.send(message, confirmationHandler);
+      }
     }
   }
 

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

@@ -22,6 +22,7 @@
 
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.stream.Constants;
 import com.rabbitmq.stream.Environment;
 import com.rabbitmq.stream.EnvironmentBuilder;
 import com.rabbitmq.stream.OffsetSpecification;
@@ -31,6 +32,8 @@
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.IntStream;
 import org.junit.jupiter.api.AfterEach;
@@ -175,6 +178,39 @@ void allMessagesSentToSuperStreamWithRoutingKeyRoutingShouldBeThenConsumed() thr
         .isEqualTo(messageCount);
   }
 
+  @Test
+  void messageIsNackedIfNoRouteFound() throws Exception {
+    int messageCount = 10_000;
+    routingKeys = new String[] {"amer", "emea", "apac"};
+    declareSuperStreamTopology(connection, superStream, routingKeys);
+    Producer producer =
+        environment.producerBuilder().stream(superStream)
+            .routing(message -> message.getApplicationProperties().get("region").toString())
+            .key()
+            .producerBuilder()
+            .build();
+
+    AtomicBoolean confirmed = new AtomicBoolean(true);
+    AtomicInteger code = new AtomicInteger();
+    CountDownLatch publishLatch = new CountDownLatch(1);
+    producer.send(
+        producer
+            .messageBuilder()
+            .applicationProperties()
+            .entry("region", "atlantis")
+            .messageBuilder()
+            .build(),
+        confirmationStatus -> {
+          confirmed.set(confirmationStatus.isConfirmed());
+          code.set(confirmationStatus.getCode());
+          publishLatch.countDown();
+        });
+
+    assertThat(latchAssert(publishLatch)).completes(5);
+    assertThat(confirmed).isFalse();
+    assertThat(code).hasValue(Constants.CODE_NO_ROUTE_FOUND);
+  }
+
   @Test
   void getLastPublishingIdShouldReturnLowestValue() throws Exception {
     int messageCount = 10_000;