package org.davidmoten.rx.pool;

import static org.junit.Assert.assertEquals;

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(timeout=30000)
    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 = 4;
        try (Pool<Long> pool = NonBlockingPool //
                .factory(() -> count.incrementAndGet()) //
                .healthCheck(n -> true) //
                .maxSize(poolSize) //
                .maxIdleTime(1, TimeUnit.MINUTES) //
                .disposer(n -> disposed.incrementAndGet()) //
                .build()) {
            Flowable //
                    .rangeLong(0, 100000) //
                    .flatMapSingle(x -> pool.member(), false, 4) //
                    .observeOn(Schedulers.from(Executors.newFixedThreadPool(1))) //
                    .doOnNext(member -> member.checkin()) //
                    .count() //
                    .blockingGet();
            assertEquals(poolSize, count.get());
            assertEquals(0, disposed.get());
        }
        assertEquals(poolSize, disposed.get());
    }

}

@@ -70,16 +70,59 @@
 tEquals;
+%0Aimport static org.junit.Assert.assertTrue;
 %0A%0Aimport
@@ -398,23 +398,8 @@
 Test
-(timeout=30000)
 %0A   
@@ -725,17 +725,17 @@
 lSize = 
-4
+5
 ;%0A      
@@ -1061,16 +1061,124 @@
 ld()) %7B%0A
+            long n = Long.parseLong(System.getProperty(%22n%22, %22100000%22));%0A            long%5B%5D c = new long%5B1%5D;%0A
         
@@ -1227,22 +1227,17 @@
 Long(0, 
-100000
+n
 ) //%0A   
@@ -1299,9 +1299,66 @@
 se, 
-4
+poolSize * 2) //%0A                    .doOnNext(x -%3E c%5B0%5D++
 ) //
@@ -1528,101 +1528,348 @@
    .
-count() //%0A                    .blockingGet();%0A            assertEquals(poolSize, count.get()
+timeout(10, TimeUnit.SECONDS) //%0A                    .doOnError(e -%3E %7B%0A                        System.out.println(%22emitted %22 + c%5B0%5D + %22, count=%22 + count);%0A                    %7D) //%0A                    .count() //%0A                    .blockingGet();%0A            // note that the last member in particular may%0A            assertEquals(c%5B0%5D, n
 );%0A 

fix concurrency test (count emissions before observeOn which buffers)