[iOS - swift] Observable, Observer, Producer, Binder, ControlEvent 개념 (RxSwift, RxCocoa)

참고) RxSwift6 기준,

• RxSwift: Observable, Observer, Producer, Binder
• RxCocoa: ControlEvent

Observable, Observer, Producer 형태

• RxSwift 프레임워크 안에 존재

https://www.polidea.com/blog/8-Mistakes-to-Avoid-while-Using-RxSwiftPart-1/

Observable의 구조

• ObservableConvertibleType 프로토콜
  • asObservable() 메소드 구현을 강제화
  • Observable로 변환할 수 있는 타입이면, asObservable 메소드가 존재

public protocol ObservableConvertibleType {
    associatedtype Element
    
    func asObservable() -> Observable<Element>
}

• ObservableType
  • ObservableType은 subsribe메소드가 필수로 존재

public protocol ObservableType: ObservableConvertibleType {
    func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element
}

extension ObservableType {
    public func asObservable() -> Observable<Element> {
        Observable.create { o in self.subscribe(o) }
    }
}

• Observable 클래스
  
  • subscribe를 구현해서 사용해야하므로, rxAbstractMethod()에서 fatalError가 나도록 구현 (즉 Observable은 추상클래스)
  • asObservable()메소드는 자기 자신이 ObservableType이므로 self를 리턴

public class Observable<Element> : ObservableType {
    public func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
        rxAbstractMethod()
    }
    
    public func asObservable() -> Observable<Element> { self }
}

---

ObserverType 프로토콜

• on(_:)이라는 메소드 구현을 강제
• on(_:) 메소드는 Event가 일어났을 때 어떤 처리를 해줄 것인가를 정의 (next, error, complete 처리)

public protocol ObserverType {
    associatedtype Element
    
    func on(_ event: Event<Element>)
}

extension ObserverType {
    public func onNext(_ element: Element) {
        self.on(.next(element))
    }
    
    public func onCompleted() {
        self.on(.completed)
    }
    
    public func onError(_ error: Swift.Error) {
        self.on(.error(error))
    }
}

-> on(_:) 메소드 예시, first()연산자 구현

on(_:)메소드 안에서 이벤트 처리

private final class FirstSink<Element, Observer: ObserverType> : Sink<Observer>, ObserverType where Observer.Element == Element? {
    typealias Parent = First<Element>

    func on(_ event: Event<Element>) {
        switch event {
        case .next(let value):
            self.forwardOn(.next(value))
            self.forwardOn(.completed)
            self.dispose()
        case .error(let error):
            self.forwardOn(.error(error))
            self.dispose()
        case .completed:
            self.forwardOn(.next(nil))
            self.forwardOn(.completed)
            self.dispose()
        }
    }
}

final class First<Element>: Producer<Element?> {
    private let _source: Observable<Element>

    init(source: Observable<Element>) {
        self._source = source
    }

    override func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element? {
        let sink = FirstSink(observer: observer, cancel: cancel)
        let subscription = self._source.subscribe(sink)
        return (sink: sink, subscription: subscription)
    }
}

Producer 개념

• Observabe 클래스를 서브클래싱한 추상클래스
• run(_:cancel:) 메소드
  • fatalError가 나므로 이를 서브클래싱하는 클래스들이 반드시 구현해야하는 메소드
  • sink라는 데이터 목적지의 인스턴스와 구독되는 인스턴스를 반환하여 값이 실행되게끔 하는 메소드
  • 위 Firrst클래스와 같이, Producer를 준수하여 run메소드를 구현

class Producer<Element> : Observable<Element> {
    override init() {
        super.init()
    }

    override func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
        if !CurrentThreadScheduler.isScheduleRequired {
            // The returned disposable needs to release all references once it was disposed.
            let disposer = SinkDisposer()
            let sinkAndSubscription = self.run(observer, cancel: disposer)
            disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

            return disposer
        }
        else {
            return CurrentThreadScheduler.instance.schedule(()) { _ in
                let disposer = SinkDisposer()
                let sinkAndSubscription = self.run(observer, cancel: disposer)
                disposer.setSinkAndSubscription(sink: sinkAndSubscription.sink, subscription: sinkAndSubscription.subscription)

                return disposer
            }
        }
    }

    func run<Observer: ObserverType>(_ observer: Observer, cancel: Cancelable) -> (sink: Disposable, subscription: Disposable) where Observer.Element == Element {
        rxAbstractMethod()
    }
}

