안드로이드 RxJava #6 RxJava 연산자-(변형)

 

안녕하세요

이번 이간에는 RxJava에서 Observable을 변형하는 연산자에 대해 알아보겠습니다.

 

 

map

	Observable<Integer> intSrc = Observable.just(1, 2, 3);
        Observable<Integer> stringSrc = intSrc.map(i -> i *10);
        stringSrc.subscribe(item -> System.out.println(item));

 

발행되는 값에 대해 원하는 수식을 적용하거나 다른 타입으로 변환시킬 수 있다.

 

다음은 Integer값을 String으로 변형한 것이다.

	Observable<Integer> intSrc = Observable.just(1, 2, 3);
        Observable<String> stringSrc = intSrc.map(i -> String.valueOf(i));
        stringSrc.subscribe(item -> System.out.println(item+" 타입: "+item.getClass().getSimpleName()));

 

 

 

flatMap

	Observable<String> src = Observable.just("a", "b", "c");
        src.flatMap(str -> Observable.just(str + "1", str + "2"))
                .subscribe(System.out::println);

flatMap 연산자는 Observable을 또 다른 Observable로 변환시킨다. 그런 다음 변환시킨 Observable의 방출되는 아이템 또한 병합하여 다시 자체적으로 다시 아이템을 방출시킨다. 

말이 조금 어렵죠?ㅎㅎ

다시 설명하자면 첫 번째 Observable를 두 번째 Observable과 합친 의미라고 해석할 수 있습니다.

 

 

첫 번째 Observable 입니다.

Observable<String> src = Observable.just("a", "b", "c");

 

첫 번째 Observable를 flatMap을 이용해 두 번째 Observable과 병합할 겁니다.

src.flatMap(str -> Observable.just(str + "1", str + "2"))

 

이후 병합된 Observable을 바로 구독해 출력해주었습니다.

.subscribe(System.out::println);

 

flatMap은 for문과 비슷하게 동작합니다.

다음은 flatMap으로 1단부터 3단까지 구구단 프로그램을 작성해 보았습니다.

	Observable.range(1,3)                       //Observable.range(1,3) 의 값은 x
                .flatMap(x -> Observable.range(1,9) //Observable.range(1,9) 의 값은 y
                        .map(y-> String.format("%d*%d=%d",x,y,x*y)))
                        .subscribe(System.out::println);

 

 

 

buffer

 

	Observable.range(1, 9)
                .buffer(3)
                .subscribe(integers -> {
                    System.out.println("버퍼 데이터 발행");
                    for (Integer i : integers) {
                        System.out.println("#" + i);
                    }
                });

 

buffer 연산자는 Observable이 발행하는 아이템을 묶어서 List로 발행합니다.

 

	Observable<String> str = Observable.create(emitter -> {
            emitter.onNext("a");
            emitter.onNext("b");
            emitter.onNext("c");
            emitter.onNext("d");
            emitter.onNext("e");
            emitter.onError(new Throwable());
            emitter.onComplete();
        });
        str.buffer(2)
                .subscribe(System.out::println,
                        throwable -> System.out.println("에러로 인해 종료되었습니다"));

만약 에러가 있다면 이미 발행된 아이템들이 버퍼에 포함되더라도 버퍼를 발행하지 않고 에러를 전달합니다.

 

아래와 같은 결과를 볼 수 있습니다.

2022-11-08 17:28:51.399 6673-6673/com.psw.rxtest3 I/System.out: [a, b]
2022-11-08 17:28:51.399 6673-6673/com.psw.rxtest3 I/System.out: [c, d]
2022-11-08 17:28:51.399 6673-6673/com.psw.rxtest3 I/System.out: 에러로 인해 종료되었습니다

 

 

 

scan

 

	Observable.range(1,5)
                .scan((x,y)->{
                    System.out.print(String.format("%d+%d=",x,y));
                    return x+y;
                })
                .subscribe(System.out::println);

scan은 발행된 아이템을 다음번 발행되는 아이템의 첫 번째 인자로 전달하는 연산자이다.

scan은 최소 2개의 아이템이 필요하므로 첫 번째 아이템인 1은 그대로 출력된다.

두 번째 발행부터는 이전에 발행된 아이템(1)이 x이고 두 번째 발행된 아이템(2)은 y이다.

세 번째 발행부터는 이전에 연산된 아이템 3(return값)이 x이고 세 번째 발행된 아이템(3)이 y이다.

 

다음과 같은 결과가 나올 것이다.

2022-11-08 18:01:03.230 7434-7434/com.psw.rxtest3 I/System.out: 1
2022-11-08 18:01:03.230 7434-7434/com.psw.rxtest3 I/System.out: 1+2=3
2022-11-08 18:01:03.231 7434-7434/com.psw.rxtest3 I/System.out: 3+3=6
2022-11-08 18:01:03.231 7434-7434/com.psw.rxtest3 I/System.out: 6+4=10
2022-11-08 18:01:03.231 7434-7434/com.psw.rxtest3 I/System.out: 10+5=15

 

 

	Observable.just("a", "b", "c", "d", "e")
        .scan((x, y) -> x + y)
        .subscribe(System.out::println);

첫 번째 발행      x=a        y=없음

두 번째 발행      x=a        y=b

세 번째 발행      x=ab      y=c

네 번째 발행      x=abc    y=d

다섯 번째 발행  x=abcd  y=e

 

 

 

groupBy(추가 예정)

 

 

 

다음 시간에는 필터링하는 연산자에 대해 알아보겠습니다

수고하셨습니다!