/*
 * Copyright (c) 2019-2025 VMware Inc. or its affiliates, All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package reactor.pool;

import java.io.Closeable;
import java.io.IOException;
import java.time.Clock;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Formatter;
import java.util.FormatterClosedException;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.assertj.core.api.Assertions;
import org.assertj.core.api.Assumptions;
import org.assertj.core.data.Offset;
import org.awaitility.Awaitility;
import org.hamcrest.CoreMatchers;
import org.jspecify.annotations.Nullable;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.params.provider.EnumSource;
import org.junit.jupiter.params.provider.MethodSource;

import org.reactivestreams.Subscription;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.Exceptions;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.SignalType;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;
import reactor.pool.InstrumentedPool.PoolMetrics;
import reactor.pool.TestUtils.ParameterizedTestWithName;
import reactor.pool.TestUtils.PoolableTest;
import reactor.scheduler.clock.SchedulerClock;
import reactor.test.StepVerifier;
import reactor.test.publisher.TestPublisher;
import reactor.test.scheduler.VirtualTimeScheduler;
import reactor.test.util.RaceTestUtils;
import reactor.test.util.TestLogger;
import reactor.util.Loggers;
import reactor.util.context.Context;

import static org.assertj.core.api.Assertions.*;
import static org.awaitility.Awaitility.await;
import static org.hamcrest.Matchers.is;

/**
 * @author Simon Baslé
 * @author Violeta Georgieva
 */
public class CommonPoolTest {

	enum PoolStyle {

		IDLE_LRU(true),
		IDLE_MRU(false);

		private final boolean isLru;

		PoolStyle(boolean isLru) {
			this.isLru = isLru;
		}

		public <T, CONF extends PoolConfig<T>> AbstractPool<T> apply(PoolBuilder<T, CONF> builder) {
			return (AbstractPool<T>) builder.idleResourceReuseOrder(isLru).buildPool();
		}
	}
	
	//keeping the methods below in case more flavors need to be added to CommonPoolTest

	static List<PoolStyle> allPools() {
		return Arrays.asList(PoolStyle.IDLE_LRU, PoolStyle.IDLE_MRU);
	}

	static List<PoolStyle> mruPools() {
		return Collections.singletonList(PoolStyle.IDLE_MRU);
	}

