package org.davidmoten.rx.pool;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;

import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.junit.Test;

import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;

public class NonBlockingPoolConcurrencyTest {

    @Test
    public void memberSingleCoverage() throws Exception {
        // attempt to get coverage of x.activeCount > 0 expression in tryEmit
        // has not been successful but covers the other statements in tryEmit
        AtomicLong count = new AtomicLong();
        AtomicLong disposed = new AtomicLong();
        int poolSize = 5;
        try (Pool<Long> pool = NonBlockingPool //
                .factory(() -> count.incrementAndGet()) //
                .healthCheck(n -> true) //
                .maxSize(poolSize) //
                .maxIdleTime(1, TimeUnit.MINUTES) //
                .disposer(n -> disposed.incrementAndGet()) //
                .build()) {
            long n = Long.parseLong(System.getProperty("n", "100000"));
            long[] c = new long[1];
            Flowable //
                    .rangeLong(0, n) //
                    .flatMapSingle(x -> pool.member(), false, poolSize * 2) //
                    .doOnNext(x -> c[0]++) //
                    .observeOn(Schedulers.from(Executors.newFixedThreadPool(1))) //
                    .doOnNext(member -> member.checkin()) //
                    .timeout(10, TimeUnit.SECONDS) //
                    .doOnError(e -> {
                        System.out.println("emitted " + c[0] + ", count=" + count);
                    }) //
                    .count() //
                    .blockingGet();
            // note that the last member in particular may
            assertEquals(c[0], n);
            assertEquals(0, disposed.get());
        }
        assertEquals(poolSize, disposed.get());
    }

}

@@ -74,51 +74,8 @@
 als;
-%0Aimport static org.junit.Assert.assertTrue;
 %0A%0Aim
@@ -686,9 +686,9 @@
 e = 
-5
+4
 ;%0A  
@@ -1264,12 +1264,8 @@
 Size
- * 2
 ) //
@@ -1307,24 +1307,159 @@
  c%5B0%5D++) //%0A
+                    // have to keep the observeOn buffer small so members don't get buffered%0A                    // and not checked in%0A
             
@@ -1525,16 +1525,26 @@
 Pool(1))
+, false, 1
 ) //%0A   

fix concurrent test in rxjava2-pool