feat(common): add AutoCloseable for buffer trigger. (PhantomThief#20)

Author: PhantomThief

src/main/java/com/github/phantomthief/collection/BufferTrigger.java

@@ -10,8 +10,11 @@
 /**
  * @author w.vela
  */
-public interface BufferTrigger<E> {
+public interface BufferTrigger<E> extends AutoCloseable {
 
+    /**
+     * @throws IllegalStateException if the instance has been shutdown.
+     */
     void enqueue(E element);
 
     void manuallyDoTrigger();
@@ -48,4 +51,7 @@ static <E> GenericBatchConsumerTriggerBuilder<E> batchBlocking() {
     static BatchConsumerTriggerBuilder<Object> batchBlockingTrigger() {
         return BatchConsumeBlockingQueueTrigger.newBuilder();
     }
+
+    @Override
+    void close(); // override to remove throws Exception.
 }

src/main/java/com/github/phantomthief/collection/impl/BatchConsumeBlockingQueueTrigger.java

@@ -3,15 +3,19 @@
 import static com.github.phantomthief.concurrent.MoreFutures.scheduleWithDynamicDelay;
 import static com.github.phantomthief.util.MoreLocks.runWithLock;
 import static com.github.phantomthief.util.MoreLocks.runWithTryLock;
+import static com.google.common.base.Preconditions.checkState;
+import static com.google.common.util.concurrent.MoreExecutors.shutdownAndAwaitTermination;
 import static com.google.common.util.concurrent.Uninterruptibles.putUninterruptibly;
 import static java.lang.Integer.max;
 import static java.lang.Math.min;
+import static java.util.concurrent.TimeUnit.DAYS;
 import static org.slf4j.LoggerFactory.getLogger;
 
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -38,6 +42,9 @@ public class BatchConsumeBlockingQueueTrigger<E> implements BufferTrigger<E> {
     private final ScheduledExecutorService scheduledExecutorService;
     private final ReentrantLock lock = new ReentrantLock();
     private final AtomicBoolean running = new AtomicBoolean();
+    private final Runnable shutdownExecutor;
+
+    private volatile boolean shutdown;
 
     BatchConsumeBlockingQueueTrigger(BatchConsumerTriggerBuilder<E> builder) {
         Supplier<Duration> linger = builder.linger;
@@ -46,7 +53,13 @@ public class BatchConsumeBlockingQueueTrigger<E> implements BufferTrigger<E> {
         this.consumer = builder.consumer;
         this.exceptionHandler = builder.exceptionHandler;
         this.scheduledExecutorService = builder.scheduledExecutorService;
-        scheduleWithDynamicDelay(scheduledExecutorService, linger, () -> doBatchConsumer(TriggerType.LINGER));
+        Future<?> future = scheduleWithDynamicDelay(scheduledExecutorService, linger, () -> doBatchConsumer(TriggerType.LINGER));
+        this.shutdownExecutor = () -> {
+            future.cancel(false);
+            if (builder.usingInnerExecutor) {
+                shutdownAndAwaitTermination(builder.scheduledExecutorService, 1, DAYS);
+            }
+        };
     }
 
     /**
@@ -60,6 +73,8 @@ public static BatchConsumerTriggerBuilder<Object> newBuilder() {
 
     @Override
     public void enqueue(E element) {
+        checkState(!shutdown, "buffer trigger was shutdown.");
+
         putUninterruptibly(queue, element);
         tryTrigBatchConsume();
     }
@@ -135,6 +150,16 @@ public long getPendingChanges() {
         return queue.size();
     }
 
+    @Override
+    public void close() {
+        shutdown = true;
+        try {
+            manuallyDoTrigger();
+        } finally {
+            shutdownExecutor.run();
+        }
+    }
+
     /**
      * just for log and debug
      */

src/main/java/com/github/phantomthief/collection/impl/BatchConsumerTriggerBuilder.java

@@ -26,12 +26,16 @@ public final class BatchConsumerTriggerBuilder<E> {
     private static final Duration DEFAULT_LINGER = ofSeconds(1);
 
     ScheduledExecutorService scheduledExecutorService;
+    boolean usingInnerExecutor;
     Supplier<Duration> linger;
     int batchSize;
     int bufferSize;
     ThrowableConsumer<List<E>, Exception> consumer;
     BiConsumer<Throwable, List<E>> exceptionHandler;
 
+    /**
+     * If you create own ScheduledExecutorService, then you have to shutdown it yourself.
+     */
     public BatchConsumerTriggerBuilder<E>
             setScheduleExecutorService(ScheduledExecutorService scheduledExecutorService) {
         this.scheduledExecutorService = scheduledExecutorService;
@@ -135,6 +139,7 @@ private void ensure() {
         }
         if (scheduledExecutorService == null) {
             scheduledExecutorService = makeScheduleExecutor();
+            usingInnerExecutor = true;
         }
     }
 

src/main/java/com/github/phantomthief/collection/impl/GenericBatchConsumerTriggerBuilder.java

@@ -29,6 +29,9 @@ public GenericBatchConsumerTriggerBuilder<E> forceConsumeEveryTick() {
         return this;
     }
 
+    /**
+     * If you create own ScheduledExecutorService, then you have to shutdown it yourself.
+     */
     public GenericBatchConsumerTriggerBuilder<E>
             setScheduleExecutorService(ScheduledExecutorService scheduledExecutorService) {
         builder.setScheduleExecutorService(scheduledExecutorService);

src/main/java/com/github/phantomthief/collection/impl/GenericSimpleBufferTriggerBuilder.java

@@ -54,6 +54,9 @@ public GenericSimpleBufferTriggerBuilder<E, C> setContainerEx(Supplier<? extends
         return this;
     }
 
+    /**
+     * If you create own ScheduledExecutorService, then you have to shutdown it yourself.
+     */
     @CheckReturnValue
     public GenericSimpleBufferTriggerBuilder<E, C>
             setScheduleExecutorService(ScheduledExecutorService scheduledExecutorService) {

src/main/java/com/github/phantomthief/collection/impl/LazyBufferTrigger.java

@@ -32,4 +32,9 @@ public void manuallyDoTrigger() {
     public long getPendingChanges() {
         return this.factory.map(BufferTrigger::getPendingChanges).orElse(0L);
     }
+
+    @Override
+    public void close() {
+        this.factory.ifPresent(BufferTrigger::close);
+    }
 }

src/main/java/com/github/phantomthief/collection/impl/SimpleBufferTrigger.java

@@ -1,6 +1,8 @@
 package com.github.phantomthief.collection.impl;
 
+import static com.google.common.base.Preconditions.checkState;
 import static com.google.common.base.Throwables.throwIfUnchecked;
+import static com.google.common.util.concurrent.MoreExecutors.shutdownAndAwaitTermination;
 import static java.lang.System.currentTimeMillis;
 import static java.util.concurrent.TimeUnit.DAYS;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
@@ -46,7 +48,9 @@ public class SimpleBufferTrigger<E, C> implements BufferTrigger<E> {
     private final ReadLock readLock;
     private final WriteLock writeLock;
     private final Condition writeCondition;
+    private final Runnable shutdownExecutor;
 
+    private volatile boolean shutdown;
     private volatile long lastConsumeTimestamp = currentTimeMillis();
 
     SimpleBufferTrigger(SimpleBufferTriggerBuilder<E, C> builder) {
@@ -70,6 +74,11 @@ public class SimpleBufferTrigger<E, C> implements BufferTrigger<E> {
         builder.scheduledExecutorService.schedule(
                 new TriggerRunnable(builder.scheduledExecutorService, builder.triggerStrategy),
                 DEFAULT_NEXT_TRIGGER_PERIOD, MILLISECONDS);
+        this.shutdownExecutor = () -> {
+            if (builder.usingInnerExecutor) {
+                shutdownAndAwaitTermination(builder.scheduledExecutorService, 1, DAYS);
+            }
+        };
     }
 
     /**
@@ -99,6 +108,8 @@ public static SimpleBufferTriggerBuilder<Object, Map<Object, Integer>> newCounte
 
     @Override
     public void enqueue(E element) {
+        checkState(!shutdown, "buffer trigger was shutdown.");
+
         long currentCount = counter.get();
         long thisMaxBufferCount = maxBufferCount.getAsLong();
         if (thisMaxBufferCount > 0 && currentCount >= thisMaxBufferCount) {
@@ -195,6 +206,16 @@ public long getPendingChanges() {
         return counter.get();
     }
 
+    @Override
+    public void close() {
+        shutdown = true;
+        try {
+            manuallyDoTrigger();
+        } finally {
+            shutdownExecutor.run();
+        }
+    }
+
     public interface TriggerStrategy {
 
         TriggerResult canTrigger(long lastConsumeTimestamp, long changedCount);
@@ -249,7 +270,9 @@ public void run() {
                     logger.error("", e);
                 }
                 nextTrigPeriod = Math.max(0, nextTrigPeriod);
-                scheduledExecutorService.schedule(this, nextTrigPeriod, MILLISECONDS);
+                if (!shutdown) {
+                    scheduledExecutorService.schedule(this, nextTrigPeriod, MILLISECONDS);
+                }
             }
         }
     }

src/main/java/com/github/phantomthief/collection/impl/SimpleBufferTriggerBuilder.java

@@ -38,6 +38,7 @@ public class SimpleBufferTriggerBuilder<E, C> {
 
     TriggerStrategy triggerStrategy;
     ScheduledExecutorService scheduledExecutorService;
+    boolean usingInnerExecutor;
     Supplier<C> bufferFactory;
     ToIntBiFunction<C, E> queueAdder;
     ThrowableConsumer<C, Throwable> consumer;
@@ -80,6 +81,9 @@ public <E1, C1> SimpleBufferTriggerBuilder<E1, C1> setContainerEx(
         return thisBuilder;
     }
 
+    /**
+     * If you create own ScheduledExecutorService, then you have to shutdown it yourself.
+     */
     public SimpleBufferTriggerBuilder<E, C>
             setScheduleExecutorService(ScheduledExecutorService scheduledExecutorService) {
         this.scheduledExecutorService = scheduledExecutorService;
@@ -261,6 +265,7 @@ private void ensure() {
         }
         if (scheduledExecutorService == null) {
             scheduledExecutorService = makeScheduleExecutor();
+            usingInnerExecutor = true;
         }
     }