TestRxPerf - simple map transform

AbstractPerformanceTester.java

package rx.perf;

import java.util.Iterator;

import rx.functions.Action0;

public abstract class AbstractPerformanceTester {

    public static final long REPETITIONS = 5 * 1000 * 1000;
    public static final int NUM_PRODUCERS = 1;

    private final long repetitions;

    protected AbstractPerformanceTester(long repetitions) {
        this.repetitions = repetitions;
    }

    public final void runTest(Action0 action) throws InterruptedException {
        for (int runNum = 0; runNum < 15; runNum++) {
            System.gc();
            Thread.sleep(1000L);

            final long start = System.nanoTime();

            action.call();

            long duration = System.nanoTime() - start;
            long opsPerSec = (repetitions * NUM_PRODUCERS * 1000L * 1000L * 1000L) / duration;
            System.out.printf("Run: %d - %,d ops/sec \n",
                    Integer.valueOf(runNum),
                    Long.valueOf(opsPerSec));
        }
    }

    /**
     * Baseline ops/second without a subject.
     * 
     * Perf along this order of magnitude:
     * 
     * Run: 10 - 316,235,532 ops/sec
     * Run: 11 - 301,886,792 ops/sec
     * Run: 12 - 310,472,228 ops/sec
     * Run: 13 - 313,469,797 ops/sec
     * Run: 14 - 305,380,809 ops/sec
     */
    public long baseline() {
        LongSumObserver o = new LongSumObserver();
        for (long l = 0; l < repetitions; l++) {
            o.onNext(l);
        }
        o.onCompleted();
        return o.sum;
    }

    public Iterable<Long> ITERABLE_OF_REPETITIONS = new Iterable<Long>() {

        @Override
        public Iterator<Long> iterator() {
            return new Iterator<Long>() {
                long count = 0;

                @Override
                public boolean hasNext() {
                    return count <= repetitions;
                }

                @Override
                public Long next() {
                    return count++;
                }

                @Override
                public void remove() {
                    // do nothing
                }

            };
        };
    };

}

IntegerSumObserver.java

package rx.perf;

import rx.Observer;


public class IntegerSumObserver implements Observer<Integer> {

    public int sum = 0;

    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {
        e.printStackTrace();
        throw new RuntimeException(e);
    }

    @Override
    public void onNext(Integer l) {
        sum += l;
    }
}

LongSumObserver.java

package rx.perf;

import rx.Observer;


public class LongSumObserver implements Observer<Long> {

    public long sum = 0;

    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {
        throw new RuntimeException(e);
    }

    @Override
    public void onNext(Long l) {
        sum += l;
    }
}

TestRxPerf.java

package rx.perf;

import rx.Observable;

public class TestRxPerf extends AbstractPerformanceTester {

    static final int NUM_ITEMS_PER_OBSERVABLE = 1;
//    static final long LOOP = Long.MAX_VALUE;
    static final long LOOP = 1000000;
    static final long REPS = NUM_ITEMS_PER_OBSERVABLE * LOOP;

    protected TestRxPerf() {
        super(REPS);
    }

    public static void main(String args[]) {

        final TestRxPerf pt = new TestRxPerf();
        final UseCaseInput input = new UseCaseInput();
        input.size = NUM_ITEMS_PER_OBSERVABLE;
        input.setup();
        try {
            pt.runTest(() -> {
                for (int i = 0; i < LOOP; i++) {
                    try {
                        pt.mapTransformation(input);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public void mapTransformation(UseCaseInput input) throws InterruptedException {
        input.observable.map(i -> {
            return String.valueOf(i);

        }).map(i -> {
            return Integer.parseInt(i);
        }).subscribe(input.observer);
        input.awaitCompletion();
    }

    public void flatMapTransformsUsingFrom(UseCaseInput input) throws InterruptedException {
        input.observable.flatMap(i -> {
            return Observable.from(i);
        }).subscribe(input.observer);
        input.awaitCompletion();
    }

    public void flatMapTransformsUsingJust(UseCaseInput input) throws InterruptedException {
        input.observable.flatMap(i -> {
            return Observable.just(i);
        }).subscribe(input.observer);
        input.awaitCompletion();
    }
}