benjchristensen

        compositeSubscription.add(scheduler.schedule(new Action1<Action0>() {
            @Override
            public void call(final Action0 self) {
                innerSubscription.set(source.subscribe(new Observer<T>() {

                    @Override
                    public void onCompleted() {
                        self.call();
                    }

benjchristensen

package perf.backend;

import rx.Observable;
import rx.Observable.Operator;
import rx.Subscriber;
import rx.subjects.AsyncSubject;
import rx.subjects.ReplaySubject;
import rx.subjects.Subject;

public class ChooseSubjectBasedOnFirstValue {

benjchristensen

Keybase proof

I hereby claim:

• I am benjchristensen on github.
• I am benjchristensen (https://keybase.io/benjchristensen) on keybase.
• I have a public key whose fingerprint is 6EC3 891B AF53 E272 976E 3060 EB6A 08CB EC72 5A7B

To claim this, I am signing this object:

benjchristensen

import rx.Notification;
import rx.Observable;
import rx.Observable.Operator;
import rx.Subscriber;
import rx.functions.Func1;

public class ConcatIfEmpty {

    public static void main(String[] args) {

benjchristensen

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;

benjchristensen

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

benjchristensen

import java.util.concurrent.TimeUnit;

import rx.Observable;

public class MergeUntilComplete {

    public static void main(String[] args) {

        Observable<String> t1 = Observable.timer(0, 100, TimeUnit.MILLISECONDS).map(i -> "A-" + i);
        Observable<String> t2 = Observable.timer(0, 300, TimeUnit.MILLISECONDS).take(5).map(i -> "B-" + i);

benjchristensen

import java.util.List;
import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

public class DebounceExample {

    public static void main(String args[]) {

benjchristensen

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

public class BufferExample {

    public static void main(String args[]) {
        // buffer every 500ms

benjchristensen

import java.util.concurrent.TimeUnit;

import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

public class WindowExample {

    public static void main(String args[]) {
        // buffer every 500ms (using 999999999 to mark start of output)