[RxSwift] 의식의 흐름대로 Observable 파헤쳐보기

여훈님은 RxSwift를 코드레벨에서 분석해보신 적이 있나요?

최근 면접에서 Rx에 대한 질문을 받은 적이 있습니다. 그때 무슨 말을 했는지 잘 기억이 안나지만.. Rx에 대해서 이해를 제대로 하지 못하고 있었음을 깨닫고 코드레벨로 한 번 뜯어보려고 합니다!

Observable

먼저 가장 많이 사용하는 Observable의 파일부터 찾아가보았습니다.

일단 클래스이고 내부에 subscribe와 asObservable 메서드가 있네요. 일단 asObservable은 자기 자신을 반환하고 있고, subscribe는 rxAbstractMethod()라는 함수의 호출 외에는 다른 코드를 가지고 있지 않습니다.

그래서 rxAbstractMethod()를 알기위해 루트 디렉토리에 있는 Rx.swift에 찾아가보았는데요,

이렇게 정의가 되어있네요. 이 코드가 들어있는 메서드는 추상 메서드이고, 이 코드가 불리면 런타임 에러를 냅니다. 따라서 subscribe는 Observable 클래스를 상속하는 각 여러 Observable의 클래스들이 직접 구현하고 있어야 한다는 것을 유추할 수 있었습니다.

더 들어가기 전에 Observable에는 ObservableType이 붙어있으니 이 부분부터 먼저 알아볼게요.

ObservableType

ObservableType은 프로토콜이네요. 내부에는 subscribe에 대한 인터페이스가 정의되어있습니다. 오키오키 subscribe는 Observable이 들고 있는게 아니라 ObservableType을 채택하고 메소드를 구현한 것이었네요!

subscribe에 대해 길게 적혀있는 주석을 해석해볼게요.

/** Subscribes `observer` to receive events for this sequence.
옵저버가 이 시퀀스로부터 이벤트를 받을 수 있도록 구독합니다.

### Grammar **Next\* (Error | Completed)?**
sequences can produce zero or more elements so zero or more `Next` events can be sent to `observer`
시퀀스들은 0개 이상의 아이템을 Next 이벤트를 통해 옵저버에 보낼 수 있습니다.

once an `Error` or `Completed` event is sent, the sequence terminates and can't produce any other elements
Error나 Completed 이벤트가 전송되면, 이 시퀀스는 종료되고 다른 아이템을 생산할 수 없습니다.

It is possible that events are sent from different threads, but no two events can be sent concurrently to `observer`.
이벤트들이 서로 다른 스레드에서 전송되는 것은 가능하지만, 두 이벤트가 동시에 옵저버에 전송될 수는 없습니다.

### Resource Management
When sequence sends `Complete` or `Error` event all internal resources that compute sequence elements will be freed.
시퀀스가 Complete이나 Error 이벤트를 보내면, 시퀀스에서 사용하던 모든 리소스들은 해제됩니다.

To cancel production of sequence elements and free resources immediately, call `dispose` on returned subscription.
시퀀스가 데이터를 생산하는 것을 종료하고 리소스를 곧바로 해제하게 하려면 반환된 구독객체에 dispose를 호출할 수 있습니다.

- returns: Subscription for `observer` that can be used to cancel production of sequence elements and free resources. */
Observer를 위한 구독객체가 반환됩니다. 이 객체는 시퀀스를 취소하고 리소스를 해제할 수 있도록 합니다.

Subscribe에 대한 설명이었는데요, 결국 subscribe는 인자로 받은 옵저버를 시퀀스에 연결하고 시퀀스를 취소할 수 있는 subscription 객체인 Disposable을 반환하는 메서드입니다.

추가적으로 ObservableType에 대한 Extension도 있으니 한번 봅시다.

asObservable을구현하고 있는데요, Observable타입이 가진 create 메서드를 불러서 전달된 인자를 subscribe 메서드의 인자로 전달하고 그 결과를 그대로 반환합니다.

create 메서드의 정의는 RxSwift/Observables/Create.swift 에서 찾을 수 있었어요.

create는 타입메서드 였습니다. 그래서 위에서도 Observable.create로 호출 할 수 있었죠. 인자로는 escaping closure를 하나 받는데요, subscribe의 정의와 동일합니다. 주석을 보니 각 옵저버블의 subscribe메서드 구현체를 인자로 전달하라고 하네요. 그리고 반환값으로는 시퀀스를 반환하는데, 전달된 subscribe 메서드를 사용하는 시퀀스가 반환됩니다.

무슨 소린지 잘 모르겠어서 실제로 반환되는 타입인 AnonymousObservable을 찾아보기로 했어요..

AnonymousObservable

