protected final void delegate(StreamMessage<T> upstream) {
        requireNonNull(upstream, "upstream");

        if (!upstreamUpdater.compareAndSet(this, null, upstream)) {
            final IllegalStateException exception = new IllegalStateException("upstream set already");
            upstream.abort(exception);
            throw exception;
        }

        final Throwable abortCause = this.abortCause;
        if (abortCause != null) {
            upstream.abort(abortCause);
        }

        if (!collectingFutureUpdater.compareAndSet(this, null, NO_COLLECTING_FUTURE)) {
            assert collectingExecutor != null;
            assert collectionOptions != null;
            upstream.collect(collectingExecutor, collectionOptions).handle((result, cause) -> {
                final CompletableFuture<List<T>> collectingFuture = this.collectingFuture;
                assert collectingFuture != null;
                if (cause != null) {
                    collectingFuture.completeExceptionally(cause);
                } else {
                    // `collectingFuture` can be completed exceptionally by `abort()`
                    if (!collectingFuture.complete(result)) {
                        for (final T obj : result) {
                            PooledObjects.close(obj);
                        }
                    }
                }
                return null;
            });
        }

        if (!whenComplete().isDone()) {
            upstream.whenComplete().handle((unused, cause) -> {
                if (cause == null) {
                    whenComplete().complete(null);
                } else {
                    whenComplete().completeExceptionally(cause);
                }
                return null;
            }).exceptionally(CompletionActions::log);
        }

        safeOnSubscribeToUpstream();
    }
Sets the upstream {@link StreamMessage} which will actually publish the stream.

@throws IllegalStateException if the upstream 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