[RxSwift] Observable

Observable과 Observer의 관계

결론부터 말하자면, 둘의 관계는 옵저버가 옵저버블을 구독하는 형태입니다. 이전글에서 설명했듯이, 옵저버블은 이벤트의 스트림을 처리하기 위해서 사용합니다. 옵저버블을 단일 항목이나 여러개 항목을 비동기적으로 처리하고 이를 방출(emit)하는데, 이때 옵저버가 옵저버블을 구독하고 있다가 배출되는 항목들에 즉각적으로 반응합니다.

Observable과 Observable의 변환

위 그림은 옵저버블과 옵저버블의 변환을 시각적으로 표현한 마블 다이어그램 입니다. 위 그림을 설명하자면 다음과 같습니다.

• 화살표들은 시간선을 나타냅니다.
• 위쪽에 있는 도형은 옵저버블이 방출하는 항목들을 나타냅니다.
• 위쪽 화살표의 세로선은 옵저버블이 모든 방출을 성공적으로 끝냈다는 것을 의미합니다.
• 도형과 연결되어 있는 점선과 점선이 가르키는 상자는 옵저버블에 'flip'이란 변환이 적용됐다는 것을 의미합니다.
• 아래쪽에 있는 도형은 변환이 적용된 결과물을 의미합니다.(위쪽 도형이 flip된 형태로 나온 것을 확인할 수 있습니다.)
• 아래쪽에 있는 X는 옵저버블이 비정상적으로 종료되었다는 것을 의미합니다.

Observable과 Observer의 의미와 동작방식

옵저버블이란, 데이터를 검색하고 변환하는 메커니즘입니다. 배열, 리스트, 데이터베이스, 파일 등 다양한 소스에서 데이터를 검색할 수 있으며 데이터를 필터링하거나, 정렬하는 작업을 수행할 수 있습니다.

옵저버블은 옵저버가 준비되어 있어야지만 동작을 시작합니다. 일반적으로 프로그램이 명령어가 입력된 순서대로 순차적으로 작동하는 것과 달리, 옵저버블은 병렬적으로 동작합니다. 옵저버블은 옵저버가 준비되어 있어야만 실행가능합니다. 일부 옵저버블은 항목을 즉시 방출하고, 일부 옵저버블은 일정 시간이 지난 후에 항목을 방출하기 때문에 옵저버는 언제든지 결과를 캡처할 수 있도록 준비되어 있어야 하기 때문입니다.

여기서 결과를 캡쳐한다는 것은, 옵저버가 옵저버블이 방출하는 항목들을 수신하고 처리한다는 것을 의미합니다. 옵저버가 항목을 캡처하는 방법은 구현에 따라 다릅니다. 일반적으로 옵저버는 항목을 수신하면 필요한 처리를 수행하고, 결과를 저장하거나 다른 작업에 활용합니다. 이러한 작업은 비동기적으로 처리될 수 있으며, 옵저버는 필요에 따라 결과를 다른 시스템에 전달하거나, 다른 옵저버블과 결합하여 새로운 옵저버블을 생성할 수도 있습니다.

다음은 rxswift를 사용하여 배열에서 짝수를 필터링하여 옵저버에게 전달하는 코드입니다.

import RxSwift

let data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]

let observable = Observable.from(data)
    .filter { \$0 % 2 == 0 }

observable.subscribe(
    onNext: { value in
        print(value)
    },
    onCompleted: {
        print("complete")
    }
)

Reacitve Programming의 장단점

장점은, 서로 의존하지 않는 여러 개의 작업이 있을 때, 각 작업이 완료될 때마다 기다리는 대신 모든 작업을 한 번에 시작할 수 있다는 것입니다. Reactive programming을 한다면 번들(bundle)의 모든 작업은 번들 내 가장 긴 작업이 완료되는 데까지 걸리는 시간만큼만 걸립니다. 또한 비동기적으로 실행되는 작업을 처리하면서도, 콜백 지옥과 같은 코드의 복잡도를 줄이고, 높은 생산성과 유지보수성을 제공할 수 있습니다.

단점은, 리액티브 스트림을 처리하는 데 필요한 연산자의 수가 많아지기 때문에 코드의 복잡도가 증가한다는 것입니다. 또한 리액티브 프로그래밍은 비동기적으로 실행되는 작업을 처리하는 데 적합하지만, 동기적으로 실행되는 작업을 처리하기에는 부적합합니다.

Ordinary Programming과 Reactive Programming의 차이점

일반적인 명령형 프로그래밍의 동작 방식은 다음과 같습니다.

1. 메소드를 호출합니다.
2. 해당 메소드의 반환값을 변수에 저장합니다.
3. 변수와 그 값으로 다음 작업을 수행합니다.

리액티브 프로그래밍의 동작 방식은 다음과 같습니다.

1. 비동기 호출의 결과를 처리하는 작업을 수행하는 메소드를 정의합니다. 이 메소드는 옵저버의 일부입니다.
2. 비동기 호출 자체를 옵저버블로 정의합니다.
3. 옵저버를 해당 옵저버블에 구독(subscribe)하여 연결합니다. 이는 옵저버블의 작업을 시작합니다.
4. 이후 다른 작업을 계속합니다. 비동기 호출이 완료되면, 옵저버의 메소드가 호출되어 반환된 값 또는 값들 (옵저버블이 방출한 아이템들) 을 처리합니다.