AnonymousObservable도 같은 Create.swift파일 안에 정의되어 있었습니다. 먼저 typealias가 눈에 띄는데요, subscribe의 파라미터와 형식이 같습니다! 구독이 있을 때 어떻게 처리할지 결정하는 클로저로 사용하는 것 같아요. 그리고 초기화 시점에서 이 클로저를 인자로 받기도 하네요.

아까 create 메서드에서 AnonymousObservable 인스턴스를 만들면서 subscribe 메서드의 구현체를 인자로 전달했는데요, 이 구현체가 subsribeHandler로 사용되는 구조인 것을 알 수 있습니다.

그럼 제일 중요해보이는 run 메서드를 볼까요?

run method도 내부는 꽤 단순한데요, 처음보는 애들이 많네요..

 

먼저 인자로는 ObservableType을 채택하는 Observer 타입들, Cacelable 타입의 데이터가 전달되고, sink와 subscription이라는 이름을 가진 튜플로 각각 Disposable의 인스턴스가 담겨 반환됩니다.

sink에 저장되는 값은 AnonymousObservableSink의 인스턴스입니다. 그래서 또 찾아가봤어요.

오 여기선 그래도 익숙한 함수가 하나 보입니다. on 메서드가 있네요..! on이 호출되면 인자로 이벤트가 들어오고 이 이벤트가 next라면 forwardOn메서드에 이벤트를 담아 보내고, error나 completed라면 forwardOn에 이벤트를 보낸 후에 시퀀스를 dispose 시키고 있습니다.

그리고 run 메서드에서는 Parent 타입(AnonymousObservable<Element>)을 받아 자기자신을 AnyObserver로 캐스팅한 뒤에 Parent가 가진 subscribe 메서드에 인자로 전달하며 실행합니다. 그리고 그 결과인 Disposable을 반환해요.

forwardOn은 어디에 있을까요? AnonymousObservableSink의 선언부를 보니 Sink<Observer>와 OberverType 두 가지를 채택하고 있는데, 슈퍼클래스인 Sink 클래스를 찾아가 보았습니다.

forwardOn는 이벤트를 받아서 자신이 가진 observer의 on 메서드를 호출하는 메서드였습니다.

의식의 흐름 정리하기 - 1

자 그럼 지금 약간 맥락을 잃은 것 같으니까.. (저도 스크롤을 계속 올렸다 내렸다 하고 있어요..) 지금까지의 의식의 흐름을 정리해볼게요.

1. 먼저 우리는 가장 많이 사용되는 Observable이 어떻게 구성되어있는지 알고싶었어요. 그래서 코드를 들여다 봤고 내부에는 추상 메서드인 subscribe와 asObservable의 기본 구현체를 발견했죠. 그리고 실제로 subscribe를 인터페이스로 제공하는 프로토콜인 ObservableType이 존재했어요. 모든 Observable들은 이 인터페이스에 대한 구현체를 각각 가지고 있는거죠.

2. asObservable에서는 Observable.create라는 메서드가 사용되고 있었는데요, 자주 사용되는 create 메서드를 알아보기 위해서 create의 구현 메서드를 찾아보았습니다.

3. 근데 막상 create를 열어보니까 별거 없었어요.. 전달받은 subscribe 메서드의 구현체를 인자로 넘겨 AnonymousObservable 인스턴스를 반환하고 있었죠.

4. 그래서 AnonymousObservable을 열어보았습니다. 먼저, 인자로 전달받은 subscribe 메서드를 subscribeHandler라는 이름의 typealias로 가지고 있었습니다.

5. 그리고 내부에 딱 하나 있던 메서드는 run 메서드였어요. 이 메서드는 sink 와 subscription이라는 이름의 두 Disposable을 튜플로 묶어서 반환해주는 함수였습니다.

6. 이제는 sink랑 subscription이 어떻게 만들어지는지 궁금해졌죠. sink를 봤더니 AnonymousObservableSink의 인스턴스를 저장하고 있었습니다. 그리고 subscription은 이 AnonymousObservableSink의 메서드 중 run을 호출한 결과를 저장하고 있었어요. run을 호출할 때는 인자로 받은 옵저버가 아니라 현재 인스턴스 자기 자신을 전달하고 있었죠.

7. AnonymousObservableSink 안에는 어딘가 익숙한 on 메서드가 있었어요. 이 메서드는 전달된 이벤트에 따라서 next가 전달되면 forwardOn이라는 메서드를 호출하고 있었고, error나 complete 이벤트에는 forwardOn 이후에 dispose까지 호출해서 시퀀스를 닫아주고 있었습니다.

8. 그럼 forwadOn도 확인해봐야겠죠? fowardOn은 부모클래스인 Sink<Observer>에 있는 메서드였고, 자신의 멤버 변수인 observer의 on을 호출해주고 있었습니다.

오 무슨 소린지 모르겠죠? 저도 그래요..

run 다시보기