Binder 개념

• Binder는 extension으로 rx 네임 스페이스로 프로퍼티에 접근하기 위해 확장할때 사용되는 RxSwift에서 정의한 타입

class MyView: UIView {
  var myProperty: String = ""
}

extension Reactive where Base: MyView {
  var myProeprty: Binder<String> { // <- Binder 사용
     Binder(base) { base, newProperty in
         base.myProperty = newProperty
     }
  }
}

• Binder 구조체 핵심은 init(_ target:, binding:) 부분
  • target 인자에는 rx 네임스페이스로 접근하려는 타입을 명시
  • binding 인자에는 클로저를 주입
  • : 인자로 target 타입의 인스턴스와, 전달받는 값 value를 받아서 next이벤트가 발생한 경우 해당 클로저를 실행하는 로직

// Binder.swift

public struct Binder<Value>: ObserverType {
    public typealias Element = Value
    
    private let binding: (Event<Value>) -> Void

    /// Initializes `Binder`
    ///
    /// - parameter target: Target object.
    /// - parameter scheduler: Scheduler used to bind the events.
    /// - parameter binding: Binding logic.
    public init<Target: AnyObject>(_ target: Target, scheduler: ImmediateSchedulerType = MainScheduler(), binding: @escaping (Target, Value) -> Void) {
        weak var weakTarget = target

        self.binding = { event in
            switch event {
            case .next(let element):
                _ = scheduler.schedule(element) { element in
                    if let target = weakTarget {
                        binding(target, element)
                    }
                    return Disposables.create()
                }
            case .error(let error):
                rxFatalErrorInDebug("Binding error: \(error)")
            case .completed:
                break
            }
        }
    }

    /// Binds next element to owner view as described in `binding`.
    public func on(_ event: Event<Value>) {
        self.binding(event)
    }

    /// Erases type of observer.
    ///
    /// - returns: type erased observer.
    public func asObserver() -> AnyObserver<Value> {
        AnyObserver(eventHandler: self.on)
    }
}

---

ControlEvent 구조

• RxCocoa에서 정의
• ControlEventType은 ObservableType을 채택한 형태

/// A protocol that extends `ControlEvent`.
public protocol ControlEventType : ObservableType {

    /// - returns: `ControlEvent` interface
    func asControlEvent() -> ControlEvent<Element>
}

• ControlEvent는 ControlEventType을 채택한 형태
  • UI 관련 events 처리 시 사용되는 ObservableType의 한 종류 (ex - touchUpInside 이벤트) 
  • events라는 Observable 프로퍼티를 가지고 있으며, init에서 events는 MainScheduler에서 구독하고 있는 형태
  • touchUpInside 이벤트는 init구문에서 MainScheduler로 구독중이고, subscribe하는 대상에게 event를 그대로 전달

public struct ControlEvent<PropertyType> : ControlEventType {
    public typealias Element = PropertyType

    let events: Observable<PropertyType>

    public init<Ev: ObservableType>(events: Ev) where Ev.Element == Element {
        self.events = events.subscribe(on: ConcurrentMainScheduler.instance)
    }

    public func subscribe<Observer: ObserverType>(_ observer: Observer) -> Disposable where Observer.Element == Element {
        self.events.subscribe(observer)
    }

    public func asObservable() -> Observable<Element> {
        self.events
    }

    public func asControlEvent() -> ControlEvent<Element> {
        self
    }
}

-> 이름이 ControlEvent인 이유?

- MainScheduler에서 구독되고 있으며 애초에 UI 이벤트에 대한 것을 겨냥하고 정의

- UI이벤트는 Target - Action으로 UIControl에서 정의된 이벤트들을 뜻하므로, Control을 비롯하여 ControlEvent로 명명한 것으로 추측

 

cf) 더 깊은 이해를 위한 아래 포스팅 글 참고
- Reactive Extension 사용 방법, RxSwift6에서 @dynamicMemberLookup을 이용한 Binder 합성