코드로 두 프로그래밍의 동작방식을 비교하면 다음과 같습니다.

// 명령형 프로그래밍
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
List<Integer> squaredNumbers = new ArrayList<>();
for (Integer number : numbers) {
    squaredNumbers.add(number * number);
}
System.out.println(squaredNumbers);

// 리액티브 프로그래밍
List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);
Observable<Integer> numberObservable = Observable.from(numbers);
Observable<Integer> squaredNumberObservable = numberObservable.map(number -> number * number);
squaredNumberObservable.subscribe(System.out::println);

명령형 프로그래밍의 경우에는 for-each 루프를 이용하여 각 요소를 제곱한 후, 새로운 리스트에 추가하는 방식으로 동작합니다. 이러한 방식은 요소의 개수가 많아질수록 코드의 길이와 복잡도가 증가할 가능성이 높습니다.

반면에, 리액티브 프로그래밍에서는 함수형 프로그래밍의 개념을 활용하여, 각 요소를 제곱하는 함수를 정의하고, 이를 Observable의 map() 연산자에 전달합니다. 이후, subscribe() 메소드를 이용하여 옵저버를 등록하면, 각 요소가 제곱된 결과가 출력됩니다. 이처럼, 리액티브 프로그래밍은 명령형 프로그래밍보다 간결하고 가독성이 높은 코드를 작성할 수 있다는 장점이 있습니다. 또한, 비동기적으로 실행되는 작업을 처리하는 데 높은 생산성과 유지보수성을 제공할 수 있습니다.

Observer의 메소드

onNext, onCompleted, and onError

옵저버블과 옵저버를 연결하기 위해서는 subscribe() 메소드를 호출해야 합니다. 이때, 옵저버는 onNext, onError, onCompleted 메소드 중 일부를 구현하게 됩니다.

onNext

옵저버블이 아이템을 방출할 때마다 호출되는 메소드입니다. 이 메소드는 방출된 아이템을 파라미터로 받습니다. 옵저버블은 onNext 메소드를 통해 0개 이상의 아이템을 방출할 수 있으며, 이후에 onCompleted나 onError 메소드 중 하나를 호출하여 작업이 완료되었음을 알립니다.

onError

옵저버블이 에러를 발생시키면 호출되는 메소드입니다. 이 메소드는 어떤 에러가 발생했는지를 파라미터로 받습니다. 에러 발생 이후에는 onNext나 onCompleted 메소드가 호출되지 않습니다.

onCompleted

옵저버블이 모든 아이템을 방출한 후에 호출되는 메소드입니다. 이 메소드는 파라미터를 받지 않습니다.

구독 해제(Unsubscribing)

Java, Kotlin, Scala, JavaScript, Swift 등의 언어에서는 Subscriber라는 특수한 옵저버 인터페이스가 있으며, 이 인터페이스는 unsubscribe() 메소드를 구현합니다. 이 메소드를 호출하면, 해당 Subscriber가 현재 구독 중인 모든 옵저버블에 더 이상 관심이 없다는 것을 나타낼 수 있습니다. 그러면, 이러한 옵저버블은 (다른 관심 있는 옵저버가 없다면) 새로운 아이템을 방출하지 않을 수 있습니다.

구독 해제 작업은, 옵저버가 구독한 옵저버블에 적용되는 연산자 체인의 모든 링크에게 전파됩니다. 이를 통해, 연산자 체인의 각 링크가 아이템을 방출하는 것을 중지합니다. 하지만, 이러한 중지 작업이 즉시 일어나지는 않으며, 옵저버가 없어졌더라도 일정 시간 동안 아이템을 생성하고 방출할 수 있습니다.

Observable의 방출 시점

Observable이 아이템을 방출하기 시작하는 시점은 Observable의 종류에 따라 다릅니다.

Hot Observable은 생성되자마자 아이템을 방출하기 시작할 수 있으므로, 이후에 해당 Observable에 구독하는 옵저버는 시퀀스 중간 어딘가부터 관찰을 시작할 수 있습니다. 다시 말해서, 구독했을 때 이벤트 시퀀스를 처음부터 관찰하지 못할 수 있습니다.

반면 Cold Observable은 옵저버가 구독하기 전까지 아이템을 방출하지 않으며, 따라서 해당 옵저버는 시퀀스의 전체를 처음부터 보게 됩니다. 일반적인 옵저버를 의미합니다.

ReactiveX의 일부 구현체(RxJava, RxJS)에서는 Connectable Observable이라는 것도 존재합니다. 이 때 Observable은 구독자가 있는지 여부와 상관없이 Connect 메소드가 호출될 때까지 아이템을 방출하지 않습니다.

다음 포스팅 소개

지금까지 옵저버 패턴을 확장한 옵저버와 옵저버블의 역할과 동작 방식에 대해 알아보았습니다. 다음 포스팅에서는 Rx가 다양한 동작을 수행할 수 있게 도와주는 Operator 에 대해 알아보겠습니다.