	static List<PoolStyle> lruPools() {
		return Collections.singletonList(PoolStyle.IDLE_LRU);
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void smokeTestFifo(PoolStyle configAdjuster) throws InterruptedException {
		AtomicInteger newCount = new AtomicInteger();

		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				//default maxUse is 5, but this test relies on it being 2
				.from(Mono.defer(() -> Mono.just(new PoolableTest(newCount.incrementAndGet(), 2))))
				.sizeBetween(2, 3)
				.releaseHandler(pt -> Mono.fromRunnable(pt::clean))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());

		Pool<PoolableTest> pool = configAdjuster.apply(builder);
		pool.warmup().block();

		List<PooledRef<PoolableTest>> acquired1 = new ArrayList<>();
		pool.acquire().subscribe(acquired1::add);
		pool.acquire().subscribe(acquired1::add);
		pool.acquire().subscribe(acquired1::add);
		List<PooledRef<PoolableTest>> acquired2 = new ArrayList<>();
		pool.acquire().subscribe(acquired2::add);
		pool.acquire().subscribe(acquired2::add);
		pool.acquire().subscribe(acquired2::add);
		List<PooledRef<PoolableTest>> acquired3 = new ArrayList<>();
		pool.acquire().subscribe(acquired3::add);
		pool.acquire().subscribe(acquired3::add);
		pool.acquire().subscribe(acquired3::add);

		assertThat(acquired1).hasSize(3);
		assertThat(acquired2).isEmpty();
		assertThat(acquired3).isEmpty();

		Thread.sleep(1000);
		for (PooledRef<PoolableTest> slot : acquired1) {
			slot.release().block();
		}
		assertThat(acquired2).hasSize(3);
		assertThat(acquired3).isEmpty();

		Thread.sleep(1000);
		for (PooledRef<PoolableTest> slot : acquired2) {
			slot.release().block();
		}
		assertThat(acquired3).hasSize(3);

		List<PoolableTest> poolables1 = acquired1.stream().map(PooledRef::poolable).collect(Collectors.toList());
		List<PoolableTest> poolables2 = acquired2.stream().map(PooledRef::poolable).collect(Collectors.toList());
		assertThat(poolables1)
				.as("poolable1 and 2 all used up")
				.hasSameElementsAs(poolables2)
				.allMatch(poolable -> poolable.usedUp == 2);

		assertThat(acquired3)
				.as("acquired3 all new")
				.allSatisfy(slot -> assertThat(slot.poolable().usedUp).isZero());
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void smokeTestInScopeFifo(PoolStyle configAdjuster) {
		AtomicInteger newCount = new AtomicInteger();

		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				//default maxUse is 5, but this test relies on it being 2
				.from(Mono.defer(() -> Mono.just(new PoolableTest(newCount.incrementAndGet(), 2))))
				.sizeBetween(2, 3)
				.releaseHandler(pt -> Mono.fromRunnable(pt::clean))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());
		Pool<PoolableTest> pool = configAdjuster.apply(builder);
		pool.warmup().block();

		TestPublisher<Integer> trigger1 = TestPublisher.create();
		TestPublisher<Integer> trigger2 = TestPublisher.create();
		TestPublisher<Integer> trigger3 = TestPublisher.create();

		List<PoolableTest> acquired1 = new ArrayList<>();

		Mono.when(
				pool.withPoolable(poolable -> Mono.just(poolable).doOnNext(acquired1::add).delayUntil(__ -> trigger1)),
				pool.withPoolable(poolable -> Mono.just(poolable).doOnNext(acquired1::add).delayUntil(__ -> trigger1)),
				pool.withPoolable(poolable -> Mono.just(poolable).doOnNext(acquired1::add).delayUntil(__ -> trigger1))
		).subscribe();

		List<PoolableTest> acquired2 = new ArrayList<>();
		Mono.when(
				pool.withPoolable(poolable -> Mono.just(poolable).doOnNext(acquired2::add).delayUntil(__ -> trigger2)),
				pool.withPoolable(poolable -> Mono.just(poolable).doOnNext(acquired2::add).delayUntil(__ -> trigger2)),
				pool.withPoolable(poolable -> Mono.just(poolable).doOnNext(acquired2::add).delayUntil(__ -> trigger2))
		).subscribe();

		List<PoolableTest> acquired3 = new ArrayList<>();
		Mono.when(
				pool.withPoolable(poolable -> Mono.just(poolable).doOnNext(acquired3::add).delayUntil(__ -> trigger3)),
				pool.withPoolable(poolable -> Mono.just(poolable).doOnNext(acquired3::add).delayUntil(__ -> trigger3)),
				pool.withPoolable(poolable -> Mono.just(poolable).doOnNext(acquired3::add).delayUntil(__ -> trigger3))
		).subscribe();

		assertThat(acquired1).as("first batch not pending").hasSize(3);
		assertThat(acquired2).as("second and third pending").hasSameSizeAs(acquired3).isEmpty();

		trigger1.emit(1);

		assertThat(acquired2).as("batch2 after trigger1").hasSize(3);
		assertThat(acquired3).as("batch3 after trigger1").isEmpty();

		trigger2.emit(1);

		assertThat(acquired3).as("batch3 after trigger2").hasSize(3);
		assertThat(newCount).as("allocated total").hasValue(6);

		assertThat(acquired1)
				.as("acquired1/2 all used up")
				.hasSameElementsAs(acquired2)
				.allSatisfy(elem -> assertThat(elem.usedUp).isEqualTo(2));

		assertThat(acquired3)
				.as("acquired3 all new (released once)")
				.allSatisfy(elem -> assertThat(elem.usedUp).isZero());
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void smokeTestAsyncFifo(PoolStyle configAdjuster) throws InterruptedException {
		AtomicInteger newCount = new AtomicInteger();

		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				//default maxUse is 5, but this test relies on it being 2
				.from(Mono.defer(() -> Mono.just(new PoolableTest(newCount.incrementAndGet(), 2)))
				          .subscribeOn(Schedulers.newParallel("poolable test allocator")))
				.sizeBetween(2, 3)
				.releaseHandler(pt -> Mono.fromRunnable(pt::clean))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());

		Pool<PoolableTest> pool = configAdjuster.apply(builder);
		pool.warmup().block();

		List<PooledRef<PoolableTest>> acquired1 = new CopyOnWriteArrayList<>();
		CountDownLatch latch1 = new CountDownLatch(3);
		pool.acquire().subscribe(acquired1::add, Throwable::printStackTrace, latch1::countDown);
		pool.acquire().subscribe(acquired1::add, Throwable::printStackTrace, latch1::countDown);
		pool.acquire().subscribe(acquired1::add, Throwable::printStackTrace, latch1::countDown);

		List<PooledRef<PoolableTest>> acquired2 = new CopyOnWriteArrayList<>();
		pool.acquire().subscribe(acquired2::add);
		pool.acquire().subscribe(acquired2::add);
		pool.acquire().subscribe(acquired2::add);

		List<PooledRef<PoolableTest>> acquired3 = new CopyOnWriteArrayList<>();
		CountDownLatch latch3 = new CountDownLatch(3);
		pool.acquire().subscribe(acquired3::add, Throwable::printStackTrace, latch3::countDown);
		pool.acquire().subscribe(acquired3::add, Throwable::printStackTrace, latch3::countDown);
		pool.acquire().subscribe(acquired3::add, Throwable::printStackTrace, latch3::countDown);

		if (!latch1.await(1, TimeUnit.SECONDS)) { //wait for creation of max elements
			fail("not enough elements created initially, missing " + latch1.getCount());
		}
		assertThat(acquired1).hasSize(3);
		assertThat(acquired2).isEmpty();
		assertThat(acquired3).isEmpty();

		Thread.sleep(1000);
		for (PooledRef<PoolableTest> slot : acquired1) {
			slot.release().block();
		}
		assertThat(acquired2).hasSize(3);
		assertThat(acquired3).isEmpty();

		Thread.sleep(1000);
		for (PooledRef<PoolableTest> slot : acquired2) {
			slot.release().block();
		}

		if (latch3.await(5, TimeUnit.SECONDS)) { //wait for the re-creation of max elements

			assertThat(acquired3).hasSize(3);

			List<PoolableTest> poolables1 = acquired1.stream().map(PooledRef::poolable).collect(Collectors.toList());
			List<PoolableTest> poolables2 = acquired2.stream().map(PooledRef::poolable).collect(Collectors.toList());

			assertThat(poolables1)
					.as("poolables 1 and 2 all used up")
					.hasSameElementsAs(poolables2)
					.allSatisfy(poolable -> assertThat(poolable.usedUp).isEqualTo(2));

			assertThat(acquired3)
					.as("acquired3 all new")
					.allSatisfy(slot -> {
						assertThat(slot).as("acquire3 slot").isNotNull();
						assertThat(slot.poolable()).as("acquire3 poolable").isNotNull();
						assertThat(slot.poolable().usedUp).as("acquire3 usedUp").isZero();
					});
		}
		else {
			fail("not enough new elements generated, missing " + latch3.getCount());
		}
	}

