Enforce super stream producer closing

Fixes #193

Author: acogoluegnes

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

@@ -27,6 +27,7 @@
 import java.util.Map.Entry;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -42,6 +43,7 @@ class SuperStreamProducer implements Producer {
   private final StreamEnvironment environment;
   private final String name;
   private final Metadata superStreamMetadata;
+  private final AtomicBoolean closed = new AtomicBoolean(false);
 
   SuperStreamProducer(
       StreamProducerBuilder producerBuilder,
@@ -91,36 +93,47 @@ public long getLastPublishingId() {
 
   @Override
   public void send(Message message, ConfirmationHandler confirmationHandler) {
-    List<String> streams = this.routingStrategy.route(message, superStreamMetadata);
-    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().superStream(null).stream(stream1).build();
-                  return p;
-                });
-        producer.send(message, confirmationHandler);
+    if (canSend()) {
+      List<String> streams = this.routingStrategy.route(message, superStreamMetadata);
+      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().superStream(null).stream(stream1).build();
+                    return p;
+                  });
+          producer.send(message, confirmationHandler);
+        }
       }
+    } else {
+      confirmationHandler.handle(
+          new ConfirmationStatus(message, false, Constants.CODE_PRODUCER_CLOSED));
     }
   }
 
+  private boolean canSend() {
+    return !this.closed.get();
+  }
+
   @Override
   public void close() {
-    for (Entry<String, Producer> entry : producers.entrySet()) {
-      try {
-        entry.getValue().close();
-      } catch (Exception e) {
-        LOGGER.info(
-            "Error while closing producer for partition {} of super stream {}: {}",
-            entry.getKey(),
-            this.superStream,
-            e.getMessage());
+    if (this.closed.compareAndSet(false, true)) {
+      for (Entry<String, Producer> entry : producers.entrySet()) {
+        try {
+          entry.getValue().close();
+        } catch (Exception e) {
+          LOGGER.info(
+              "Error while closing producer for partition {} of super stream {}: {}",
+              entry.getKey(),
+              this.superStream,
+              e.getMessage());
+        }
       }
     }
   }

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

@@ -30,6 +30,7 @@
 import com.rabbitmq.stream.Producer;
 import io.netty.channel.EventLoopGroup;
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
@@ -290,4 +291,43 @@ void getLastPublishingIdShouldReturnLowestValue() throws Exception {
     assertThat(counts.values().stream().map(AtomicLong::get).reduce(0L, Long::sum))
         .isEqualTo(messageCount);
   }
+
+  @Test
+  void producerShouldNotPublishMessagesOnceClosed() throws Exception {
+    int messageCount = 100;
+    declareSuperStreamTopology(connection, superStream, partitions);
+    String producerName = "super-stream-application";
+    Producer producer =
+        environment
+            .producerBuilder()
+            .name(producerName)
+            .superStream(superStream)
+            .routing(message -> message.getProperties().getMessageIdAsString())
+            .producerBuilder()
+            .build();
+
+    producer.close();
+
+    Set<Short> confirmationCodes = ConcurrentHashMap.newKeySet(1);
+    CountDownLatch publishLatch = new CountDownLatch(messageCount);
+
+    IntStream.range(0, messageCount)
+        .forEach(
+            i ->
+                producer.send(
+                    producer
+                        .messageBuilder()
+                        .publishingId(i)
+                        .properties()
+                        .messageId(UUID.randomUUID().toString())
+                        .messageBuilder()
+                        .build(),
+                    confirmationStatus -> {
+                      confirmationCodes.add(confirmationStatus.getCode());
+                      publishLatch.countDown();
+                    }));
+
+    assertThat(latchAssert(publishLatch)).completes(5);
+    assertThat(confirmationCodes).hasSize(1).containsExactly(Constants.CODE_PRODUCER_CLOSED);
+  }
 }