@Override
    public final void abort(Throwable cause) {
        requireNonNull(cause, "cause");
        if (!abortCauseUpdater.compareAndSet(this, null, cause)) {
            return;
        }

        if (!subscribedToUpstreamUpdater.compareAndSet(this, 0, 1)) {
            // Already subscribed to upstream so just abort upstream.
            final StreamMessage<T> upstream = this.upstream;
            assert upstream != null;
            upstream.abort(cause);
            return;
        }

        // The upstream wouldn't be subscribed to by the downstream.
        // So we need to abort upstream and complete the downstream with the cause.
        // Upstream.abort() is called here and in delegate(StreamMessage<T> upstream) method.
        // It's safe to call upstream.abort() multiple times.

        // The downstream.onError() is called in downstreamOnError method if it's already set.
        // If it wasn't set yet, downstream.onError() will be called in
        // CancellableStreamMessage.failLateSubscriber().

        // Abort upstream if it's set.
        final StreamMessage<T> upstream = this.upstream;
        if (upstream != null) {
            upstream.abort(cause);
        }

        final SubscriptionImpl newSubscription = new SubscriptionImpl(
                this, AbortingSubscriber.get(cause), ImmediateEventExecutor.INSTANCE, EMPTY_OPTIONS);
        if (downstreamSubscriptionUpdater.compareAndSet(this, null, newSubscription)) {
            // Downstream wasn't set yet.
            whenComplete().completeExceptionally(cause);
            return;
        }

        // Downstream was already set but upstream wasn't set yet or wouldn't be subscribed by the downstream.

        //noinspection unchecked
        final CompletableFuture<List<?>> collectingFuture =
                (CompletableFuture<List<?>>) (CompletableFuture<?>) this.collectingFuture;
        if (collectingFuture != null && collectingFuture != NO_COLLECTING_FUTURE) {
            collectingFuture.completeExceptionally(cause);
        }

        final SubscriptionImpl downstreamSubscription = this.downstreamSubscription;
        assert downstreamSubscription != null;
        if (downstreamSubscription.needsDirectInvocation()) {
            downstreamOnError(cause, downstreamSubscription);
        } else {
            downstreamSubscription.executor().execute(
                    () -> downstreamOnError(cause, downstreamSubscription));
        }
    }
Closes the deferred stream without setting a delegate.

@throws IllegalStateException if the delegate has been set already or
                              if {@link #close()} or {@link #close(Throwable)} was called already.
core/src/main/java/com/linecorp/armeria/common/stream/DeferredStreamMessage.java