@Override
        public void request(long n) {
            if (n <= 0) {
                final Throwable cause = new IllegalArgumentException(
                        "n: " + n + " (expected: > 0, see Reactive Streams specification rule 3.9)");
                unsubscribe(cause);
                return;
            }

            cumulativeDemand = LongMath.saturatedAdd(cumulativeDemand, n);
            processor.requestDemand(cumulativeDemand);

            for (;;) {
                final long oldDemand = demand;
                final long newDemand = LongMath.saturatedAdd(oldDemand, n);
                if (demandUpdater.compareAndSet(this, oldDemand, newDemand)) {
                    if (oldDemand == 0) {
                        signal();
                    }
                    break;
                }
            }
        }
{@link CancelledSubscriptionException} if cancelled. {@link Throwable} if aborted.
core/src/main/java/com/linecorp/armeria/common/stream/DefaultStreamMessageDuplicator.java