어디에서 흐름을 놓쳤는지 곰곰히 생각해보다가

AnonymousObservableSink에서 run 메서드 부분을 이해하지 못했다는 생각이 들었어요. parent 를 전달하는 것 까지는 그렇다 치는데.. subscribeHandler는 분명 클로저를 저장하는데 저런 코드가 어떻게 나오지..? 라는 생각이요.

그래서 AnyObserver를 보기로 했습니다.

내부에는 두 종류의 생성자가 존재하고 있는데요, 하나는 EventHandler 클로저를 받는 생성자이고, 하나는 ObserverType 을 받는 클로저가 있어요.

아까 run 메서드에서는 self를 전달하고 있었으니까 ObserverType을 채택하고 있는 AnonymousObservableSink 인스턴스가 전달이 되었겠죠. 그럼 두번째 생성자에서 어떤 일이 일어나는지 봐야겠네요.

옵저버의 이벤트를 받는 클로저를 자신의 EventHandler로 할당하고 있었습니다.

다시 이 코드로 돌아가면

run 메서드는 여기서 subscription을 만들기 위해 호출이 되는데요, run 호출과 함께 전달된 Observer로 sink라는 변수의 인스턴스를 만들고 자기자신을 sink.run의 인자로 전달해 subscriptionHandler를 실행한 결과를 반환하고 있었습니다.

그럼 결국에는 생성된 AnonymousObservable에서 run을 호출하면 호출과 함께 전달된 옵저버에 대한 AnonymousObservableSink 인스턴스를 만듭니다. 그리고 이 인스턴스에 대한 run을 호출하면서 자기자신을 전달하면 자신이 가지고 있던 subscribe 메서드의 구현체를 호출하면서 인자의 옵저버로 AnonymousObservableSink인스턴스를 전달해주게 됩니다.

오 뭔가 좀 느낌이 오지 않나요??

코드랑 같이 보기

Observable<String>.create({ (observer) -> Disposable in 
	observer.onNext("1") 
    return Disposables.create() 
})

create를 사용하면 이런 코드를 만들 수 있었죠?

그럼 아까 봤었던 이 코드가 실행될 텐데요, Eelemt는 런타임에 String 으로 정해질테고, 내부 구현대로 AnonymousObservable이 실행되겠네요.

그리고 인자로 전달되는 클로저는 우리가 만들었던

{ (observer) -> Disposable in 
	observer.onNext("1") 
	return Disposables.create() 
}

이게 되겠습니다! 우리가 모르고 있었지만 observer라고 만들었던 파라미터가 AnyObserver를 의미하고 있었네요. 그래서 내부에서 onNext, onError 같은 메서드들을 사용할 수 있었던거예요!

자 이제 이 클로저와 함께 AnonymousObservable의 인스턴스가 만들어집니다.

인스턴스를 만들면서 인자로 전달되는 subscribeHandler는 우리가 정의한 그 클로저 그대로겠죠!

자 그럼 대망의 저 run 메서드가 언제 호출되는지 알아야되는데, 알고보니까 AnonymousObservable이 상속하고 있는 클래스가 있더라구요.. 바로 Producer 입니다!

잠깐.. Producer부터 보자!

Producer는 Observable을 상속하고 있고, RxSwift의 모든 Operator들은 이 클래스를 상속하고 있어요! 위에 있었던 run 메서드도 이 클래스를 상속받아서 오버라이딩한 메서드였습니다.

이제 run이 언제 불리는지 알 수 있게되었는데요, 구현된 subscribe의 내부를 보겠습니다.

subscribe가 호출되면 어떤 스케줄러이든지 상관없이 하는 일들이 있는데요, SinkDisposer 인스턴스를 만들고 run 메서드를 실행한 결과를 저장한 sinkAndSubscription을 만든 뒤에 disposer에 sink와 subscription을 넣어주고 반환해줍니다.

아까 run 메서드의 반환타입이 sink와 subscription이라는 이름의 두 Disposable이었죠! 여기서 퍼즐이 맞춰지네요..!

그럼 SinkDisposer가 뭔지 또 봐야겠죠?

다른건 복잡하니까 안보고 setSinkAndSubscription에서 결국 하는 일은 현재 인스턴스의 sink와 subscription에 Disposable들을 넣어주는거예요.

뭐하러 이렇게 하나 했더니 SinkDisposer의 프로토콜이 Cancelable이네요..

Cancelable은 isDisposed 프로퍼티를 제공해서 구독 상태를 확인할 수 있게 해주는 프로토콜이고, 이 상위에 있는 프로토콜이 우리가 최종적으로 전달받는 Disposable입니다. 여기서 dispose 같은 메서드를 사용할 수 있죠.

자 이제 어느정도 정리가 되었습니다.

다시 코드로

