[RxJava] Chap08. 테스팅과 Flowable

public class JUnitBasic {
    @Test
    public void testGetShapeObservable() {
        String[] data = {"1", "2-R", "3-T"};
        // source : 실제 결과.
        Observable<String> source = Observable.fromArray(data)
                .map(Shape::getShape);

        // expected : 예상 결과.
        String[] expected = {Shape.BALL, Shape.RECTANGLE, Shape.TRIANGLE};
        source.test().assertResult(expected);
    }
}

@Test
    public void assertFailureExample() {
        String[] data = {"100", "200", "%300"};
        Observable<Integer> source = Observable.fromArray(data)
                .map(Integer::parseInt);

        source.test().assertFailure(NumberFormatException.class, 100, 200);
    }

@Test
public void assertFailureAndMessage() {
    String[] data = {"100", "200", "%300"};
    Observable<Integer> source = Observable.fromArray(data)
            .map(Integer::parseInt);

    source.test().assertFailureAndMessage(NumberFormatException.class,
            "For input string : \"%300\"", 100, 200);
}

public class TestAsync {
    @Test
    public void testInterval() {
        Observable<Integer> source = Observable.interval(100L, TimeUnit.MILLISECONDS)
                .take(5)
                .map(Long::intValue);

        source.doOnNext(Log::d)
                .test()
                .awaitDone(1L, TimeUnit.SECONDS)
                .assertResult(0, 1, 2, 3, 4);

    }
}

@Test
public void testHttp() {
    final String url = "http://api.github.com/users/yudong80";
    Observable<String> source = Observable.just(url)
            .subscribeOn(Schedulers.io())
            .map(OkHttpHelper::get)
            .doOnNext(Log::d) // json 을 로그로 찍는다.
            .map(json -> GsonHelper.parseValue(json, "name"))
            .observeOn(Schedulers.newThread());

    // json 중에서 name 만 뽑은 것을 로그로 찍는다.
    String expected = "Dong Hwan Yu";
    source.doOnNext(Log::i)
            .test()
            .awaitDone(3, TimeUnit.SECONDS)
            .assertResult(expected);
}

public class FlowableSample {
    public static void main(String[] args) {
        CommonUtils.exampleStart();

        PublishSubject<Integer> subject = PublishSubject.create();
        subject.observeOn(Schedulers.computation())
                .subscribe(data -> {
                    CommonUtils.sleep(100); // 100ms 후에 데이터 처리.
                    Log.it(data);
                }, err -> Log.e(err.toString()));

        // 뜨거운 Observable 로 50,000,000개의 데이터를 연속으로 발행함.
        for (int i = 0; i < 50000000; i++) {
            subject.onNext(i);
        }

        subject.onComplete();
    }
}

// 기본값(128)의 버퍼 개수가 있다.
public final Flowable<T> onBackpressureBuffer()

// delayError 여부를 지정할 수 있다. 
// true : 예외가 발생했을 때 버퍼에 쌓인 데이터를 모두 처리할 때가지 예외를 던지지 않는다.
// false : 예외가 발생했을 때 바로 다운스트림에 예외를 던진다.
// 기본값은 false다.
public final Flowable<T> onBackpressureBuffer(boolean delayError)

// capacity 인자로 버퍼의 개수를 지정한다.
// onOverflow 인자에 버퍼가 넘쳤을 때 실행할 동작을 지정한다.
public final Flowable<T> onBackpressureBuffer(int capacity, Action onOverflow)

// 버퍼가 가득찼을 때 추가로 실행하는 전략을 지정할 수 있다.
public final Flowable<T> onBackpressureBuffer(long capacity, Action onOverflow, BackpressureOverflowStrategy overflowStrategy)

public class onBackPressureBufferSample {
    public static void main(String[] args) {
        CommonUtils.exampleStart();

        Flowable.range(1, 50000000)
                .onBackpressureBuffer(128, () -> {
                }, BackpressureOverflowStrategy.DROP_OLDEST)
                .observeOn(Schedulers.computation())
                .subscribe(data -> {
                    CommonUtils.sleep(100);
                    Log.it(data);
                }, error -> Log.e(error.getMessage()));
    }
}

public class onBackPressureDropSample {
    public static void main(String[] args) {
        CommonUtils.exampleStart();

        Flowable.range(1, 50000000)
                .onBackpressureDrop()
                .observeOn(Schedulers.computation())
                .subscribe(data -> {
                    CommonUtils.sleep(100);
                    Log.it(data);
                }, error -> Log.e(error.getMessage()));

        CommonUtils.sleep(20000);
    }
}
// 결과
... 생략
RxComputationThreadPool-1 | 13004 | value = 124
RxComputationThreadPool-1 | 13105 | value = 125
RxComputationThreadPool-1 | 13207 | value = 126
RxComputationThreadPool-1 | 13309 | value = 127
RxComputationThreadPool-1 | 13411 | value = 128

public class onBackPressureLatest {
    public static void main(String[] args) {
        CommonUtils.exampleStart();

        Flowable.range(1, 50000000)
                .onBackpressureLatest()
                .observeOn(Schedulers.computation())
                .subscribe(data -> {
                    CommonUtils.sleep(100);
                    Log.it(data);
                }, error -> Log.e(error.getMessage()));

        CommonUtils.sleep(20000);
    }
}
// 결과
RxComputationThreadPool-1 | 13416 | value = 127
RxComputationThreadPool-1 | 13518 | value = 128
RxComputationThreadPool-1 | 13622 | value = 50000000