Merge pull request #3031 from benjchristensen/fix-flatMapIntPassthruAsync

Fix OperatorFlatMapPerf.flatMapIntPassthruAsync Perf Test

Author: benjchristensen

src/perf/java/rx/operators/OperatorFlatMapPerf.java

@@ -28,6 +28,7 @@
 import rx.Observable;
 import rx.functions.Func1;
 import rx.jmh.InputWithIncrementingInteger;
+import rx.jmh.LatchedObserver;
 import rx.schedulers.Schedulers;
 
 @BenchmarkMode(Mode.Throughput)
@@ -62,14 +63,16 @@ public Observable<Integer> call(Integer i) {
 
     @Benchmark
     public void flatMapIntPassthruAsync(Input input) throws InterruptedException {
+        LatchedObserver<Integer> latchedObserver = input.newLatchedObserver();
         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);
+        }).subscribe(latchedObserver);
+        latchedObserver.latch.await();
     }
 
     @Benchmark