@UnstableApi
    static <T> StreamMessage<T> createStreamMessageFrom(
            CompletableFuture<? extends Publisher<? extends T>> future) {
        requireNonNull(future, "future");

        if (future.isDone()) {
            if (!future.isCompletedExceptionally()) {
                final Publisher<? extends T> publisher = future.getNow(null);
                return StreamMessage.of(publisher);
            }

            try {
                future.join();
                // Should never reach here.
                throw new Error();
            } catch (Throwable cause) {
                return StreamMessage.aborted(Exceptions.peel(cause));
            }
        }

        final DeferredStreamMessage<T> deferred = new DeferredStreamMessage<>();
        //noinspection unchecked
        deferred.delegateOnCompletion((CompletionStage<? extends Publisher<T>>) future);
        return deferred;
    }
Creates a new {@link StreamMessage} that delegates to the {@link StreamMessage} produced by the specified
{@link CompletionStage}. If the specified {@link CompletionStage} fails, the returned
{@link StreamMessage} will be closed with the same cause as well.

@param future the {@link CompletionStage} which will produce the actual {@link StreamMessage}
core/src/main/java/com/linecorp/armeria/common/stream/StreamMessageUtil.java