private void abort0(@Nullable Throwable cause) {
        if (state == State.CLEANUP) {
            return;
        }

        if (cause == null) {
            cause = AbortedStreamException.get();
        }

        SubscriptionImpl subscription = this.subscription;
        if (subscription == null) {
            final SubscriptionImpl newSubscription = new SubscriptionImpl(
                    this, AbortingSubscriber.get(cause), ImmediateEventExecutor.INSTANCE, EMPTY_OPTIONS);
            if (subscriptionUpdater.compareAndSet(this, null, newSubscription)) {
                // We don't need to invoke onSubscribe() for AbortingSubscriber because it's just a placeholder.
                invokedOnSubscribe = true;
                subscription = newSubscription;
            } else {
                subscription = this.subscription;
            }
        }
        assert subscription != null;
        final SubscriptionImpl abortedSubscription = subscription;

        if (setState(State.OPEN, State.CLEANUP)) {
            notifySubscriberOfCloseEvent(abortedSubscription, newCloseEvent(cause));
            return;
        }

        if (setState(State.CLOSED, State.CLEANUP)) {
            // close() or close(cause) has been called before cancel() or abort() is called.
            if (abortedSubscription.needsDirectInvocation()) {
                abort0(cause, abortedSubscription);
            } else {
                final Throwable finalCause = cause;
                abortedSubscription.executor().execute(() -> abort0(finalCause, abortedSubscription));
            }
        }
    }
Invoked whenever a new demand is requested.
@param n Newly requested demand
core/src/main/java/com/linecorp/armeria/common/stream/DefaultStreamMessage.java