feat(simpleBufferTrigger): add listener for back-pressure. (PhantomThief#19)

Author: PhantomThief

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

@@ -4,14 +4,33 @@
 
 import javax.annotation.Nullable;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 /**
  * @author w.vela
  * Created on 2019-07-30.
  */
 class BackPressureHandler<T> implements RejectHandler<T> {
 
+    private static final Logger logger = LoggerFactory.getLogger(BackPressureHandler.class);
+
+    @Nullable
+    private final BackPressureListener<T> listener;
+
+    BackPressureHandler(BackPressureListener<T> listener) {
+        this.listener = listener;
+    }
+
     @Override
     public boolean onReject(T element, @Nullable Condition condition) {
+        if (listener != null) {
+            try {
+                listener.onHandle(element);
+            } catch (Throwable e) {
+                logger.error("", e);
+            }
+        }
         assert condition != null;
         condition.awaitUninterruptibly();
         return true;

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

@@ -0,0 +1,10 @@
+package com.github.phantomthief.collection.impl;
+
+/**
+ * @author w.vela
+ * Created on 2020-06-08.
+ */
+public interface BackPressureListener<T> {
+
+    void onHandle(T element);
+}

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

@@ -173,6 +173,19 @@ public GenericSimpleBufferTriggerBuilder<E, C> enableBackPressure() {
         return this;
     }
 
+    /**
+     * 开启背压(back-pressure)能力
+     * 注意，当开启背压时，需要配合 {@link #maxBufferCount(long)}
+     * 并且不要设置 {@link #rejectHandler}
+     *
+     * 当buffer达到最大值时，会阻塞入队线程，直到消费完当前buffer后再继续执行
+     */
+    @CheckReturnValue
+    public GenericSimpleBufferTriggerBuilder<E, C> enableBackPressure(BackPressureListener<E> listener) {
+        builder.enableBackPressure(listener);
+        return this;
+    }
+
     /**
      * use for debug and stats, like trigger thread's name.
      */

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

@@ -186,11 +186,22 @@ public <E1, C1> SimpleBufferTriggerBuilder<E1, C1> rejectHandler(Consumer<? supe
      * 当buffer达到最大值时，会阻塞入队线程，直到消费完当前buffer后再继续执行
      */
     public <E1, C1> SimpleBufferTriggerBuilder<E1, C1> enableBackPressure() {
+        return enableBackPressure(null);
+    }
+
+    /**
+     * 开启背压(back-pressure)能力
+     * 注意，当开启背压时，需要配合 {@link #maxBufferCount(long)}
+     * 并且不要设置 {@link #rejectHandler}
+     *
+     * 当buffer达到最大值时，会阻塞入队线程，直到消费完当前buffer后再继续执行
+     */
+    public <E1, C1> SimpleBufferTriggerBuilder<E1, C1> enableBackPressure(BackPressureListener<E1> listener) {
         if (this.rejectHandler != null) {
             throw new IllegalStateException("cannot enable back-pressure while reject handler was set.");
         }
         SimpleBufferTriggerBuilder<E1, C1> thisBuilder = (SimpleBufferTriggerBuilder<E1, C1>) this;
-        thisBuilder.rejectHandler = new BackPressureHandler<>();
+        thisBuilder.rejectHandler = new BackPressureHandler<>(listener);
         return thisBuilder;
     }
 

src/test/java/com/github/phantomthief/collection/impl/BackPressureTest.java

@@ -10,6 +10,7 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 
@@ -30,8 +31,9 @@ class BackPressureTest {
     @Test
     void test() {
         List<String> consumed = new ArrayList<>();
+        List<String> backPressured = Collections.synchronizedList(new ArrayList<>());
         BufferTrigger<String> buffer = BufferTrigger.<String, List<String>> simple()
-                .enableBackPressure()
+                .enableBackPressure(backPressured::add)
                 .maxBufferCount(10)
                 .interval(1, SECONDS)
                 .setContainer(() -> synchronizedList(new ArrayList<>()), List::add)
@@ -52,6 +54,7 @@ void test() {
             });
         }
         shutdownAndAwaitTermination(executor, 1, DAYS);
+        assertTrue(backPressured.size() > 10);
         buffer.manuallyDoTrigger();
         assertEquals(30, consumed.size());
         cost = System.currentTimeMillis() - cost;