Stop thread in DynamicBatch

Author: acogoluegnes

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

@@ -26,7 +26,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-class DynamicBatch<T> {
+final class DynamicBatch<T> implements AutoCloseable {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(DynamicBatch.class);
   private static final int MIN_BATCH_SIZE = 32;
@@ -35,11 +35,13 @@ class DynamicBatch<T> {
   private final BlockingQueue<T> requests = new LinkedBlockingQueue<>();
   private final Predicate<List<T>> consumer;
   private final int configuredBatchSize;
+  private final Thread thread;
 
   DynamicBatch(Predicate<List<T>> consumer, int batchSize) {
     this.consumer = consumer;
     this.configuredBatchSize = min(max(batchSize, MIN_BATCH_SIZE), MAX_BATCH_SIZE);
-    new Thread(this::loop).start();
+    this.thread = new Thread(this::loop);
+    this.thread.start();
   }
 
   void add(T item) {
@@ -103,4 +105,9 @@ private boolean completeBatch(List<T> items) {
       return false;
     }
   }
+
+  @Override
+  public void close() {
+    this.thread.interrupt();
+  }
 }

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

@@ -151,4 +151,9 @@ private boolean publish(List<Object> entities) {
       return false;
     }
   }
+
+  @Override
+  public void close() {
+    this.dynamicBatch.close();
+  }
 }

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

@@ -17,11 +17,14 @@
 import com.rabbitmq.stream.ConfirmationHandler;
 import com.rabbitmq.stream.Message;
 
-interface MessageAccumulator {
+interface MessageAccumulator extends AutoCloseable {
 
   void add(Message message, ConfirmationHandler confirmationHandler);
 
   int size();
 
   void flush(boolean force);
+
+  @Override
+  void close();
 }

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

@@ -112,4 +112,7 @@ private void publishBatch(boolean stateCheck) {
       producer.publishInternal(entities);
     }
   }
+
+  @Override
+  public void close() {}
 }

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

@@ -441,6 +441,7 @@ public void close() {
   }
 
   void closeFromEnvironment() {
+    this.accumulator.close();
     this.closingCallback.run();
     cancelConfirmTimeoutTask();
     this.closed.set(true);