@Override
    public CompletableFuture<List<T>> collect(EventExecutor executor, SubscriptionOption... options) {
        requireNonNull(executor, "executor");
        requireNonNull(options, "options");

        if (!downstreamSubscriptionUpdater.compareAndSet(this, null, NOOP_SUBSCRIPTION)) {
            final SubscriptionImpl downstreamSubscription = this.downstreamSubscription;
            assert downstreamSubscription != null;
            final Subscriber<Object> subscriber = downstreamSubscription.subscriber();
            final Throwable cause = abortedOrLate(subscriber);
            final CompletableFuture<List<T>> collectingFuture = new CompletableFuture<>();
            collectingFuture.completeExceptionally(cause);
            return collectingFuture;
        }

        // An atomic operation on multiple subscribers will be handled by upstream.collect()
        final StreamMessage<T> upstream = this.upstream;
        if (upstream != null) {
            return upstream.collect(executor, options);
        }

        final CompletableFuture<List<T>> collectingFuture = new CompletableFuture<>();
        collectingExecutor = executor;
        collectionOptions = options;
        if (collectingFutureUpdater.compareAndSet(this, null, collectingFuture)) {
            final Throwable abortCause = this.abortCause;
            if (abortCause != null) {
                collectingFuture.completeExceptionally(abortCause);
            }
            return collectingFuture;
        } else {
            final StreamMessage<T> upstream0 = this.upstream;
            assert upstream0 != null;
            return upstream0.collect(executor, options);
        }
    }
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