	@ParameterizedTestWithName
	@MethodSource("mruPools")
	void smokeTestMruIdle(PoolStyle configAdjuster) {
		AtomicInteger newCount = new AtomicInteger();
		PoolBuilder<PoolableTest, ?> builder =
				PoolBuilder.from(Mono.fromSupplier(() -> new PoolableTest(newCount.incrementAndGet())))
				           .sizeBetween(0, 3)
				           .releaseHandler(pt -> Mono.fromRunnable(pt::clean))
				           .evictionPredicate((value, metadata) -> !value.isHealthy());
		AbstractPool<PoolableTest> pool = configAdjuster.apply(builder);

		PooledRef<PoolableTest> ref1 = pool.acquire().block();
		assertThat(ref1).isNotNull();
		PooledRef<PoolableTest> ref2 = pool.acquire().block();
		assertThat(ref2).isNotNull();
		PooledRef<PoolableTest> ref3 = pool.acquire().block();
		assertThat(ref3).isNotNull();

		ref2.release().block();
		ref1.release().block();
		ref3.release().block();

		assertThat(pool.idleSize()).as("pool fully idle").isEqualTo(3);

		pool.acquire()
		    .as(StepVerifier::create)
		    .assertNext(ref -> assertThat(ref.poolable().id).as("MRU first call returns ref3").isEqualTo(3))
		    .verifyComplete();

		pool.acquire()
		    .as(StepVerifier::create)
		    .assertNext(ref -> assertThat(ref.poolable().id).as("MRU second call returns ref1").isEqualTo(1))
		    .verifyComplete();

		assertThat(pool.metrics().idleSize()).as("idleSize after 2 MRU calls").isOne();
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void firstAcquireCausesWarmupWithMinSize(PoolStyle configAdjuster)
			throws InterruptedException {
		AtomicInteger allocationCount = new AtomicInteger();

		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				.from(Mono.fromCallable(() -> {
					int id = allocationCount.incrementAndGet();
					return new PoolableTest(id);
				}))
				.sizeBetween(4, 8)
				.releaseHandler(p -> Mono.fromRunnable(p::clean));

		final Pool<PoolableTest> pool = configAdjuster.apply(builder);
		//notice no explicit warmup in this one
		assertThat(pool).isInstanceOf(InstrumentedPool.class);
		final PoolMetrics metrics = ((InstrumentedPool) pool).metrics();

		assertThat(metrics.allocatedSize()).as("constructor allocated").isEqualTo(0);

		final PooledRef<PoolableTest> firstAcquire = pool.acquire().block();
		assertThat(firstAcquire).isNotNull();

		assertThat(metrics.allocatedSize()).as("first acquire allocated").isEqualTo(4);
		assertThat(metrics.idleSize()).as("warmup idle").isEqualTo(3);

		final PooledRef<PoolableTest> secondAcquire = pool.acquire().block();
		assertThat(secondAcquire).isNotNull();

		assertThat(metrics.allocatedSize()).as("second acquire allocated").isEqualTo(4);
		assertThat(metrics.idleSize()).as("second acquire idle").isEqualTo(2);

		//give time for an unexpected warmup to take place so that the last assertion is valid
		Thread.sleep(100);
		assertThat(allocationCount).as("total allocations").hasValue(4);
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void destroyBelowMinSizeThreshold(PoolStyle configAdjuster)
			throws InterruptedException {
		AtomicInteger allocationCount = new AtomicInteger();

		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				.from(Mono.fromCallable(() -> {
					int id = allocationCount.incrementAndGet();
					return new PoolableTest(id);
				}))
				.sizeBetween(3, 4)
				.releaseHandler(p -> Mono.fromRunnable(p::clean));

		final Pool<PoolableTest> pool = configAdjuster.apply(builder);
		assertThat(pool).isInstanceOf(InstrumentedPool.class);
		final PoolMetrics metrics = ((InstrumentedPool) pool).metrics();

		final PooledRef<PoolableTest> ref1 = pool.acquire().block();
		assertThat(ref1).isNotNull();
		final PooledRef<PoolableTest> ref2 = pool.acquire().block();
		assertThat(ref2).isNotNull();
		final PooledRef<PoolableTest> ref3 = pool.acquire().block();
		assertThat(ref3).isNotNull();
		final PooledRef<PoolableTest> ref4 = pool.acquire().block();
		assertThat(ref4).isNotNull();

		assertThat(metrics.allocatedSize()).as("initial allocated").isEqualTo(4);
		assertThat(metrics.idleSize()).as("initial idle").isEqualTo(0);

		ref1.invalidate().block();
		ref2.invalidate().block();
		ref3.invalidate().block();

		assertThat(metrics.allocatedSize()).as("3 releases: allocated").isEqualTo(1);
		assertThat(metrics.idleSize()).as("3 releases: idle").isEqualTo(0);

		PooledRef<PoolableTest> ref5 = pool.acquire().block();
		assertThat(ref5).isNotNull();

		assertThat(metrics.allocatedSize()).as("second acquire allocated").isEqualTo(3); //1 borrowed, 1 acquiring, 1 extra matches min of 3
		assertThat(metrics.idleSize()).as("second acquire idle").isEqualTo(1); //only 1 extra created this time

		//give time for an unexpected warmup to take place so that the last assertion is valid
		Thread.sleep(100);
		assertThat(allocationCount).as("total allocations").hasValue(6);
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void returnedNotReleasedIfBorrowerCancelledEarly(PoolStyle configAdjuster) {
		AtomicInteger releasedCount = new AtomicInteger();

		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				.from(Mono.fromCallable(PoolableTest::new))
				.sizeBetween(1, 1)
				.releaseHandler(poolableTest -> Mono.fromRunnable(() -> {
					poolableTest.clean();
					releasedCount.incrementAndGet();
				}))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());

		Pool<PoolableTest> pool = configAdjuster.apply(builder);
		pool.warmup().block();

		//acquire the only element and hold on to it
		PooledRef<PoolableTest> slot = pool.acquire().block();
		assertThat(slot).isNotNull();

		pool.acquire().subscribe().dispose();

		assertThat(releasedCount).as("before returning").hasValue(0);

		//release the element, which should only mark it once, as the pending acquire should not be visible anymore
		slot.release().block();

		assertThat(releasedCount).as("after returning").hasValue(1);
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void fixedPoolReplenishOnDestroy(PoolStyle configAdjuster) throws Exception{
		AtomicInteger releasedCount = new AtomicInteger();
		AtomicInteger destroyedCount = new AtomicInteger();

		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				.from(Mono.fromCallable(PoolableTest::new))
				.sizeBetween(1, 1)
				.destroyHandler(poolableTest -> Mono.fromRunnable(() -> {
					poolableTest.clean();
					releasedCount.incrementAndGet();
				}))
				.releaseHandler(poolableTest -> Mono.fromRunnable(() -> {
					poolableTest.clean();
					destroyedCount.incrementAndGet();
				}))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());

		Pool<PoolableTest> pool = configAdjuster.apply(builder);
		pool.warmup().block();

		//acquire the only element
		PooledRef<PoolableTest> slot = pool.acquire().block();
		AtomicReference<PooledRef<PoolableTest>> acquired = new AtomicReference<>();
		CountDownLatch latch = new CountDownLatch(1);
		pool.acquire().subscribe(p -> {
			acquired.set(p);
			latch.countDown();
		});
		assertThat(slot).isNotNull();

		slot.invalidate().block();

		assertThat(releasedCount).as("before returning").hasValue(1);
		assertThat(destroyedCount).as("before returning").hasValue(0);

		//release the element, which should forward to the cancelled second acquire, itself also cleaning
		latch.await(5, TimeUnit.SECONDS);
		assertThat(acquired.get()).isNotNull();

		acquired.get().release().block();

		assertThat(destroyedCount).as("after returning").hasValue(1);
		assertThat(releasedCount).as("after returning").hasValue(1);
	}


	@ParameterizedTestWithName
	@MethodSource("allPools")
	void returnedNotReleasedIfBorrowerInScopeCancelledEarly(PoolStyle configAdjuster) {
		AtomicInteger releasedCount = new AtomicInteger();
		PoolBuilder<PoolableTest, ?> builder =
				PoolBuilder.from(Mono.fromCallable(PoolableTest::new))
				           .sizeBetween(1, 1)
				           .releaseHandler(poolableTest -> Mono.fromRunnable(() -> {
					           poolableTest.clean();
					           releasedCount.incrementAndGet();
				           }))
				           .evictionPredicate((poolable, metadata) -> !poolable.isHealthy());
		Pool<PoolableTest> pool = configAdjuster.apply(builder);
		pool.warmup().block();

		//acquire the only element and hold on to it
		PooledRef<PoolableTest> slot = pool.acquire().block();
		assertThat(slot).isNotNull();

		pool.withPoolable(Mono::just).subscribe().dispose();

		assertThat(releasedCount).as("before returning").hasValue(0);

		//release the element after the second acquire has been cancelled, so it should never "see" it
		slot.release().block();

		assertThat(releasedCount).as("after returning").hasValue(1);
	}


	@ParameterizedTestWithName
	@MethodSource("allPools")
	void allocatedReleasedIfBorrowerCancelled(PoolStyle configAdjuster) {
		Scheduler scheduler = Schedulers.newParallel("poolable test allocator");
		AtomicInteger newCount = new AtomicInteger();
		AtomicInteger releasedCount = new AtomicInteger();

		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				.from(Mono.defer(() -> Mono.delay(Duration.ofMillis(50)).thenReturn(new PoolableTest(newCount.incrementAndGet())))
				          .subscribeOn(scheduler))
				.sizeBetween(0, 1)
				.releaseHandler(poolableTest -> Mono.fromRunnable(() -> {
					poolableTest.clean();
					releasedCount.incrementAndGet();
				}))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());

		Pool<PoolableTest> pool = configAdjuster.apply(builder);

		//acquire the only element and immediately dispose
		pool.acquire().subscribe().dispose();

		//release due to cancel is async, give it a bit of time
		await()
				.atMost(100, TimeUnit.MILLISECONDS)
				.with().pollInterval(10, TimeUnit.MILLISECONDS)
				.untilAsserted(
						() -> assertThat(releasedCount).as("released").hasValue(1));

		assertThat(newCount).as("created").hasValue(1);
	}


	@ParameterizedTestWithName
	@MethodSource("allPools")
	void allocatedReleasedIfBorrowerInScopeCancelled(PoolStyle configAdjuster) {
		Scheduler scheduler = Schedulers.newParallel("poolable test allocator");
		AtomicInteger newCount = new AtomicInteger();
		AtomicInteger releasedCount = new AtomicInteger();

		PoolBuilder<PoolableTest, ?> builder =
				PoolBuilder.from(Mono.defer(() -> Mono.delay(Duration.ofMillis(50)).thenReturn(new PoolableTest(newCount.incrementAndGet())))
				                     .subscribeOn(scheduler))
				           .sizeBetween(0, 1)
				           .releaseHandler(poolableTest -> Mono.fromRunnable(() -> {
					           poolableTest.clean();
					           releasedCount.incrementAndGet();
				           }))
				           .evictionPredicate((poolable, metadata) -> !poolable.isHealthy());
		Pool<PoolableTest> pool = configAdjuster.apply(builder);

		//acquire the only element and immediately dispose
		pool.withPoolable(Mono::just).subscribe().dispose();

		//release due to cancel is async, give it a bit of time
		await()
				.atMost(100, TimeUnit.MILLISECONDS)
				.with().pollInterval(10, TimeUnit.MILLISECONDS)
				.untilAsserted(
						() -> assertThat(releasedCount).as("released").hasValue(1));

		assertThat(newCount).as("created").hasValue(1);
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void pendingLimitSync(PoolStyle configAdjuster) {
		AtomicInteger allocatorCount = new AtomicInteger();
		Disposable.Composite composite = Disposables.composite();

		try {
			PoolBuilder<Integer, ?> builder = PoolBuilder.from(Mono.fromCallable(allocatorCount::incrementAndGet))
			                                             .sizeBetween(1, 1)
			                                             .maxPendingAcquire(1);
			AbstractPool<Integer> pool = configAdjuster.apply(builder);
			pool.warmup().block();
			PooledRef<Integer> hold = pool.acquire().block();
			assertThat(hold).isNotNull();

			AtomicReference<Throwable> error = new AtomicReference<>();
			AtomicInteger errorCount = new AtomicInteger();
			AtomicInteger otherTerminationCount = new AtomicInteger();

			for (int i = 0; i < 2; i++) {
				composite.add(
						pool.acquire()
						    .doFinally(fin -> {
							    if (SignalType.ON_ERROR == fin) errorCount.incrementAndGet();
							    else otherTerminationCount.incrementAndGet();
						    })
						    .doOnError(error::set)
						    .subscribe()
				);
			}

			assertThat(pool.pendingAcquireSize()).as("pending counter limited to 1").isEqualTo(1);

			assertThat(errorCount).as("immediate error of extraneous pending").hasValue(1);
			assertThat(otherTerminationCount).as("no other immediate termination").hasValue(0);
			assertThat(error.get()).as("extraneous pending error")
			                       .isInstanceOf(PoolAcquirePendingLimitException.class)
			                       .hasMessage("Pending acquire queue has reached its maximum size of 1");

			hold.release().block();

			assertThat(errorCount).as("error count stable after release").hasValue(1);
			assertThat(otherTerminationCount).as("pending succeeds after release").hasValue(1);
		}
		finally {
			composite.dispose();
		}
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void pendingLimitAsync(PoolStyle configAdjuster) {
		AtomicInteger allocatorCount = new AtomicInteger();
		final Disposable.Composite composite = Disposables.composite();

		try {
			PoolBuilder<Integer, ?> builder = PoolBuilder.from(Mono.fromCallable(allocatorCount::incrementAndGet))
			                                             .sizeBetween(1, 1)
			                                             .maxPendingAcquire(1);
			AbstractPool<Integer> pool = configAdjuster.apply(builder);
			pool.warmup().block();
			PooledRef<Integer> hold = pool.acquire().block();
			assertThat(hold).isNotNull();

			AtomicReference<Throwable> error = new AtomicReference<>();
			AtomicInteger errorCount = new AtomicInteger();
			AtomicInteger otherTerminationCount = new AtomicInteger();

			final Runnable runnable = () -> composite.add(
					pool.acquire()
					    .doFinally(fin -> {
					    	if (SignalType.ON_ERROR == fin) errorCount.incrementAndGet();
					    	else otherTerminationCount.incrementAndGet();
					    })
					    .doOnError(error::set)
					    .subscribe(v -> {}, e -> {})
			);
			RaceTestUtils.race(runnable, runnable);

			assertThat(pool.pendingAcquireSize()).as("pending counter limited to 1").isEqualTo(1);

			assertThat(errorCount).as("immediate error of extraneous pending").hasValue(1);
			assertThat(otherTerminationCount).as("no other immediate termination").hasValue(0);
			assertThat(error.get()).as("extraneous pending error")
			                       .isInstanceOf(PoolAcquirePendingLimitException.class)
			                       .hasMessage("Pending acquire queue has reached its maximum size of 1");

			hold.release().block();

			assertThat(errorCount).as("error count stable after release").hasValue(1);
			assertThat(otherTerminationCount).as("pending succeeds after release").hasValue(1);
		}
		finally {
			composite.dispose();
		}
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void cleanerFunctionError(PoolStyle configAdjuster) {
		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				.from(Mono.fromCallable(PoolableTest::new))
				.sizeBetween(0, 1)
				.releaseHandler(poolableTest -> Mono.fromRunnable(() -> {
					poolableTest.clean();
					throw new IllegalStateException("boom");
				}))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());

		Pool<PoolableTest> pool = configAdjuster.apply(builder);

		PooledRef<PoolableTest> slot = pool.acquire().block();

		assertThat(slot).isNotNull();

		StepVerifier.create(slot.release())
		            .verifyErrorMessage("boom");
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void cleanerFunctionErrorDiscards(PoolStyle configAdjuster) {
		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				.from(Mono.fromCallable(PoolableTest::new))
				.sizeBetween(0, 1)
				.releaseHandler(poolableTest -> Mono.fromRunnable(() -> {
					poolableTest.clean();
					throw new IllegalStateException("boom");
				}))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());

		Pool<PoolableTest> pool = configAdjuster.apply(builder);

		PooledRef<PoolableTest> slot = pool.acquire().block();

		assertThat(slot).isNotNull();

		StepVerifier.create(slot.release())
		            .verifyErrorMessage("boom");

		assertThat(slot.poolable().discarded).as("discarded despite cleaner error").isEqualTo(1);
	}


	@ParameterizedTestWithName
	@MethodSource("allPools")
	void disposingPoolDisposesElements(PoolStyle configAdjuster) {
		AtomicInteger cleanerCount = new AtomicInteger();

		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				.from(Mono.fromCallable(PoolableTest::new))
				.sizeBetween(0, 3)
				.releaseHandler(p -> Mono.fromRunnable(cleanerCount::incrementAndGet))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());

		AbstractPool<PoolableTest> pool = configAdjuster.apply(builder);

		PoolableTest pt1 = new PoolableTest(1);
		PoolableTest pt2 = new PoolableTest(2);
		PoolableTest pt3 = new PoolableTest(3);

		pool.elementOffer(pt1);
		pool.elementOffer(pt2);
		pool.elementOffer(pt3);
		//important: acquire the permits to simulate the proper creation of resources
		assertThat(pool.poolConfig.allocationStrategy().getPermits(3)).as("permits granted").isEqualTo(3);

		//this releases the permits acquired above
		pool.dispose();

		assertThat(pool.idleSize()).as("idleSize").isZero();
		assertThat(cleanerCount).as("recycled elements").hasValue(0);
		assertThat(pt1.isDisposed()).as("pt1 disposed").isTrue();
		assertThat(pt2.isDisposed()).as("pt2 disposed").isTrue();
		assertThat(pt3.isDisposed()).as("pt3 disposed").isTrue();
		assertThat(pool.poolConfig.allocationStrategy().estimatePermitCount()).as("permits available post dispose").isEqualTo(3);
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void disposingPoolFailsPendingBorrowers(PoolStyle configAdjuster) {
		AtomicInteger cleanerCount = new AtomicInteger();
		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				.from(Mono.fromCallable(PoolableTest::new))
				.sizeBetween(3, 3)
				.releaseHandler(p -> Mono.fromRunnable(cleanerCount::incrementAndGet))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());

		AbstractPool<PoolableTest> pool = configAdjuster.apply(builder);
		pool.warmup().block();

		PooledRef<PoolableTest> slot1 = pool.acquire().block();
		PooledRef<PoolableTest> slot2 = pool.acquire().block();
		PooledRef<PoolableTest> slot3 = pool.acquire().block();
		assertThat(slot1).as("slot1").isNotNull();
		assertThat(slot2).as("slot2").isNotNull();
		assertThat(slot3).as("slot3").isNotNull();

		PoolableTest acquired1 = slot1.poolable();
		PoolableTest acquired2 = slot2.poolable();
		PoolableTest acquired3 = slot3.poolable();


		AtomicReference<Throwable> borrowerError = new AtomicReference<>();
		Mono<PooledRef<PoolableTest>> pendingBorrower = pool.acquire();
		pendingBorrower.subscribe(v -> fail("unexpected value " + v),
				borrowerError::set);

		pool.dispose();

		assertThat(pool.idleSize()).as("idleSize").isZero();
		assertThat(cleanerCount).as("recycled elements").hasValue(0);
		assertThat(acquired1.isDisposed()).as("acquired1 held").isFalse();
		assertThat(acquired2.isDisposed()).as("acquired2 held").isFalse();
		assertThat(acquired3.isDisposed()).as("acquired3 held").isFalse();
		assertThat(borrowerError.get()).hasMessage("Pool has been shut down");
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void doAcquireNotCalledIfBorrowerInScopeCancelledEarly(PoolStyle configAdjuster) {
		AtomicInteger allocator = new AtomicInteger();
		PoolBuilder<Integer, ?> builder = PoolBuilder.from(Mono.fromCallable(allocator::incrementAndGet));
		Pool<Integer> pool = configAdjuster.apply(builder);

		assertThat(allocator).as("before invoking acquire").hasValue(0);

		// Borrower is in state cancelled before the actual acquisition
		pool.acquire().doOnSubscribe(Subscription::cancel).subscribe();

		assertThat(allocator).as("after invoking acquire").hasValue(0);
	}

	@ParameterizedTestWithName
	@MethodSource("allPools")
	void releasingToDisposedPoolDisposesElement(PoolStyle configAdjuster) {
		AtomicInteger cleanerCount = new AtomicInteger();
		PoolBuilder<PoolableTest, ?> builder = PoolBuilder
				.from(Mono.fromCallable(PoolableTest::new))
				.sizeBetween(3, 3)
				.releaseHandler(p -> Mono.fromRunnable(cleanerCount::incrementAndGet))
				.evictionPredicate((poolable, metadata) -> !poolable.isHealthy());
		AbstractPool<PoolableTest> pool = configAdjuster.apply(builder);
		pool.warmup().block();

		PooledRef<PoolableTest> slot1 = pool.acquire().block();
		PooledRef<PoolableTest> slot2 = pool.acquire().block();
		PooledRef<PoolableTest> slot3 = pool.acquire().block();

		assertThat(slot1).as("slot1").isNotNull();
		assertThat(slot2).as("slot2").isNotNull();
		assertThat(slot3).as("slot3").isNotNull();

		pool.dispose();

		assertThat(pool.idleSize()).as("idleSize").isZero();

