RxSwift 소스 분석 - 기본구조이해

RxSwift 소스 분석해보기

1. 기본구조이해
2. 기본구조이해 - 2
3. Scheduler에 대한 이해
4. Subject에 대한 이해

https://github.com/ReactiveX/RxSwift

기초 이해

• https://github.com/ReactiveX/RxSwift/blob/master/Documentation/GettingStarted.md#creating-your-own-observable-aka-observable-sequence

시작하기 위한 초간단 아이템 호출 구조

• Observable.create를 통해 Observable<E>를 생성한다. -> AnonymousObservable이 생성된다. ObservableType+Creation.swift

extension Observable {
  // element가 없는 AnonymousObservable을 생성한다.
  public static func create(_ subscribe: @escaping (AnyObserver<E>) -> Disposable) -> Observable<E> {
    return AnonymousObservable(subscribe)
  }
}

• AnonymousObservable에 subscribeHandler를 넘겨준다. -> subscribeHandler는 (AnyObserver<E> -> Disposable)이다.

class AnonymousObservable<Element> : Producer<Element> {
  let _subscribeHandler:  SubscribeHandler

  // 내부에 SubscribeHandler를 저장
  init(_ subscribeHandler: @escaping SubscribeHandler) {
    _subscribeHandler = subscribeHandler
  }
}

• 생성된 Observable에 subscribe를 하면 내부에서 Producer의 subscribe를 호출한다. -> AnonymousObservable은 Producer를 상속한다. ObservableType+Extensions.swift

extension ObservableType {
  public func subscribe(onNext: ((E) -> Void)? = nil, onError: ((Swift.Error) -> Void)? = nil, onCompleted: (() -> Void)? = nil, onDisposed: (() -> Void)? = nil) -> Disposable {
    ...
    // self.asObservable().subscribe(observer)를 호출한다.
    // observer는 사용자가 지정해준 onNext, onError등을 Event에 따라 실행해주는 AnonymousObserver이다.
  }
}

• Producer의 subscribe에서 AnonymousObservable의 run을 호출한다.

class Producer<Element> : Observable<Element> {
  override func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == Element {
    // scheduler에서 AnonymousObservable의 run을 실행한다.
  }
}

• AnonymousObservableSink의 run을 호출하여 subscribeHandler를 호출한다. subscribeHandler는 사용자가 넣어준 handler인데 조건에 따라 observer의 on을 호출한다.(AnyObserver를 통한 forwarding -> AnonymousObservableSink.on -> AnonymousObserver.on)

class AnonymousObservable<Element> : Producer<Element> {
  override func run<O : ObserverType>(_ observer: O, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where O.E == Element {
    // AnonymousObservableSink의 run을 호출한다.
  }
}

class AnonymousObservableSink<O: ObserverType> : Sink<O>, ObserverType {
  func on(_ event: Event<E>) {
    // AnonymousObserver를 통해 사용자가 넣어준 onNext, onError등의 함수를 호출한다.
  }

  func run(_ parent: Parent) -> Disposable {
    // observable을 생성할 때 넣어준 SubscribeHandler를 호출한다.
    // SubscribeHandler 안에서 observer.on을 호출하면 위의 on 함수가 호출된다.
  }
}

Subscription에 대한 취소 구조

• subscription을 호출하면 이에 대해 Disposable을 리턴한다. -> Disposable은 SinkDisposer이다.

class Producer<Element> : Observable<Element> {
  override func subscribe<O: ObserverType>(_ observer: O) -> Disposable where O.E == Element {
    let disposer = SinkDisposer()
    ...
    return disposer
  }
}

• AnonymousObservable의 run을 실행할 때 파라메터로 SinkDisposer를 넘겨주어서 AnonymousObservableSink도 SinkDisposer를 같이 바라보게 한다.
• SinkDisposer는 sink와 subscription을 가지고 있는다.
• • subscribe: _subscribeHandler의 리턴
• • sink: event를 실행하는 AnonymousObservableSink
• 사용자가 subscription을 cancel하고 싶으면 Disposable.dispose를 호출한다.
• • sink와 subscription 각각에 dispose를 호출한다.
• • sink에 dispose가 불리게 되면 observer의 on이 불리지 않게 된다. -> observer.on의 호출 전 dispose 여부 확인
• • subscription에 dispose가 불리게 되면 observable을 만들때의 subscription handler의 구현에 따라 관련 내용이 취소되도록 한다: ex) 네트워크 요청이 취소되도록 한다. 타이머가 중단되도록 한다.

Scheduler를 사용하는 경우의 구조

• observeOn을 사용해서 observer가 실행되는 scheduler를 지정할 수 있다.
• observeOn의 리턴은 원래의 observable과 scheduler를 가지고 있는 ObserveOnSerialDispatchQueue이다.
• ObserveOnSerialDispatchQueue는 sink로는 ObserveOnSerialDispatchQueueSink를, subscription으로는 원래의 source에 subscribe한것을 갖는다.
• 결국 사용자가 subscribe를 실행하면 ObserveOnSerialDispatchQueueSink가 불리게 되는데 여기서 scheduler를 통해 원래의 observer를 event를 통해 실행한다.

Simple UITextField bindings

• text property를 위한 reactive wrapper를 만든다.

public var text: ControlProperty<String?> {
  return UIControl.rx.value(...)
}

• UIControl.rx.value에서 Observable.create를 써서 observable을 만들고 UIBindingObserver의 bindingObserver를 사용해서 최종적으로 ControlProperty를 만든다.
• ControlProperty는 ObservableType이고 ObserverType이다.