@SuppressWarnings("unchecked")
    public static CompletableFuture<Object> collectFrom(Publisher<?> publisher, ServiceRequestContext ctx) {
        requireNonNull(publisher, "publisher");
        if (publisher instanceof StreamMessage) {
            final StreamMessage<?> stream = (StreamMessage<?>) publisher;
            ctx.whenRequestCancelling().thenAccept(cause -> stream.abort(cause));
            return (CompletableFuture<Object>) (CompletableFuture<?>) stream.collect();
        }
        final CompletableFuture<Object> future = new CompletableFuture<>();
        if (MONO_CLASS != null && MONO_CLASS.isAssignableFrom(publisher.getClass())) {
            publisher.subscribe(new CollectingSingleObjectSubscriber<>(future, ctx));
        } else {
            publisher.subscribe(new CollectingMultipleObjectsSubscriber<>(future, ctx));
        }
        return future;
    }
Collects objects published from the specified {@link Publisher}.

@param publisher publishes objects
@param ctx {@link ServiceRequestContext}
@return a {@link CompletableFuture} which will complete when all published objects are collected
core/src/main/java/com/linecorp/armeria/internal/common/util/ObjectCollectingUtil.java