RxJava Observable 변형하기

Observable 변형

RxJava에서는 아이템을 그대로 발행하는 것 뿐 아니라, 발행되는 아이템을 변환하여 다른 아이템으로 변경할 수 도 있습니다.

Map()

Map은 발행되는 아이템을 변환하는 가장 기본적인 방법으로 발행된 아이템에 원하는 수식을 적용하거나 다른 방법으로 변환시킬 수 있습니다.

map을 통해 데이터 각각에 10을 곱하고 이를 String으로 반환해줍니다.

fun main() {
	val intStream = Observable.just(1,2,3)
    val stringStream = intStream.map {
     (it * 10).toString()
 }
 
 	stringStream.subscribe {
    	println("[${it.javaClass}] : $it")
    }
}

[class java.lang.String]: 10 
[class java.lang.String]: 20 
[class java.lang.String]: 30 

flatMap()

flatMap()의 경우 Observable에서 발행된 데이터들 각각에 대해서, map처럼 다른 값으로 변환하여 이를 또 다른 Observable 형태로 반환해주는 역할을 합니다. 반환 형태가 결국 Observable 이기 때문에 새롭게 생성된 데이터들이 하나씩 순서대로 발행되는 Flatten 된 형태입니다.

조금 더 쉽게 설명하자면 flatMap()은 가공한 데이터를 그대로 return 하지 않고, Observable들로 return한 다음 return 값들을 merge해서 다시 Observablefh return 해줍니다.

fun main() {
	val stream = Observable.just("A","B","C")
    
    stream.flatMap {
    	Observable.just(
        	"[$it] - 1",
            "[$it] - 2"
            )
        }.subscribe{
        	println(it)
        }
   }
            

[A] - 1
[A] - 2
[B] - 1
[B] - 2
[C] - 1
[C] - 2

flatMap()은 Observable 작업을 여러 번 연계하여 사용할 때 유용합니다.

scan()

scan()은 발행된 데이터를 다음에 발행되는 데이터의 첫 번째 인자로 전달해주는 메서드입니다.

fun main() {
	Observable.just("A","B","C","D","E")
    .scan { x:String , y: String ->
    	x * y
    }
    .subscribe {
    	println(it)
    }
}

A
AB
ABC
ABCD
ABCDE

데이터를 발행하면 해당 데이터를 다음에 발행되는 데이터의 첫 번째 인자(x)로 전달합니다.

buffer()

buffer의 경우 설명보다는 간단한 코드로 설명하겠습니다.

fun main() {
	Observable.range(0,8)
    .buffer(3)
    .subscribe{
    	println("버퍼 시작점")
        it.forEach {
        	println(it)
        }
    }
}

버퍼 시작점
0
1
2
버퍼 시작점
3
4
5
버퍼 시작점
6
7

다음과 같이 결과가 출력되게 됩니다. 만약을 buffer(4)를 하게된다면 0부터 4까지 출력할 것입니다. 이처럼 buffer()는 List형태로 데이터들을 묶어주는 역할을 합니다.

groupBy()

groupBy() 메서드는 어떤 기준으로 단일 Observable을 여러 개로 이루어진 Observable 그룹으로 만드는 메서드입니다.

fun main(){
    val objs = arrayOf("6", "4", "2-T", "2", "6-T", "4-T")
    var source = objs.toObservable().groupBy {it-> getShape(it)}

    source.subscribe{obj->
        obj.subscribe { value -> println("GROUP: ${obj.getKey()} value:+$value") }
    }

}

private fun getShape(obj: String): String {
    if (obj == "") {
        return "NO-SHAPE"
    }
    return if (obj.contains("-T")) {
        "TRIANGLE"
    } else "BALL"
}

GROUP: BALL value: 6
GROUP: BALL value: 4
GROUP: TRIANGLE value: 2-T
GROUP: BALL value: 2
GROUP: TRIANGLE value: 6-T
GROUP: TRIANGLE value: 4-T

grouBy를 쓸 경우 그 항목에 대한 key값을 구분할 수 있고 그걸로 group화 할 수 있습니다.
또한 특정 그룹만 처리하고 싶다면 filter를 사용할 수 도 있습니다.

fun main(){
    val objs = arrayOf("6", "4", "2-T", "2", "6-T", "4-T")
    var source = objs.toObservable().groupBy {it-> getShape(it)}

    source.subscribe{obj->
        obj.filter {it->obj.key.equals("BALL")}
            .subscribe { value -> println("GROUP: ${obj.getKey()} value: $value") }
    }

}

private fun getShape(obj: String): String {
    if (obj == "") {
        return "NO-SHAPE"
    }
    return if (obj.contains("-T")) {
        "TRIANGLE"
    } else "BALL"
}

GROUP: BALL value: 6
GROUP: BALL value: 4
GROUP: BALL value: 2