@UnstableApi
    protected <U extends AggregatedHttpMessage> CompletableFuture<U> aggregate(AggregationOptions options) {
        requireNonNull(options, "options");

        if (!(this instanceof HttpMessage)) {
            // This API should be used internally. Should never reach here.
            throw new UnsupportedOperationException("aggregate() is only supported for " + HttpMessage.class);
        }

        final HttpMessage httpMessage = (HttpMessage) this;

        final RequestHeaders headers;
        if (httpMessage instanceof HttpRequest) {
            headers = ((HttpRequest) httpMessage).headers();
        } else {
            headers = null;
        }
        final ByteBufAllocator alloc = options.alloc();
        final SubscriptionOption[] subscriptionOptions = alloc != null ? POOLED_OBJECTS : EMPTY_OPTIONS;

        final CompletableFuture<?> aggregation = this.aggregation;
        if (aggregation != null) {
            return handleAggregation(options, aggregation);
        }

        if (!options.cacheResult()) {
            if (!aggregationUpdater.compareAndSet(this, null, NO_CACHE)) {
                final CompletableFuture<Object> aggregation0 = this.aggregation;
                assert aggregation0 != null;
                return handleAggregation(options, aggregation0);
            }
            EventExecutor executor = options.executor();
            if (executor == null) {
                executor = httpMessage.defaultSubscriberExecutor();
            }
            return httpMessage.collect(executor, subscriptionOptions)
                              .thenApply(objects -> aggregate(objects, headers, alloc));
        }

        final CompletableFuture<U> aggregationFuture = new CompletableFuture<>();
        if (!aggregationUpdater.compareAndSet(this, null, aggregationFuture)) {
            final CompletableFuture<?> aggregation0 = this.aggregation;
            assert aggregation0 != null;
            return handleAggregation(options, aggregation0);
        }
        // Propagate the result to `aggregation`
        EventExecutor executor = options.executor();
        if (executor == null) {
            executor = httpMessage.defaultSubscriberExecutor();
        }
        httpMessage.collect(executor, subscriptionOptions)
                   .thenApply(objects -> aggregate(objects, headers, alloc)).handle((res, cause) -> {
                       if (cause != null) {
                           cause = Exceptions.peel(cause);
                           aggregationFuture.completeExceptionally(cause);
                       } else {
                           //noinspection unchecked
                           aggregationFuture.complete((U) res);
                       }
                       return null;
                   });
        return aggregationFuture;
    }
Aggregates an {@link HttpMessage} into an {@link AggregatedHttpMessage} using
the specified {@link AggregationOptions}.

<p>Note that this method is added for internal usage. Therefore, you <strong>must not</strong> override
or call this method if you are not familiar with Armeria's internal implementation.
core/src/main/java/com/linecorp/armeria/common/stream/AggregationSupport.java