/**
 * Copyright 2014 Netflix, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package rx.operators;

import java.util.concurrent.TimeUnit;

import org.openjdk.jmh.annotations.BenchmarkMode;
import org.openjdk.jmh.annotations.GenerateMicroBenchmark;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
import org.openjdk.jmh.annotations.Param;
import org.openjdk.jmh.annotations.Scope;
import org.openjdk.jmh.annotations.State;

import rx.Observable;
import rx.functions.Func1;
import rx.jmh.InputWithIncrementingInteger;
import rx.schedulers.Schedulers;

@BenchmarkMode(Mode.Throughput)
@OutputTimeUnit(TimeUnit.SECONDS)
@State(Scope.Thread)
public class OperatorFlatMapPerf {

    @State(Scope.Thread)
    public static class Input extends InputWithIncrementingInteger {

        @Param({ "1", "1000", "1000000" })
        public int size;

        @Override
        public int getSize() {
            return size;
        }

    }

    @GenerateMicroBenchmark
    public void flatMapIntPassthruSync(Input input) throws InterruptedException {
        input.observable.flatMap(new Func1<Integer, Observable<Integer>>() {

            @Override
            public Observable<Integer> call(Integer i) {
                return Observable.just(i);
            }

        }).subscribe(input.observer);
    }

    @GenerateMicroBenchmark
    public void flatMapIntPassthruAsync(Input input) throws InterruptedException {
        input.observable.flatMap(new Func1<Integer, Observable<Integer>>() {

            @Override
            public Observable<Integer> call(Integer i) {
                return Observable.just(i).subscribeOn(Schedulers.computation());
            }

        }).subscribe(input.observer);
    }

    @GenerateMicroBenchmark
    public void flatMapTwoNestedSync(final Input input) throws InterruptedException {
        Observable.range(1, 10).flatMap(new Func1<Integer, Observable<Integer>>() {

            @Override
            public Observable<Integer> call(Integer i) {
                return input.observable;
            }

        }).subscribe(input.observer);
    }

    @GenerateMicroBenchmark
    public void flatMapTenNestedAsync(final Input input) throws InterruptedException {
        Observable.range(1, 10).flatMap(new Func1<Integer, Observable<Integer>>() {

            @Override
            public Observable<Integer> call(Integer i) {
                return input.observable.subscribeOn(Schedulers.computation());
            }

        }).subscribe(input.observer);
    }

}

@@ -2396,34 +2396,33 @@
 rvable.range(1, 
-10
+2
 ).flatMap(new Fu
@@ -2632,32 +2632,79 @@
 server);%0A    %7D%0A%0A
+    // this runs out of memory currently%0A    //
     @GenerateMic
@@ -2707,32 +2707,38 @@
 eMicroBenchmark%0A
+    //
     public void 
@@ -2745,18 +2745,18 @@
 flatMapT
-en
+wo
 NestedAs
@@ -2800,32 +2800,38 @@
 ptedException %7B%0A
+    //
         Observab
@@ -2842,18 +2842,17 @@
 ange(1, 
-10
+2
 ).flatMa
@@ -2889,33 +2889,45 @@
 e%3CInteger%3E%3E() %7B%0A
-%0A
+    //%0A    //
             @Ove
@@ -2924,32 +2924,38 @@
       @Override%0A
+    //
             publ
@@ -2991,32 +2991,34 @@
 nteger i) %7B%0A    
+//
             retu
@@ -3005,32 +3005,36 @@
   //            
+    
 return input.obs
@@ -3072,32 +3072,38 @@
 computation());%0A
+    //
             %7D%0A%0A 
@@ -3092,33 +3092,45 @@
 //            %7D%0A
-%0A
+    //%0A    //
         %7D).subsc
@@ -3147,21 +3147,27 @@
 .observer);%0A
+    //
     %7D%0A%0A%7D%0A

Disable flatMapTwoNestedAsync for now