Fix Incorrect Calculation of Cumulative Demand in Duplicator (#6150)

Motivation:
The cumulative demand in a duplicator is mistakenly calculated, causing
it to reach the maximum value incorrectly.

Modifications:
- Corrected the calculation logic for cumulative demand in the
duplicator.

Result:
- The cumulative demand in a duplicator is correctly calculated.

Author: minwoox

core/src/main/java/com/linecorp/armeria/common/stream/DefaultStreamMessageDuplicator.java

@@ -340,7 +340,7 @@ void doRequestDemand(long cumulativeDemand) {
             }
 
             final long delta = cumulativeDemand - requestedDemand;
-            requestedDemand += delta;
+            requestedDemand = cumulativeDemand;
             upstreamSubscription.request(delta);
         }
 
@@ -658,18 +658,12 @@ public void request(long n) {
                 return;
             }
 
-            accumulateDemand(n);
+            cumulativeDemand = LongMath.saturatedAdd(cumulativeDemand, n);
             processor.requestDemand(cumulativeDemand);
 
             for (;;) {
                 final long oldDemand = demand;
-                final long newDemand;
-                if (oldDemand >= Long.MAX_VALUE - n) {
-                    newDemand = Long.MAX_VALUE;
-                } else {
-                    newDemand = oldDemand + n;
-                }
-
+                final long newDemand = LongMath.saturatedAdd(oldDemand, n);
                 if (demandUpdater.compareAndSet(this, oldDemand, newDemand)) {
                     if (oldDemand == 0) {
                         signal();
@@ -679,14 +673,6 @@ public void request(long n) {
             }
         }
 
-        private void accumulateDemand(long n) {
-            if (n == Long.MAX_VALUE || Long.MAX_VALUE - n >= cumulativeDemand) {
-                cumulativeDemand = Long.MAX_VALUE;
-            } else {
-                cumulativeDemand += n;
-            }
-        }
-
         void signal() {
             if (downstreamExecutor.inEventLoop()) {
                 doSignal();

core/src/main/java/com/linecorp/armeria/common/stream/DeferredStreamMessage.java

@@ -31,6 +31,8 @@
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 
+import com.google.common.math.LongMath;
+
 import com.linecorp.armeria.common.annotation.Nullable;
 import com.linecorp.armeria.common.annotation.UnstableApi;
 import com.linecorp.armeria.common.util.CompletionActions;
@@ -273,7 +275,7 @@ private void doRequest(long n) {
         if (upstreamSubscription != null) {
             upstreamSubscription.request(n);
         } else {
-            pendingDemand += n;
+            pendingDemand = LongMath.saturatedAdd(pendingDemand, n);
         }
     }
 

core/src/main/java/com/linecorp/armeria/internal/common/SplitHttpMessageSubscriber.java

@@ -234,6 +234,7 @@ private void onNext0(HttpData httpData) {
         if (!usePooledObject) {
             httpData = PooledObjects.copyAndClose(httpData);
         }
+        final Subscriber<? super HttpData> downstream = this.downstream;
         assert downstream != null;
         downstream.onNext(httpData);
     }

core/src/test/java/com/linecorp/armeria/common/stream/DuplicatorCumulativeDemandTest.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2025 LINE Corporation
+ *
+ * LINE Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+package com.linecorp.armeria.common.stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.junit.jupiter.api.Test;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+final class DuplicatorCumulativeDemandTest {
+
+    @Test
+    void cumulativeDemand() throws InterruptedException {
+        final AtomicLong demand = new AtomicLong();
+        final PublisherBasedStreamMessage<Integer> streamMessage = new PublisherBasedStreamMessage<>(
+                s -> s.onSubscribe(new Subscription() {
+                    @Override
+                    public void request(long n) {
+                        demand.addAndGet(n);
+                    }
+
+                    @Override
+                    public void cancel() {}
+                }));
+        streamMessage.toDuplicator().duplicate().subscribe(new Subscriber<Integer>() {
+            @Override
+            public void onSubscribe(Subscription s) {
+                s.request(1);
+            }
+
+            @Override
+            public void onNext(Integer integer) {}
+
+            @Override
+            public void onError(Throwable t) {}
+
+            @Override
+            public void onComplete() {}
+        });
+        await().untilAsserted(() -> assertThat(demand.get()).isEqualTo(1));
+        Thread.sleep(100);
+        // The demand is still 1.
+        assertThat(demand.get()).isEqualTo(1);
+    }
+}

core/src/test/java/com/linecorp/armeria/GracefulShutdownBuilderTest.java core/src/test/java/com/linecorp/armeria/server/GracefulShutdownBuilderTest.java

@@ -14,7 +14,7 @@
  * under the License.
  */
 
-package com.linecorp.armeria;
+package com.linecorp.armeria.server;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -25,7 +25,6 @@
 
 import com.linecorp.armeria.common.ShuttingDownException;
 import com.linecorp.armeria.internal.testing.AnticipatedException;
-import com.linecorp.armeria.server.GracefulShutdown;
 
 class GracefulShutdownBuilderTest {