// Path: rxjava2-pool/src/main/java/org/davidmoten/rx/internal/FlowableSingleDeferUntilRequest.java
// public final class FlowableSingleDeferUntilRequest<T> extends Flowable<T> {
// 
//     private final Single<T> single;
// 
//     public FlowableSingleDeferUntilRequest(Single<T> single) {
//         this.single = single;
//     }
// 
//     @Override
//     protected void subscribeActual(Subscriber<? super T> s) {
//         SingleSubscription<T> sub = new SingleSubscription<T>(single, s);
//         s.onSubscribe(sub);
//     }
// 
//     private static final class SingleSubscription<T> extends AtomicBoolean implements Subscription, SingleObserver<T> {
// 
//         private static final long serialVersionUID = -4290226935675014466L;
//         
//         private final Subscriber<? super T> s;
//         private final Single<T> single;
//         private final AtomicReference<Disposable> disposable = new AtomicReference<Disposable>();
// 
//         SingleSubscription(Single<T> single, Subscriber<? super T> s) {
//             super();
//             this.single = single;
//             this.s = s;
//         }
// 
//         @Override
//         public void request(long n) {
//             if (n > 0 && this.compareAndSet(false, true)) {
//                 Disposable d = disposable.get();
//                 if (d == null) {
//                     single.subscribe(this);
//                 }
//             }
//         }
// 
//         @Override
//         public void cancel() {
//             if (disposable.compareAndSet(null, Disposables.disposed())) {
//                 return;
//             } else {
//                 disposable.get().dispose();
//                 // clear for GC
//                 disposable.set(Disposables.disposed());
//             }
//         }
// 
//         @Override
//         public void onSubscribe(Disposable d) {
//             if (!disposable.compareAndSet(null, d)) {
//                 // already cancelled
//                 d.dispose();
//                 disposable.set(Disposables.disposed());
//             }
//         }
// 
//         @Override
//         public void onSuccess(T t) {
//             s.onNext(t);
//             s.onComplete();
//         }
// 
//         @Override
//         public void onError(Throwable e) {
//             s.onError(e);
//         }
//     }
// 
// }


import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.davidmoten.rx.internal.FlowableSingleDeferUntilRequest;
import org.junit.Test;
import io.reactivex.Flowable;
import io.reactivex.Scheduler;
import io.reactivex.Single;
import io.reactivex.SingleObserver;
import io.reactivex.disposables.Disposable;
import io.reactivex.exceptions.UndeliverableException;
import io.reactivex.functions.Consumer;
import io.reactivex.observers.TestObserver;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.schedulers.TestScheduler;
import io.reactivex.subscribers.TestSubscriber;
package org.davidmoten.rx.pool;





public class NonBlockingPoolTest {

    @Test
    public void testMaxIdleTime() throws InterruptedException {
        TestScheduler s = new TestScheduler();
        AtomicInteger count = new AtomicInteger();
        AtomicInteger disposed = new AtomicInteger();
        Pool<Integer> pool = NonBlockingPool //
                .factory(() -> count.incrementAndGet()) //
                .healthCheck(n -> true) //
                .maxSize(3) //
                .maxIdleTime(1, TimeUnit.MINUTES) //
                .disposer(n -> disposed.incrementAndGet()) //
                .scheduler(s) //
                .build();
// Path: rxjava2-pool/src/main/java/org/davidmoten/rx/internal/FlowableSingleDeferUntilRequest.java
// public final class FlowableSingleDeferUntilRequest<T> extends Flowable<T> {
// 
//     private final Single<T> single;
// 
//     public FlowableSingleDeferUntilRequest(Single<T> single) {
//         this.single = single;
//     }
// 
//     @Override
//     protected void subscribeActual(Subscriber<? super T> s) {
//         SingleSubscription<T> sub = new SingleSubscription<T>(single, s);
//         s.onSubscribe(sub);
//     }
// 
//     private static final class SingleSubscription<T> extends AtomicBoolean implements Subscription, SingleObserver<T> {
// 
//         private static final long serialVersionUID = -4290226935675014466L;
//         
//         private final Subscriber<? super T> s;
//         private final Single<T> single;
//         private final AtomicReference<Disposable> disposable = new AtomicReference<Disposable>();
// 
//         SingleSubscription(Single<T> single, Subscriber<? super T> s) {
//             super();
//             this.single = single;
//             this.s = s;
//         }
// 
//         @Override
//         public void request(long n) {
//             if (n > 0 && this.compareAndSet(false, true)) {
//                 Disposable d = disposable.get();
//                 if (d == null) {
//                     single.subscribe(this);
//                 }
//             }
//         }
// 
//         @Override
//         public void cancel() {
//             if (disposable.compareAndSet(null, Disposables.disposed())) {
//                 return;
//             } else {
//                 disposable.get().dispose();
//                 // clear for GC
//                 disposable.set(Disposables.disposed());
//             }
//         }
// 
//         @Override
//         public void onSubscribe(Disposable d) {
//             if (!disposable.compareAndSet(null, d)) {
//                 // already cancelled
//                 d.dispose();
//                 disposable.set(Disposables.disposed());
//             }
//         }
// 
//         @Override
//         public void onSuccess(T t) {
//             s.onNext(t);
//             s.onComplete();
//         }
// 
//         @Override
//         public void onError(Throwable e) {
//             s.onError(e);
//         }
//     }
// 
// }


// Path: rxjava2-pool/src/test/java/org/davidmoten/rx/pool/NonBlockingPoolTest.java
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.davidmoten.rx.internal.FlowableSingleDeferUntilRequest;
import org.junit.Test;
import io.reactivex.Flowable;
import io.reactivex.Scheduler;
import io.reactivex.Single;
import io.reactivex.SingleObserver;
import io.reactivex.disposables.Disposable;
import io.reactivex.exceptions.UndeliverableException;
import io.reactivex.functions.Consumer;
import io.reactivex.observers.TestObserver;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.Schedulers;
import io.reactivex.schedulers.TestScheduler;
import io.reactivex.subscribers.TestSubscriber;

package org.davidmoten.rx.pool;





public class NonBlockingPoolTest {

    @Test
    public void testMaxIdleTime() throws InterruptedException {
        TestScheduler s = new TestScheduler();
        AtomicInteger count = new AtomicInteger();
        AtomicInteger disposed = new AtomicInteger();
        Pool<Integer> pool = NonBlockingPool //
                .factory(() -> count.incrementAndGet()) //
                .healthCheck(n -> true) //
                .maxSize(3) //
                .maxIdleTime(1, TimeUnit.MINUTES) //
                .disposer(n -> disposed.incrementAndGet()) //
                .scheduler(s) //
                .build();
        TestSubscriber<Member<Integer>> ts = new FlowableSingleDeferUntilRequest<>( //