src/jmh/java/io/reactivex/OperatorMergePerf.java
@Benchmark
    public void oneStreamOfNthatMergesIn1(final InputMillion input) throws InterruptedException {
        Flowable<Flowable<Integer>> os = Flowable.range(1, input.size)
                .map(new Function<Integer, Flowable<Integer>>() {
                    @Override
                    public Flowable<Integer> apply(Integer v) {
                        return Flowable.just(v);
                    }
                });
        PerfSubscriber o = input.newLatchedObserver();
        Flowable.merge(os).subscribe(o);

        if (input.size == 1) {
            while (o.latch.getCount() != 0) { }
        } else {
            o.latch.await();
        }
    }
@Benchmark
    public void oneStreamOfNthatMergesIn1(final InputMillion input) throws InterruptedException {
        Flowable<Flowable<Integer>> os = Flowable.range(1, input.size)
                .map(new Function<Integer, Flowable<Integer>>() {
                    @Override
                    public Flowable<Integer> apply(Integer v) {
                        return Flowable.just(v);
                    }
                });
        PerfSubscriber o = input.newLatchedObserver();
        Flowable.merge(os).subscribe(o);

        if (input.size == 1) {
            while (o.latch.getCount() != 0) { }
        } else {
            o.latch.await();
        }
    }
@ Benchmark public void oneStreamOfNthatMergesIn1 ( final InputMillion input ) throws InterruptedException { Flowable < Flowable < Integer >> os = Flowable . range ( 1 , input . size ) . map ( new Function < Integer , Flowable < Integer > > ( ) { @ Override public Flowable < Integer > apply ( Integer v ) { return Flowable . just ( v ) ; } } ) ; PerfSubscriber o = input . newLatchedObserver ( ) ; Flowable . merge ( os ) . subscribe ( o ) ; if ( input . size == 1 ) { while ( o . latch . getCount ( ) != 0 ) { } } else { o . latch . await ( ) ; } }
flatMap
flatMap