package hu.akarnokd.comparison;

import java.util.concurrent.TimeUnit;

import org.openjdk.jmh.annotations.*;

import com.typesafe.config.*;

import akka.NotUsed;
import akka.actor.ActorSystem;
import akka.stream.ActorMaterializer;
import akka.stream.javadsl.*;
import akka.stream.scaladsl.Sink;
import hu.akarnokd.akka.ActorScheduler2;
import io.reactivex.Flowable;
import io.reactivex.schedulers.Schedulers;

@BenchmarkMode(Mode.Throughput)
@Warmup(iterations = 5)
@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@OutputTimeUnit(TimeUnit.SECONDS)
@Fork(value = 1)
@State(Scope.Thread)
@SuppressWarnings({ "rawtypes", "unchecked" })
public class MaterializationBenchmarkPerf {

    @Param({"1", "10"})
    int complexity;
    
    RunnableGraph<NotUsed> akkaMap;
    
    Flowable<Integer> flowableMap;

    Flowable<Integer> flowableMapAkkaScheduler;

    Flowable<Integer> flowableMapSync;

    ActorSystem actorSystem;

    ActorMaterializer materializer;
    
    @Setup
    public void setup() {
        Config cfg = ConfigFactory.parseResources(ReactiveStreamsImplsAsync.class, "/akka-streams.conf").resolve();
        actorSystem = ActorSystem.create("sys", cfg);

        materializer = ActorMaterializer.create(actorSystem);
        
        Source<Integer, NotUsed> src = Source.single(1);
        
        for (int i = 0; i < complexity; i++) {
            src = src.map(v -> v);
        }
        akkaMap = src.to((Sink)Sink.ignore());
        
        Flowable<Integer> f = Flowable.just(1).subscribeOn(Schedulers.computation());

        for (int i = 0; i < complexity; i++) {
            f = f.map(v -> v).observeOn(Schedulers.computation(), true, 1);
        }

        
        flowableMap = f;
        
        Flowable<Integer> g = Flowable.just(1);

        for (int i = 0; i < complexity; i++) {
            g = g.map(v -> v);
        }
        
        flowableMapSync = g;
        
        ActorScheduler2 as = new ActorScheduler2(actorSystem, true);
        
        Flowable<Integer> h = Flowable.just(1).subscribeOn(as);
        for (int i = 0; i < complexity; i++) {
            h = h.map(v -> v).observeOn(as, true, 1);
        }

        
        flowableMapAkkaScheduler = h;
    }
    
    @TearDown
    public void teardown() {
        actorSystem.terminate();
    }
    
    @Benchmark
    public Object akkaMap() {
        return akkaMap.run(materializer);
    }

    @Benchmark
    public Object flowableMap() {
        return flowableMap.blockingLast();
    }

    @Benchmark
    public Object flowableMapAkkaScheduler() {
        return flowableMapAkkaScheduler.blockingLast();
    }

    @Benchmark
    public Object flowableMapSync() {
        return flowableMapSync.blockingLast();
    }
    
    public static void main(String[] args) {
        MaterializationBenchmarkPerf bench = new MaterializationBenchmarkPerf();
        bench.complexity = 10;
        bench.setup();
        
        for (int i = 0; i < 100000; i++) {
            bench.flowableMapAkkaScheduler();
        }
        
        bench.teardown();
    }
}

@@ -818,16 +818,49 @@
 leMap;%0A%0A
+    Flowable%3CInteger%3E flowable;%0A%0A
     Flow
@@ -2253,22 +2253,271 @@
 er = h;%0A
+        %0A        Flowable%3CInteger%3E k = Flowable.just(1).subscribeOn(Schedulers.computation());%0A%0A        for (int i = 0; i %3C complexity; i++) %7B%0A            k = k.observeOn(Schedulers.computation(), true, 1);%0A        %7D%0A%0A        %0A        flowable = k;%0A
     %7D%0A
-
     %0A   
@@ -2596,24 +2596,26 @@
 %0A    %7D%0A    %0A
+//
     @Benchma
@@ -2609,32 +2609,32 @@
 /    @Benchmark%0A
-
     public Objec
@@ -2688,32 +2688,34 @@
 alizer);%0A    %7D%0A%0A
+//
     @Benchmark%0A 
@@ -2705,32 +2705,32 @@
 /    @Benchmark%0A
-
     public Objec
@@ -2804,32 +2804,127 @@
 %0A    @Benchmark%0A
+    public Object flowable() %7B%0A        return flowable.blockingLast();%0A    %7D%0A%0A//    @Benchmark%0A
     public Objec
@@ -3009,32 +3009,34 @@
 gLast();%0A    %7D%0A%0A
+//
     @Benchmark%0A 

Add no-map (thus fused) benchmark entry