BoundedReplaySubjectBenchmark.java

/**
 * 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 rxjava;

import rx.subjects.BoundedReplaySubject;
import rx.subjects.PublishSubject;
import rx.subjects.ReplaySubject;
import rx.subjects.Subject;
import rx.util.functions.Action1;
import rx.util.functions.Func0;

/**
 *
 */
public class BoundedReplaySubjectBenchmark {
    static int n = 10;
    static int c = 500_000;
    static void benchmark(final Func0<Subject<Object, Object>> replayer, final Func0<Subject<Object, Object>> publisher, int flags) {
        if ((flags & 1) != 0) {
            System.out.println("ReplaySubject fill");
            Util.measureLoop(n, c, new Runnable() {
                public void run() {
                    Subject<Object, Object> rs = replayer.call();
                    for (int i = 0; i < c; i++) {
                        rs.onNext(i);
                    }
                }
            });
        }
        // --------------------------------------------------
        if ((flags & 1) != 0) {
            System.out.println("ReplaySubject fill 2");
            Util.measureLoop(n, c, new Runnable() {
                public void run() {
                    Subject<Object, Object> rs = replayer.call();
                    for (int i = 0; i < c; i++) {
                        rs.onNext(i);
                    }
                }
            });
        }
        // --------------------------------------------------
        
        if ((flags & 2) != 0) {
            System.out.println("ReplaySubject just replay");
            final Subject<Object, Object> rs0 = replayer.call();
            for (int i = 0; i < c; i++) {
                rs0.onNext(i);
            }
            Util.measureLoop(n, c, new Runnable() {
                public void run() {
                    final int[] value = new int[1];
                    rs0.subscribe(new Action1<Object>(){ public void call(Object o) { value[0]++; }});
                    if (value[0] != c) {
                        System.err.println("Events lost: " + value[0]);
                    }
                }
            });
        }
        // --------------------------------------------------
        
        if ((flags & 4) != 0) {
            System.out.println("ReplaySubject fill & consume");
            Util.measureLoop(n, c, new Runnable() {
                public void run() {
                    Subject<Object, Object> rs = replayer.call();
                    final int[] value = new int[1];
                    rs.subscribe(new Action1<Object>(){ public void call(Object o) { value[0]++; }});
                    for (int i = 0; i < c; i++) {
                        rs.onNext(i);
                    }
                   if (value[0] != c) {
                       System.err.println("Events lost: " + value[0]);
                   }
                }
            });
        }

        // --------------------------------------------------
        
        if ((flags & 8) != 0) {
            System.out.println("PublishSubject fill & consume");
            Util.measureLoop(n, c, new Runnable() {
                public void run() {
                    publishSubjectFillConsume(publisher);
                }
            });
        }
    }
    static void publishSubjectFillConsume(Func0<Subject<Object, Object>> publisher) {
        Subject<Object, Object> rs = publisher.call();
            final int[] value = new int[1];
            rs.subscribe(new Action1<Object>(){ public void call(Object o) { value[0]++; }});
            produce1(rs);
            if (value[0] != c) {
                System.err.println("Events lost: " + value[0]);
            }
    }
    static void produce1(Subject<Object, Object> rs) {
            for (int i = 0; i < c; i++) {
                rs.onNext(i);
            }
    }
    
    public static void main(String[] args) {
        System.out.println("BASELINE");
        /*
        benchmark(new Func0<Subject<Object, Object>>() {
            @Override
            public Subject<Object, Object> call() {
                return ReplaySubject.create();
            }
            
        }, new Func0<Subject<Object, Object>>() {

            @Override
            public Subject<Object, Object> call() {
                return PublishSubject.create();
            }
            
        }, 15);
        */
        System.out.println("XOXOXOXOXOXOXOXOXOXOXOXOXOXO");
        System.out.println("ALTERNATIVE");
        benchmark(new Func0<Subject<Object, Object>>() {

            @Override
            public Subject<Object, Object> call() {
                return BoundedReplaySubject.create();
            }
            
        }, new Func0<Subject<Object, Object>>() {

            @Override
            public Subject<Object, Object> call() {
                return BoundedReplaySubject.create(0);
            }
            
        }, 8);
    }
}