fix(pauseDemand): changes after review, to squash

Author: ozangunalp

documentation/docs/guides/controlling-demand.md

@@ -98,7 +98,10 @@ The `lateSubscription()` option delays the upstream subscription until the strea
 
 When a stream is paused, the operator stops requesting new items from upstream.
 However, items that were already requested (due to downstream demand) may still arrive.
-Buffer strategies control what happens to these in-flight items:
+Buffer strategies control what happens to these in-flight items.
+
+The `pauseDemand()` operator supports three buffer strategies: `BUFFER` (default), `DROP`, and `IGNORE`.
+Configuring any other strategy will throw an `IllegalArgumentException`.
 
 #### BUFFER strategy (default)
 
@@ -109,7 +112,7 @@ Already-requested items are buffered while paused and delivered when resumed:
 ```
 
 You can configure the buffer size:
-- `bufferSize(0)`: Unbounded buffer (uses default buffer size internally)
+- `bufferUnconditionally()`: Unbounded buffer
 - `bufferSize(n)`: Buffer up to `n` items, then fail with buffer overflow
 
 When the buffer overflows, the stream fails with an `IllegalStateException`.

documentation/src/test/java/guides/operators/PausingDemandTest.java

@@ -24,12 +24,12 @@ void basicPausing() {
         AssertSubscriber<Integer> sub = AssertSubscriber.create();
         Multi.createFrom().range(0, 100)
                 .pauseDemand().using(pauser)
-                // throttle the multi
+                // Throttle the multi
                 .onItem().call(i -> Uni.createFrom().nullItem()
                         .onItem().delayIt().by(Duration.ofMillis(10)))
                 .subscribe().withSubscriber(sub);
 
-        // request unbound
+        // Unbounded request
         sub.request(Long.MAX_VALUE);
         // Wait for some items
         await().untilAsserted(() -> assertThat(sub.getItems()).hasSizeGreaterThan(10));
@@ -180,7 +180,7 @@ void bufferManagement() {
                 .select().first(100)
                 .pauseDemand()
                 .bufferStrategy(BackPressureStrategy.BUFFER)
-                .bufferSize(0)  // Unbounded buffer
+                .bufferUnconditionally()  // Unbounded buffer
                 .using(pauser)
                 .subscribe().withSubscriber(AssertSubscriber.create(Long.MAX_VALUE));
 

implementation/revapi.json

@@ -51,12 +51,7 @@
     "criticality" : "highlight",
     "minSeverity" : "POTENTIALLY_BREAKING",
     "minCriticality" : "documented",
-    "differences" : [{
-      "code": "java.method.addedToInterface",
-      "new": "method io.smallrye.mutiny.groups.MultiDemandPausing<T> io.smallrye.mutiny.Multi<T>::pauseDemand()",
-      "justification": "add demand pauser to Multi"
-    }
-    ]
+    "differences" : [ ]
   }
 }, {
   "extension" : "revapi.reporter.json",

implementation/src/main/java/io/smallrye/mutiny/Multi.java

@@ -640,7 +640,7 @@ default Multi<ItemWithContext<T>> attachContext() {
     /**
      * Allows pausing and resuming demand propagation to upstream using a {@link io.smallrye.mutiny.subscription.DemandPauser}.
      * <p>
-     * Unlike cancellation which terminates the subscription, pausing temporarily suspends demand without unsubscribing.
+     * Unlike cancellation which terminates the subscription, temporarily pausing suspends the demand without unsubscribing.
      * This is useful for implementing flow control patterns where demand needs to be temporarily suspended based on
      * external conditions (e.g., downstream system availability, rate limiting, resource constraints).
      * <p>
@@ -666,7 +666,9 @@ default Multi<ItemWithContext<T>> attachContext() {
      * @see io.smallrye.mutiny.subscription.DemandPauser
      */
     @CheckReturnValue
-    MultiDemandPausing<T> pauseDemand();
+    default MultiDemandPausing<T> pauseDemand() {
+        throw new UnsupportedOperationException("Default method added to limit binary incompatibility");
+    }
 
     /**
      * Cap all downstream subscriber requests to a maximum value.

implementation/src/main/java/io/smallrye/mutiny/groups/MultiDemandPausing.java

@@ -1,7 +1,7 @@
 package io.smallrye.mutiny.groups;
 
 import static io.smallrye.mutiny.helpers.ParameterValidation.nonNull;
-import static io.smallrye.mutiny.helpers.ParameterValidation.positiveOrZero;
+import static io.smallrye.mutiny.helpers.ParameterValidation.positive;
 
 import io.smallrye.common.annotation.CheckReturnValue;
 import io.smallrye.mutiny.Multi;
@@ -29,6 +29,7 @@ public class MultiDemandPausing<T> {
     private boolean paused = false;
     private boolean lateSubscription = false;
     private int bufferSize = Infrastructure.getMultiOverflowDefaultBufferSize();
+    private boolean unbounded = false;
     private BackPressureStrategy bufferStrategy = BackPressureStrategy.BUFFER;
 
     public MultiDemandPausing(AbstractMulti<T> upstream) {
@@ -74,21 +75,31 @@ public MultiDemandPausing<T> lateSubscription(boolean lateSubscription) {
      * Sets the maximum buffer size for already-requested items when using {@link BackPressureStrategy#BUFFER}.
      * <p>
      * When the stream is paused, items that were already requested from upstream can be buffered.
-     * This parameter controls the maximum number of items to buffer:
-     * <ul>
-     * <li>{@code 0}: Unbounded buffer (uses default buffer size internally)</li>
-     * <li>{@code n > 0}: Buffer up to {@code n} items, then fail with {@link IllegalStateException}</li>
-     * </ul>
      * <p>
      * Note: The buffer only holds items that were already requested from upstream before pausing.
      * When paused, no new requests are issued to upstream.
      *
-     * @param bufferSize the maximum buffer size, must be non-negative
+     * @param bufferSize the maximum buffer size, must be positive
      * @return this configuration instance
      */
     @CheckReturnValue
     public MultiDemandPausing<T> bufferSize(int bufferSize) {
-        this.bufferSize = positiveOrZero(bufferSize, "bufferSize");
+        this.bufferSize = positive(bufferSize, "bufferSize");
+        this.unbounded = false;
+        return this;
+    }
+
+    /**
+     * Sets the buffer size for already-requested items to unbounded when using {@link BackPressureStrategy#BUFFER}.
+     * <p>
+     * When the stream is paused, items that were already requested from upstream can be buffered.
+     *
+     * @return this configuration instance
+     */
+    @CheckReturnValue
+    public MultiDemandPausing<T> bufferUnconditionally() {
+        this.bufferSize = Infrastructure.getMultiOverflowDefaultBufferSize();
+        this.unbounded = true;
         return this;
     }
 
@@ -126,7 +137,7 @@ public MultiDemandPausing<T> bufferStrategy(BackPressureStrategy bufferStrategy)
     public Multi<T> using(DemandPauser pauser) {
         DemandPauser p = nonNull(pauser, "pauser");
         MultiDemandPausingOp<T> pausingMulti = new MultiDemandPausingOp<>(upstream,
-                paused, lateSubscription, bufferSize, bufferStrategy);
+                paused, lateSubscription, bufferSize, unbounded, bufferStrategy);
         p.bind(pausingMulti);
         return Infrastructure.onMultiCreation(pausingMulti);
     }

implementation/src/main/java/io/smallrye/mutiny/operators/multi/MultiDemandPausingOp.java

@@ -3,13 +3,12 @@
 import java.util.Queue;
 import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
-import java.util.function.Supplier;
 
 import io.smallrye.mutiny.Multi;
 import io.smallrye.mutiny.helpers.Subscriptions;
 import io.smallrye.mutiny.helpers.queues.Queues;
-import io.smallrye.mutiny.infrastructure.Infrastructure;
 import io.smallrye.mutiny.operators.MultiOperator;
 import io.smallrye.mutiny.subscription.BackPressureStrategy;
 import io.smallrye.mutiny.subscription.MultiSubscriber;
@@ -36,14 +35,16 @@ public class MultiDemandPausingOp<T> extends MultiOperator<T, T> implements Paus
     private final AtomicBoolean subscribed = new AtomicBoolean();
     private final boolean lateSubscription;
     private final int bufferSize;
+    private final boolean unbounded;
     private final BackPressureStrategy backPressureStrategy;
 
     public MultiDemandPausingOp(Multi<T> upstream, boolean initiallyPaused, boolean lateSubscription, int bufferSize,
-            BackPressureStrategy backPressureStrategy) {
+            boolean unbounded, BackPressureStrategy backPressureStrategy) {
         super(upstream);
         this.paused = new AtomicBoolean(initiallyPaused);
         this.lateSubscription = lateSubscription;
         this.bufferSize = bufferSize;
+        this.unbounded = unbounded;
         this.backPressureStrategy = backPressureStrategy;
     }
 
@@ -104,19 +105,16 @@ private class PausableProcessor extends MultiOperatorProcessor<T, T> {
 
         private final AtomicLong demand = new AtomicLong();
         private final Queue<T> queue;
-        private final int maxBufferSize;
+        private final AtomicInteger strictBoundCounter = new AtomicInteger(0);
 
         PausableProcessor(MultiSubscriber<? super T> downstream) {
             super(downstream);
             // Determine if we need a queue based on strategy and buffer size
             if (backPressureStrategy == BackPressureStrategy.BUFFER) {
-                Supplier<Queue<T>> qs = bufferSize == 0 ? Queues.unbounded(Infrastructure.getMultiOverflowDefaultBufferSize())
-                        : Queues.get(bufferSize);
-                this.queue = qs.get();
+                this.queue = unbounded ? Queues.<T> unbounded(bufferSize).get() : Queues.<T> get(bufferSize).get();
             } else {
                 this.queue = null;
             }
-            this.maxBufferSize = bufferSize;
         }
 
         void resume() {
@@ -140,17 +138,22 @@ void drainQueue() {
             Queue<T> qe = queue;
             // Drain all buffered items - these were already requested from upstream
             // so we don't need to check downstream demand here
-            while (!qe.isEmpty() && !paused.get()) {
+            while (!paused.get()) {
                 T item = qe.poll();
                 if (item == null) {
+                    // queue empty
                     break;
                 }
+                if (!unbounded) {
+                    strictBoundCounter.decrementAndGet();
+                }
                 downstream.onItem(item);
             }
         }
 
         void clearQueue() {
             if (queue != null) {
+                strictBoundCounter.set(0);
                 queue.clear();
             }
         }
@@ -162,40 +165,19 @@ int queueSize() {
         @Override
         public void onItem(T item) {
             if (backPressureStrategy != BackPressureStrategy.IGNORE && paused.get()) {
+                if (backPressureStrategy == BackPressureStrategy.DROP) {
+                    return;
+                }
                 // When paused buffer items if necessary
-                if (!offerToQueue(item)) {
-                    handleOverflow();
+                if ((!unbounded && strictBoundCounter.getAndIncrement() >= bufferSize) || !queue.offer(item)) {
+                    // Buffer is full, throw exception
+                    onFailure(new IllegalStateException("Buffer overflow: cannot buffer more than " + bufferSize + " items"));
                 }
             } else {
                 super.onItem(item);
             }
         }
 
-        private void handleOverflow() {
-            switch (backPressureStrategy) {
-                case BUFFER:
-                    // Buffer is full, throw exception
-                    onFailure(new IllegalStateException(
-                            "Buffer overflow: cannot buffer more than " + maxBufferSize + " items"));
-                    return;
-                case DROP:
-                    // Drop the new item (do nothing)
-                    return;
-            }
-        }
-
-        private boolean offerToQueue(T item) {
-            if (queue == null) {
-                return false;
-            }
-            // If unbounded (maxBufferSize == 0), always succeed
-            if (maxBufferSize == 0) {
-                return queue.offer(item);
-            }
-            // For bounded queues, check size before offering
-            return queue.size() < maxBufferSize && queue.offer(item);
-        }
-
         @Override
         public void request(long numberOfItems) {
             if (numberOfItems <= 0) {
@@ -223,8 +205,21 @@ public void request(long numberOfItems) {
 
         @Override
         public void cancel() {
+            clearQueue();
             processor = null;
             super.cancel();
         }
+
+        @Override
+        public void onFailure(Throwable failure) {
+            clearQueue();
+            super.onFailure(failure);
+        }
+
+        @Override
+        public void onCompletion() {
+            clearQueue();
+            super.onCompletion();
+        }
     }
 }

implementation/src/main/java/io/smallrye/mutiny/subscription/PausableMulti.java

@@ -14,12 +14,12 @@
 public interface PausableMulti {
 
     /**
-     * Pauses demand propagation to upstream.
+     * Pauses demand propagation to the upstream.
      */
     void pause();
 
     /**
-     * Resumes demand propagation to upstream.
+     * Resumes demand propagation to the upstream.
      */
     void resume();