class AsyncConnectionPoolSupport {

    private AsyncConnectionPoolSupport() {
    }

    /**
     * Create and initialize asynchronously a new {@link BoundedAsyncPool} using the {@link Supplier}. Allocated instances are
     * wrapped and must not be returned with {@link AsyncPool#release(Object)}.
     * <p>
     * Since Lettuce 6, this method is blocking as it awaits pool initialization (creation of idle connections).Use
     * {@link #createBoundedObjectPoolAsync(Supplier, BoundedPoolConfig)} to obtain a {@link CompletionStage} for non-blocking
     * synchronization.
     *
     * @param connectionSupplier must not be {@code null}.
     * @param config must not be {@code null}.
     * @param <T> connection type.
     * @return the connection pool.
     */
    public static <T extends StatefulConnection<?, ?>> BoundedAsyncPool<T> createBoundedObjectPool(
            Supplier<CompletionStage<T>> connectionSupplier, BoundedPoolConfig config) {
        return createBoundedObjectPool(connectionSupplier, config, true);
    }

    /**
     * Create and initialize asynchronously a new {@link BoundedAsyncPool} using the {@link Supplier}.
     * <p>
     * Since Lettuce 6, this method is blocking as it awaits pool initialization (creation of idle connections).Use
     * {@link #createBoundedObjectPoolAsync(Supplier, BoundedPoolConfig, boolean)} to obtain a {@link CompletionStage} for
     * non-blocking synchronization.
     *
     * @param connectionSupplier must not be {@code null}.
     * @param config must not be {@code null}.
     * @param wrapConnections {@code false} to return direct connections that need to be returned to the pool using
     *        {@link AsyncPool#release(Object)}. {@code true} to return wrapped connection that are returned to the pool when
     *        invoking {@link StatefulConnection#close()}/{@link StatefulConnection#closeAsync()}.
     * @param <T> connection type.
     * @return the connection pool.
     */
    public static <T extends StatefulConnection<?, ?>> BoundedAsyncPool<T> createBoundedObjectPool(
            Supplier<CompletionStage<T>> connectionSupplier, BoundedPoolConfig config, boolean wrapConnections) {

        try {
            return createBoundedObjectPoolAsync(connectionSupplier, config, wrapConnections).toCompletableFuture().join();
        } catch (Exception e) {
            throw Exceptions.bubble(Exceptions.unwrap(e));
        }
    }

    /**
     * Create and initialize asynchronously a new {@link BoundedAsyncPool} using the {@link Supplier}. Allocated instances are
     * wrapped and must not be returned with {@link AsyncPool#release(Object)}.
     *
     * @param connectionSupplier must not be {@code null}.
     * @param config must not be {@code null}.
     * @param <T> connection type.
     * @return {@link CompletionStage} emitting the connection pool upon completion.
     * @since 5.3.3
     */
    public static <T extends StatefulConnection<?, ?>> CompletionStage<BoundedAsyncPool<T>> createBoundedObjectPoolAsync(
            Supplier<CompletionStage<T>> connectionSupplier, BoundedPoolConfig config) {
        return createBoundedObjectPoolAsync(connectionSupplier, config, true);
    }

    /**
     * Create and initialize asynchronously a new {@link BoundedAsyncPool} using the {@link Supplier}.
     *
     * @param connectionSupplier must not be {@code null}.
     * @param config must not be {@code null}.
     * @param wrapConnections {@code false} to return direct connections that need to be returned to the pool using
     *        {@link AsyncPool#release(Object)}. {@code true} to return wrapped connection that are returned to the pool when
     *        invoking {@link StatefulConnection#close()}/{@link StatefulConnection#closeAsync()}.
     * @param <T> connection type.
     * @return {@link CompletionStage} emitting the connection pool upon completion.
     * @since 5.3.3
     */
    public static <T extends StatefulConnection<?, ?>> CompletionStage<BoundedAsyncPool<T>> createBoundedObjectPoolAsync(
            Supplier<CompletionStage<T>> connectionSupplier, BoundedPoolConfig config, boolean wrapConnections) {

        BoundedAsyncPool<T> pool = doCreatePool(connectionSupplier, config, wrapConnections);
        CompletableFuture<BoundedAsyncPool<T>> future = new CompletableFuture<>();

        pool.createIdle().whenComplete((v, throwable) -> {

            if (throwable == null) {
                future.complete(pool);
            } else {
                pool.closeAsync().whenComplete((v1, throwable1) -> {
                    future.completeExceptionally(new RedisConnectionException("Could not create pool", throwable));
                });
            }
        });

        return future;
    }

    protected static <T extends StatefulConnection<?, ?>> BoundedAsyncPool<T> doCreatePool(
            Supplier<CompletionStage<T>> connectionSupplier, BoundedPoolConfig config, boolean wrapConnections) {<FILL_FUNCTION_BODY>}

    /**
     * @author Mark Paluch
     * @since 5.1
     */
    private static class RedisPooledObjectFactory<T extends StatefulConnection<?, ?>> implements AsyncObjectFactory<T> {

        private final Supplier<CompletionStage<T>> connectionSupplier;

        RedisPooledObjectFactory(Supplier<CompletionStage<T>> connectionSupplier) {
            this.connectionSupplier = connectionSupplier;
        }

        @Override
        public CompletableFuture<T> create() {
            return connectionSupplier.get().toCompletableFuture();
        }

        @Override
        public CompletableFuture<Void> destroy(T object) {
            return object.closeAsync();
        }

        @Override
        public CompletableFuture<Boolean> validate(T object) {
            return CompletableFuture.completedFuture(object.isOpen());
        }
    }

    private static class AsyncPoolWrapper<T> implements Origin<T> {

        private final AsyncPool<T> pool;

        AsyncPoolWrapper(AsyncPool<T> pool) {
            this.pool = pool;
        }

        @Override
        public void returnObject(T o) {
            returnObjectAsync(o).join();
        }

        @Override
        public CompletableFuture<Void> returnObjectAsync(T o) {
            return pool.release(o);
        }

    }

}


        LettuceAssert.notNull(connectionSupplier, "Connection supplier must not be null");
        LettuceAssert.notNull(config, "BoundedPoolConfig must not be null");

        AtomicReference<Origin<T>> poolRef = new AtomicReference<>();

        BoundedAsyncPool<T> pool = new BoundedAsyncPool<T>(new RedisPooledObjectFactory<T>(connectionSupplier), config, false) {

            @Override
            public CompletableFuture<T> acquire() {

                CompletableFuture<T> acquire = super.acquire();

                if (wrapConnections) {
                    return acquire.thenApply(it -> ConnectionWrapping.wrapConnection(it, poolRef.get()));
                }

                return acquire;
            }

            @Override
            @SuppressWarnings("unchecked")
            public CompletableFuture<Void> release(T object) {

                if (wrapConnections && object instanceof HasTargetConnection) {
                    return super.release((T) ((HasTargetConnection) object).getTargetConnection());
                }

                return super.release(object);
            }

        };

        poolRef.set(new AsyncPoolWrapper<>(pool));
        return pool;
    
<no_super_class>