[RxSwift] Observable, Observer

RxSwift를 학습하며 스스로 정리하고 문답하는 게시글.
(틀린 내용이 있다면 지적해주시면 감사드리겠습니다.)

먼저 observable 은 값을 관찰할 수 있는 형태를 가진다. 100이라는 값을 방출할 수 있는 observable 을 먼저 생성해보자.

// 100이라는 값을 가지는 관찰 가능한 형태
let observable = Observable<Int>.just(100) 

observable 형태의 변수를 생성했지만, 이 상태에서는 observable의 값을 관찰할 수단이 존재하지 않으며 해당 값을 받아올 수 있는 수단도 존재하지 않는다.

observable 을 관찰하고 데이터를 받아오기 위해서는 observer 가 필요하다. 우린 observer 를 만들기 위해서 subscribe(onNext:, onError:..) 함수를 사용할 것이다.

갑자기 observer 가 필요하다고 하면서 함수를 호출하는 것에 의문을 가질수도 있지만, 그 의문의 해결책은 바로 subscribe(onNext:, onError:..) 함수에 있다.

이 함수가 observer를 생성하기 때문이다.

observable.subscribe( onNext : { event in
    print(event)
}, onError: { error in
    print(error)
}, onCompleted: {
    print("completed")
})

// ---- 출력 ----
100
completed

ObservableType에 채택되어 있는 subscribe(onNext:, onError:..) 함수를 통해서 observable의 값을 관찰하고 전달받을 수 있다. 바로 observer를 생성하기 때문이다. 어떻게 이런 일이 가능한걸까?

의문을 해결하기 위해서 subscribe(onNext:, onError..) 함수의 구현부를 뜯어보자.

public func subscribe(onNext: ((Element) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil)
        -> Disposable {
        	// (1) AnonymousObserver 생성
            let observer = AnonymousObserver<Element> { event in
                
                #if DEBUG
                    synchronizationTracker.register(synchronizationErrorMessage: .default)
                    defer { synchronizationTracker.unregister() }
                #endif
                
                switch event {
                case .next(let value):
                    onNext?(value)
                case .error(let error):
                    if let onError = onError {
                        onError(error)
                    }
                    else {
                        Hooks.defaultErrorHandler(callStack, error)
                    }
                    disposable.dispose()
                case .completed:
                    onCompleted?()
                    disposable.dispose()
                }
            }
            return Disposables.create(
            	// (2) Observable에 구독하는 과정.
                self.asObservable().subscribe(observer),
                disposable
            )
    }

위의 코드 subscribe 의 함수 구현체에서 AnonymousObserver 를 생성하여 Observable 에 구독을 하는 것을 볼 수 있다. 이것이 Observable 에 명시적으로 Observer를 만들고 직접 연결하지 않아도 우리가 데이터의 변화를 관찰할 수 있는 이유다.

즉, 우리가 명시적으로 Observer를 생성해서 Observable에 구독을 하는 형태가 아니라, subscribe() 메소드를 사용했을 때 메소드 내부 구현에서 AnonymousObserver 를 만들어주는 것을 알 수 있다.

subscribe(onNext:.., onError:..) 함수의 사용법을 더 살펴보자.

observable.subscribe( onNext : { event in
    print(event)
}, onError: { error in
    print(error)
}, onCompleted: {
    print("completed")
})

// ---- 출력 ----
100
completed

• onNext → 방출하는 이벤트를 전달받음.
• onError → 예기치 못하게 발생한 에러를 전달받음.
• onCompleted → 이벤트가 종료되어 스트림이 끊어졌을 때 전달받음.

스트림이란 쉽게 생각하면 observable과 observer 의 연결 관계라고 생각할 수 있다. 스트림이 끊어진다는 것은 더 이상 observable 의 값을 발행받을 수 없게 된다는 것을 의미한다.

결론은 observable과 subscribe는 거의 한 쌍이라고 볼 수 있다. 관찰 가능한 형태의 비동기 이벤트에 구독을 통해서 이벤트를 전달받을 수 있기 때문이다. observable 은 observer 가 존재하지 않는다면 아무런 의미가 존재하지 않으며, subscribe(onNext:..) 을 통해 구독이 됐을 때에 이벤트가 방출될 수 있다.

참고자료
https://babbab2.tistory.com/185
https://sujinnaljin.medium.com/rxswift-observer-fdc8d2772d6c
https://www.youtube.com/watch?v=iHKBNYMWd5I&t=10131s