package hu.akarnokd.comparison.classic;

import java.util.concurrent.*;

import org.openjdk.jmh.annotations.*;
import org.openjdk.jmh.infra.Blackhole;

import io.reactivex.Flowable;
import io.reactivex.disposables.Disposable;
import io.reactivex.processors.PublishProcessor;

@BenchmarkMode(Mode.Throughput)
@Warmup(iterations = 5)
@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@OutputTimeUnit(TimeUnit.SECONDS)
@Fork(value = 1)
@State(Scope.Thread)
public class ClassicAsyncPerf {
    @Param({ "1", "10", "100", "1000", "10000", "100000", "1000000" })
    public int count;

    ExecutorService exec;

    rx.subjects.PublishSubject<Integer> ps;
    
    PublishProcessor<Integer> pp;

    rx.Observable<Integer> observableExec;
    
    rx.Observable<Integer> observableFJ;
    
    Flowable<Integer> flowableExec;
    
    Flowable<Integer> flowableFJ;
    
    @Setup
    public void setup() throws Exception {
        exec = Executors.newSingleThreadExecutor();
        
        rx.Scheduler s1 = rx.schedulers.Schedulers.from(exec);
        io.reactivex.Scheduler s2 = io.reactivex.schedulers.Schedulers.from(exec);

        rx.Scheduler fj1 = rx.schedulers.Schedulers.from(ForkJoinPool.commonPool());
        io.reactivex.Scheduler fj2 = io.reactivex.schedulers.Schedulers.from(ForkJoinPool.commonPool());
        
        ps = rx.subjects.PublishSubject.create();
        
        pp = io.reactivex.processors.PublishProcessor.create();

        observableExec = ps.onBackpressureBuffer().observeOn(s1);
        
        observableFJ = ps.onBackpressureBuffer().observeOn(fj1);
        
        flowableExec = pp.onBackpressureBuffer().observeOn(s2);
        
        flowableFJ = pp.onBackpressureBuffer().observeOn(fj2);
    }

    @TearDown
    public void teardown() {
        exec.shutdownNow();
    }
    
    void await(CountDownLatch cdl, int c) throws Exception {
        if (c <= 1000) {
            while (cdl.getCount() != 0L) ;
        } else {
            cdl.await(5, TimeUnit.SECONDS);
        }
    }
    
    @Benchmark
    public void plainExecutor(Blackhole bh) throws Exception {
        CountDownLatch cdl = new CountDownLatch(1);
        int c = count;
        
        for (int i = 1; i <= c; i++) {
            int j = i;
            Callable<Integer> call = () -> {
                if (j == c) {
                    cdl.countDown();
                } else {
                    bh.consume(j);
                }
                return null;
            };
            
            exec.submit(call);
        }
        
        await(cdl, c);
    }
    
    @Benchmark
    public void plainForkJoin(Blackhole bh) throws Exception {
        CountDownLatch cdl = new CountDownLatch(1);
        int c = count;
        
        ExecutorService fj = ForkJoinPool.commonPool();
        
        for (int i = 1; i <= c; i++) {
            int j = i;
            Callable<Integer> call = () -> {
                if (j == c) {
                    cdl.countDown();
                } else {
                    bh.consume(j);
                }
                return null;
            };
            
            fj.submit(call);
        }
        
        await(cdl, c);
    }
    
    @Benchmark
    public void observableExecutor(Blackhole bh) throws Exception {
        int c = count;
        CountDownLatch cdl = new CountDownLatch(1);
        
        rx.Subscription s = observableExec.subscribe(j -> {
            if (j == c) {
                cdl.countDown();
            } else {
                bh.consume(j);
            }
        });
        
        for (int i = 1; i <= c; i++) {
            pp.onNext(i);
        }
        
        await(cdl, c);
        s.unsubscribe();
    }
    
    @Benchmark
    public void observableForkJoin(Blackhole bh) throws Exception {
        int c = count;
        CountDownLatch cdl = new CountDownLatch(1);
        
        rx.Subscription s = observableFJ.subscribe(j -> {
            if (j == c) {
                cdl.countDown();
            } else {
                bh.consume(j);
            }
        });
        
        for (int i = 1; i <= c; i++) {
            pp.onNext(i);
        }
        
        await(cdl, c);
        s.unsubscribe();
    }
    
    @Benchmark
    public void flowableExecutor(Blackhole bh) throws Exception {
        int c = count;
        CountDownLatch cdl = new CountDownLatch(1);
        
        Disposable d = flowableExec.subscribe(j -> {
            if (j == c) {
                cdl.countDown();
            } else {
                bh.consume(j);
            }
        });
        
        for (int i = 1; i <= c; i++) {
            pp.onNext(i);
        }
        
        await(cdl, c);
        d.dispose();
    }
    
    @Benchmark
    public void flowableForkJoin(Blackhole bh) throws Exception {
        int c = count;
        CountDownLatch cdl = new CountDownLatch(1);
        
        Disposable d = flowableFJ.subscribe(j -> {
            if (j == c) {
                cdl.countDown();
            } else {
                bh.consume(j);
            }
        });
        
        for (int i = 1; i <= c; i++) {
            pp.onNext(i);
        }
        
        await(cdl, c);
        d.dispose();
    }
}

@@ -3636,33 +3636,33 @@
  %7B%0A            p
-p
+s
 .onNext(i);%0A    
@@ -4150,33 +4150,33 @@
  %7B%0A            p
-p
+s
 .onNext(i);%0A    

Fix wrong emission target