[RxJava2] RxJava 에서의 디버깅 : doOnNext(), doOnComplete(), doOnError(), doOnEach(), doOnSubscribe(), doOnDispose(), etc.

공감 및 댓글은 포스팅 하는데  아주아주 큰 힘이 됩니다!! 포스팅 내용이 찾아주신 분들께  도움이 되길 바라며 더 깔끔하고 좋은 포스팅을  만들어 나가겠습니다^^

이번 포스팅에서는 디버깅에 대해 공부합니다.

일반적으로 코딩하면서 코드 중간 중간에 로그를 넣는 경우가 많은데요.

그 이유는 중간 중간 순서가 제대로 맞지 않거나하는 잘못된 경우를

위해서 입니다.

BUT RxJava에는 로그를 넣을 공간이 없는데요.

대신에 부수효과를 일으키는 

doOnXXX 함수들을 제공합니다.

doOnNext(), doOnComplete(), doOnError()

Observable의 알림 이벤트에 해당.

doOnNext()

Observable이 어떠한 데이터를 발행했을 때 이벤트 발생.

doOnError()

중간에 에러가 발생했을 때 이벤트 발생.

doOnComplete()

Observable이 모든 데이터를 발행하면 이벤트 발생.

예제1: 에러가 발생하지 않고 정상적으로 모든 발행이 끝났을 경우

package rx.java.chapter07;
 
import io.reactivex.Observable;
 
public class DoOnXXX {
 
    public static void main(String[] args) {
        Observable<String> src = Observable.just("123", "456", "789");
        src.doOnNext(DoOnXXX::print)
        .doOnComplete(() -> DoOnXXX.print("complete()"))
        .doOnError(e -> DoOnXXX.print(e.getMessage()))
        .subscribe();
    }
    
 
    public static void print(String str) {
        System.out.println(Thread.currentThread().getName() + "|Debug : " + str);
    }
    
}
 

모든 데이터가 정상적으로 발행되므로 doOnNext() 와 doOnComplete()가 호출됨.

예제2: 중간에 에러가 발생했을 경우

package rx.java.chapter07;
 
import io.reactivex.Observable;
 
public class DoOnXXXWithError {
 
    public static void main(String[] args) {
 
        Observable<Integer> src = Observable.just(100, 10, 1, 0);
        src.map(key -> 1000 / key)
        .map(val -> String.valueOf(val))
        .doOnNext(DoOnXXXWithError::print)
        .doOnComplete(() -> DoOnXXXWithError.print("complete()"))
        .doOnError(e -> DoOnXXXWithError.print(e.getMessage()))
        .subscribe(DoOnXXXWithError::print);
        
    }
    
    public static void print(String str) {
        System.out.println(Thread.currentThread().getName() + "|Debug : " + str);
    }
    
}
 

모든 숫자는 0으로는 나눌 수 없기 때문에 doOnNext() 이벤트가 발생하다가

0이 key로 들어오면 doOnError() 이벤트 발생.

doOnEach()

위 3가지 알림 이벤트를 한 번에 처리합니다.

package rx.java.chapter07;
 
import io.reactivex.Observable;
import io.reactivex.Observer;
import io.reactivex.disposables.Disposable;
 
public class DoOnEachEx {
    public static void main(String[] args) {
        
        Observable<String> src = Observable.just("123", "436", "151");
        src.doOnEach(notification -> {
            if(notification.isOnNext()) DoOnEachEx.print(notification.getValue());
            if(notification.isOnComplete()) DoOnEachEx.print("complete");
            if(notification.isOnError()) DoOnEachEx.print(notification.getError().getMessage());
        }).subscribe(System.out::println);
                
        DoOnEachEx.useObserverDoOnEach();
        
    }
 
    public static void print(String str) {
        System.out.println(Thread.currentThread().getName() + "|Debug : " + str);
    }
    
    public static void useObserverDoOnEach() {
        Observable<String> src2 = Observable.just("123", "436", "151");
        src2.doOnEach(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {
                // TODO Auto-generated method stub                
            }
 
            @Override
            public void onNext(String t) {
                DoOnEachEx.print(t);                
            }
 
            @Override
            public void onError(Throwable e) {
                DoOnEachEx.print(e.getMessage());
            }
 
            @Override
            public void onComplete() {
                DoOnEachEx.print("complete");                
            }            
        }).subscribe(System.out::println);
    }    
}
 

doOnSubscribe()

Observable을 구독했을 때 이벤트 발생.

doOnDispose()

Observable 구독을 해지했을 때 이벤트 발생.

doOnLifeCycle()

doOnSubscribe() + doOnDispose()

doOnTerminate()

Observable이 끝나는 조건인 onComplete() 이나 onError()이 호출되면 이벤트 발생.

package rx.java.chapter07;
 
import java.util.concurrent.TimeUnit;
 
import io.reactivex.Observable;
import io.reactivex.disposables.Disposable;
 
public class DoOnEtc {
 
    public static void main(String[] args) {
 
        String threadName = Thread.currentThread().getName();
        try {
            //doOnSubscribe() and doOnDispose()
            Observable<String> src = Observable.just("1", "3", "5", "7", "9")
                    .zipWith(Observable.interval(100L, TimeUnit.MILLISECONDS), (a, b) -> a)
                    .doOnSubscribe(d -> System.out.println(threadName + " :subscribe"))
                    .doOnDispose(() -> System.out.println(threadName + " :dispose"));
            Disposable disposable = src.subscribe(str -> System.out.println(threadName + " :" + str));
            Thread.sleep(200);
            disposable.dispose();
            Thread.sleep(300);
            System.out.println("===============");
            
            //doOnLifeCycle
            Observable<String> src2 = Observable.just("1", "3", "5", "7", "9")
                    .zipWith(Observable.interval(100L, TimeUnit.MILLISECONDS), (a, b) -> a)
                    .doOnLifecycle(d -> System.out.println(threadName + " :subscribe"),
                            () -> System.out.println(threadName + " :dispose"));
            disposable = src2.subscribe(str -> System.out.println(threadName + " :" + str));
            Thread.sleep(200);
            disposable.dispose();
            Thread.sleep(300);
            System.out.println("===============");
            
            //doOnTermitate()
            Observable<String> src3 = Observable.just("1", "3", "5", "7", "9");
            src3.doOnTerminate(() -> System.out.println(threadName + " :terminate"))
            .doOnComplete(() -> System.out.println(threadName + " :complete"))
            .doOnError(e -> System.out.println(threadName + " :" + e.getMessage()))
            .subscribe(str -> System.out.println(threadName + " :" + str));
        } catch (Exception e) {
            // TODO: handle exception
        }
    }
 
}