package reactivestreams.commons.publisher;

import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
import java.util.function.LongSupplier;

import org.junit.Assert;
import org.junit.Test;
import reactivestreams.commons.test.TestSubscriber;
import reactivestreams.commons.util.ConstructorTestBuilder;

public class PublisherIntervalTest {

    @Test
    public void constructors() {
        
        ScheduledExecutorService exec = Executors.newScheduledThreadPool(1);
        
        ConstructorTestBuilder ctb = new ConstructorTestBuilder(PublisherInterval.class);
        
        ctb.addLong("initialDelay", Long.MIN_VALUE, Long.MAX_VALUE);
        ctb.addLong("period", 0, Long.MAX_VALUE);
        ctb.addRef("unit", TimeUnit.SECONDS);
        ctb.addRef("executor", exec);
        ctb.addRef("asyncExecutor", (BiFunction<Runnable, Long, Runnable>)(r, a) -> r);
        ctb.addRef("now", (LongSupplier)() -> 1L);
        
        exec.shutdown();
    }
    
    @Test
    public void normal() {
        ScheduledExecutorService exec = Executors.newScheduledThreadPool(1);
        try {
            TestSubscriber<Long> ts = new TestSubscriber<>();
            
            ts.values().add(System.currentTimeMillis());
            
            new PublisherInterval(100, 100, TimeUnit.MILLISECONDS, exec)
            .take(5)
            .map(v -> System.currentTimeMillis()).subscribe(ts);
            
            ts.await(5, TimeUnit.SECONDS);
            
            ts.assertValueCount(6)
            .assertNoError()
            .assertComplete();
            
            List<Long> list = ts.values();
            for (int i = 0; i < list.size() - 1; i++) {
                long diff = list.get(i + 1) - list.get(i);
                
                if (diff < 50 || diff > 150) {
                    Assert.fail("Period failure: " + diff);
                }
            }
            
        } finally {
            exec.shutdown();
        }
    }
}

@@ -33,24 +33,49 @@
 publisher;%0A%0A
+import java.util.Arrays;%0A
 import java.
@@ -291,24 +291,48 @@
 gSupplier;%0A%0A
+import org.junit.After;%0A
 import org.j
@@ -344,16 +344,41 @@
 Assert;%0A
+import org.junit.Before;%0A
 import o
@@ -388,24 +388,25 @@
 junit.Test;%0A
+%0A
 import react
@@ -544,29 +544,71 @@
 Test %7B%0A%0A    
-@Test
+ScheduledExecutorService exec;%0A    %0A    @Before
 %0A    public 
@@ -616,20 +616,14 @@
 oid 
-constructors
+before
 () %7B
@@ -635,42 +635,8 @@
     
-%0A        ScheduledExecutorService 
 exec
@@ -667,32 +667,155 @@
 dThreadPool(1);%0A
+    %7D%0A    %0A    @After%0A    public void after() %7B%0A        exec.shutdown();%0A    %7D%0A%0A    @Test%0A    public void constructors() %7B%0A
         %0A       
@@ -1248,24 +1248,16 @@
 -%3E 1L);%0A
-        
 %0A       
@@ -1253,37 +1253,32 @@
 );%0A%0A        
-exec.shutdown
+ctb.test
 ();%0A    %7D%0A  
@@ -1321,85 +1321,8 @@
 ) %7B%0A
-        ScheduledExecutorService exec = Executors.newScheduledThreadPool(1);%0A
     
@@ -2204,10 +2204,578 @@
 %7D%0A    %7D%0A
+    %0A    @Test%0A    public void flatMap() throws Exception %7B%0A        TestSubscriber%3CObject%3E ts = new TestSubscriber%3C%3E();%0A        %0A        PublisherBase.interval(30, TimeUnit.MILLISECONDS, exec).flatMap(v -%3E %0A                PublisherBase.%3CString%3EfromIterable(Arrays.asList(%22A%22))%0A                .flatMap(w -%3E PublisherBase.%3CList%3CInteger%3E%3EfromCallable(() -%3E Arrays.asList(1, 2)).subscribeOn(exec).flatMap(PublisherBase::fromIterable)))%0A        .subscribe(ts);%0A        %0A        Thread.sleep(5000);%0A        %0A        ts.cancel();%0A        %0A        ts.assertNoError();%0A    %7D%0A
 %7D%0A

Fix interval constructor test, add some flatMap test case