Observable<String>.create({ (observer) -> Disposable in 
	observer.onNext("1") return Disposables.create() 
})

다시 이 create 로 새로운 옵저버블은 만드는 코드로 와볼게요. create를 하면서 전달한 클로저가 AnonymousObservable에 전달이 된다고 했죠? 그럼 AnonymousObservable의 인스턴스가 생성되고 그대로 반환됩니다.

let observable = Observable<String>.create({ (observer) -> Disposable in 
	observer.onNext("1") return Disposables.create() 
}) 

observable.subscribe( onNext: { data in print(data) })

그리고 이 AnonymousObservable을 subscribe하면, Producer에 있는 subscribe 메서드가 실행되죠.

앗 근데 subscribe의 파라미터 타입은 Observer인데, 우리는 이런거 준 적이 없는데요..?

ObservableType+Extension 파일을 열어보면 이런게 있거든요! 우리는 각 이벤트에 대한 핸들링 클로저만 전달해주어도 내부적으로 AnonymousObserver를 생성해서 반환합니다. 결국 우리는 Observer를 간접적으로 만들어주고 있었던거죠..!

그럼 run메서드가 실행되면 이 AnonymousObserver가 전달이 되겠죠?

이 부분입니다! cancel에는 아까 만들었던 disposer가 들어갑니다!

이제 AnonymousObserverSink 인스턴스가 만들어지고, 내부에 들어간 observer가 on이 불렸을 때 forwardOn이 불리면서 우리가 만든 클로저의 onNext를 실행해줄거예요.

subscription은 sink.run(self)로 만들어지는데요, 여기서 self는 우리가 정의한 클로저를 통해 만들어진 AnonymousObserver이고 self는 지금 구독하려는 Observable이에요. observable.subscribe로 불렀으니까요.

그럼 이 친구가 실행이 될 텐데요, 여기서 parent는 발행자가될 Observable이겠죠? 얘가 구현하고 있는 subscribeHandler, 즉 create안에서 만들었던 방출코드가 실행이 됩니다. 그리고 그 이벤트를 받는 주체가 AnonymousOberver가 됩니다.

간단하게 말하면 run이 호출되면 create에서 만들었던 이벤트들이 실행되면서 subscribe에서 만든 클로저들이 실행되는 것이죠.

의식의 흐름 정리 - 2

와 절대 이해하지 못할 줄 알았는데 조금은 이해가 되었습니다.. 정리해볼게요

1. create로 옵저버블을 만들면 create 메서드를 통해 생성된 AnonymousObservable이 반환된다.

Observable<String>.create({ (observer) -> Disposable in 
	observer.onNext("1") 
    return Disposables.create() 
})

2. 이 코드에서 create 안에 들어간 클로저가 AnonymousObservable의 subscribeHandler로 지정된다.

3. subscribe 메서드를 어떤 옵저버블에 대해 호출하면서 클로저로 onNext에 대한 클로저를 지정하면 ObservableType에서 이 클로저들을 포함하는 AnonymousObserver를 만들고 발행자의 상위 클래스인 Producer가 가진 run 메서드에 인자로 넘겨 호출한다.

observable.subscribe( onNext: { data in print(data) }) 

// onNext: { data in print(data) } --> AnonymousObserver로 생성 이름을 AO로 하자. 

// observable.subscribe(AO) 가 되고,
// Producer에서는 observable.run(AO, disposer)을 호출

 

4. Producer에서 호출된 run에서는 sink와 subscription을 만든다. sink는 이벤트에 대해 정의한 클로저를 이벤트 핸들러로 등록된 disposable을 반환한다. subsription은 AnonymousObservableSink 의 run 메서드를 호출한다.

5. run 메서드는 create에서 만들어서 등록된 subscribeHandler를 호출하고 그 인자로 subscribe 메서드에서 만든 AnonymousObserver을 전달한다.

Observable<String>.create({ (observer) -> Disposable in // 여기 observer가 subscribe에서 만들어진 AnonymousObserver 
	observer.onNext("1") 
    return Disposables.create() 
})

 

6. AnonymousObserver로 이벤트가 전달되고 내부의 on 메서드를 통해 이벤트가 처리된다.

마무리

괜히 시작했다..

쓰고나서 다시 읽어보는데 진짜 의식의 흐름대로 적어놨네요.. 나중에 시간을 내서 다시 정리해보겠습니다ㅠ 다른 자료를 안보고 RxSwift 깃허브를 계속 파면서 공부한거라 제 Swift 지식이 짧아 완전히 이해하지 못하고 적은 내용도 있을 수 있어요ㅠ

혹시 틀린 내용이 있다면 알려주시고 RxSwift의 내부 동작에 대해 잘 정리가 된 자료가 있다면 추천해주세요!

길고 혼란스러운 글 읽어주셔서 감사합니다~!