Consider isRejecting() hints with MoreGatherers.last() (#54)

pivovarit · Oct 18, 2024 · 6e18a8b · 6e18a8b
1 parent 2cdb067
commit 6e18a8b
Showing 1 changed file with 6 additions and 2 deletions.
diff --git a/src/main/java/com/pivovarit/gatherers/LastGatherer.java b/src/main/java/com/pivovarit/gatherers/LastGatherer.java
@@ -41,7 +41,11 @@ public Integrator<AppendOnlyCircularBuffer<T>, T, T> integrator() {
 
         @Override
         public BiConsumer<AppendOnlyCircularBuffer<T>, Downstream<? super T>> finisher() {
-            return (state, downstream) -> state.forEach(downstream::push);
+            return (state, downstream) -> {
+                if (!downstream.isRejecting()) {
+                    state.forEach(downstream::push);
+                }
+            };
         }
 
         static class AppendOnlyCircularBuffer<T> {
@@ -90,7 +94,7 @@ public Integrator<ValueHolder<T>, T, T> integrator() {
         @Override
         public BiConsumer<ValueHolder<T>, Downstream<? super T>> finisher() {
             return (state, downstream) -> {
-                if (state.isSet) {
+                if (state.isSet && !downstream.isRejecting()) {
                     downstream.push(state.value);
                 }
             };