仿写ReSwift接口

RxDemo-01.swift

print("Start")

enum Event<Element>{
    case next(Element)
    case error(Error)
    case complete
}

protocol ObserverType {
    associatedtype Element

    //
    func on(event: Event<Element>)
}

class Observer<Element>: ObserverType{
    private let _handle: (Event<Element>) -> Void

    init(handle: @escaping (Event<Element>) -> Void) {
        _handle = handle
    }

    func on(event: Event<Element>) {
        _handle(event)
    }

}

protocol ObservableType {
    associatedtype Element

    func subscrible<O: ObserverType>(observer: O) where O.Element == Element
}

class Observable<Element>: ObservableType {

    private let _eventGenerator: (Observer<Element>) -> Void

    init(eventGenerator: @escaping (Observer<Element>)->Void) {
        _eventGenerator = eventGenerator
    }

    func subscrible<O: ObserverType>(observer: O) where O.Element == Element {
        // 生成事件
        _eventGenerator(observer as! Observer<Element>)
    }

}

// Test
let myobservable = Observable<Int> { (observer) in
    print("send 0")
    observer.on(event: .next((0)))
    print("send 1")
    observer.on(event: .next((1)))
    print("send 2")
    observer.on(event: .next((2)))
    print("send 3")
    observer.on(event: .next((3)))
    print("send completed")
    observer.on(event: .complete)
}

let myobserver = Observer<Int> { (event) in
    switch  event{
    case .next(let value):
        print("receive \(value)")
    case .error(let error):
        print("receive \(error)")
    case .complete:
        print("receive compeleted")
    }
}

myobservable.subscrible(observer: myobserver)



print("End")

RxDemo-02.swift

import Dispatch

print("Start")

enum Event<Element>{
    case next(Element)
    case error(Error)
    case complete
}

protocol ObserverType {
    associatedtype Element

    //
    func on(event: Event<Element>)
}

class Observer<Element>: ObserverType{
    private let _handle: (Event<Element>) -> Void

    init(handle: @escaping (Event<Element>) -> Void) {
        _handle = handle
    }

    func on(event: Event<Element>) {
        _handle(event)
    }

}

protocol ObservableType {
    associatedtype Element

    func subscrible<O: ObserverType>(observer: O) -> Disposable where O.Element == Element
}

class Observable<Element>: ObservableType {

    private let _eventGenerator: (Observer<Element>) -> Disposable


    init(eventGenerator: @escaping (Observer<Element>)-> Disposable) {
        _eventGenerator = eventGenerator
    }

    func subscrible<O: ObserverType>(observer: O) -> Disposable where O.Element == Element {
        let composite = CompositeDisposable()
        // 生成事件
        let disposable = _eventGenerator(Observer { (event) in
            guard !composite.isDisposed else {
                return
            }
            // 事件传递给原始的Observer
            observer.on(event: event)
            // 通过 composite 管理 error、completed时，取消订阅
            switch event {
            case .error(_), .complete:
                composite.dispose()
            default:
                break
            }
        })
        composite.add(disposable: disposable)
        return composite
    }
}

protocol Disposable {
    // 取消订阅
    func dispose()
}

class AnonymousDisposable: Disposable {
    private let _disposeHandler: ()->Void

    init(_ disposeHandle: @escaping ()->Void) {
        _disposeHandler = disposeHandle
    }

    func dispose() {
        _disposeHandler()
    }
}

class CompositeDisposable: Disposable {
    // 用于管理一组 Disposable
    private(set) var isDisposed: Bool = false
    private var disposables: [Disposable] = []

    func add(disposable: Disposable) {
        if isDisposed {
            disposable.dispose()
            return
        }
        disposables.append(disposable)
    }

    func dispose() {
        guard !isDisposed else {
            return
        }
        // 销毁所有的Disposable
        disposables.forEach { (disposable) in
            disposable.dispose()
        }
        isDisposed = true
    }
}

// Test
let myobservable = Observable<Int> { (observer) in
    print("send 0")
    observer.on(event: .next((0)))
    print("send 1")
    observer.on(event: .next((1)))
    print("send 2")
    observer.on(event: .next((2)))
    print("send 3")
    observer.on(event: .next((3)))
    DispatchQueue.main.asyncAfter(deadline: .now()+3) {
        print("send completed")
        observer.on(event: .complete)
    }
    return AnonymousDisposable{
        print("disposed")
    }

}

let myobserver = Observer<Int> { (event) in
    switch  event{
    case .next(let value):
        print("receive \(value)")
    case .error(let error):
        print("receive \(error)")
    case .complete:
        print("receive compeleted")
    }
}


let disposable = myobservable.subscrible(observer: myobserver)

DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
    disposable.dispose()
}

print("End")

RxDemo-03.swift

import Dispatch

print("Start")

enum Event<Element>{
    case next(Element)
    case error(Error)
    case complete
}

protocol ObserverType {
    associatedtype Element

    //
    func on(event: Event<Element>)
}

class Observer<Element>: ObserverType{
    private let _handle: (Event<Element>) -> Void

    init(handle: @escaping (Event<Element>) -> Void) {
        _handle = handle
    }

    func on(event: Event<Element>) {
        _handle(event)
    }

}

protocol ObservableType {
    associatedtype Element

    func subscrible<O: ObserverType>(observer: O) -> Disposable where O.Element == Element
}

class Observable<Element>: ObservableType {

    private let _eventGenerator: (Observer<Element>) -> Disposable


    init(eventGenerator: @escaping (Observer<Element>)-> Disposable) {
        _eventGenerator = eventGenerator
    }

    func subscrible<O: ObserverType>(observer: O) -> Disposable where O.Element == Element {
        let sink = Sink(forward: observer, eventGenerator: _eventGenerator)
        sink.run()
        return sink
    }
}

protocol Disposable {
    // 取消订阅
    func dispose()
}

class AnonymousDisposable: Disposable {
    private let _disposeHandler: ()->Void

    init(_ disposeHandle: @escaping ()->Void) {
        _disposeHandler = disposeHandle
    }

    func dispose() {
        _disposeHandler()
    }
}

class CompositeDisposable: Disposable {
    // 用于管理一组 Disposable
    private(set) var isDisposed: Bool = false
    private var disposables: [Disposable] = []

    func add(disposable: Disposable) {
        if isDisposed {
            disposable.dispose()
            return
        }
        disposables.append(disposable)
    }

    func dispose() {
        guard !isDisposed else {
            return
        }
        // 销毁所有的Disposable
        disposables.forEach { (disposable) in
            disposable.dispose()
        }
        isDisposed = true
    }
}

class Sink<O: ObserverType>: Disposable {
    private var _disposed: Bool = false
    private let _forward: O
    private let _eventGenerator: (Observer<O.Element>) -> Disposable
    private let _composite = CompositeDisposable()

    init(forward: O, eventGenerator: @escaping (Observer<O.Element>) -> Disposable) {
        _forward = forward
        _eventGenerator = eventGenerator
    }

    func run(){
        let observer = Observer<O.Element>(handle: forward)  // 原observer转成中间observer
        _composite.add(disposable: _eventGenerator(observer))
    }

    private func forward(event: Event<O.Element>){
        guard !_disposed else { return }
        _forward.on(event: event)
        switch event {
        case .complete, .error(_):
            dispose()
        default:
            break
        }
    }

    func dispose() {
        _disposed = true;
        _composite.dispose()
    }
}

// Test
let myobservable = Observable<Int> { (observer) in
    print("send 0")
    observer.on(event: .next((0)))
    print("send 1")
    observer.on(event: .next((1)))
    print("send 2")
    observer.on(event: .next((2)))
    print("send 3")
    observer.on(event: .next((3)))
    DispatchQueue.main.asyncAfter(deadline: .now()+3) {
        print("send completed")
        observer.on(event: .complete)
    }
    return AnonymousDisposable{
        print("disposed")
    }

}

let myobserver = Observer<Int> { (event) in
    switch  event{
    case .next(let value):
        print("receive \(value)")
    case .error(let error):
        print("receive \(error)")
    case .complete:
        print("receive compeleted")
    }
}


let disposable = myobservable.subscrible(observer: myobserver)

DispatchQueue.main.asyncAfter(deadline: .now() + 2) {
    disposable.dispose()
}

print("End")

RxDemo-04.swift

extension ObservableType {
    func map<Result>(_ transform: @escaping (Element) throws -> Result) -> Observable<Result> {
        return Observable<Result>{ (observer) in  // observer 为原始 observer
            return self.subscrible(observer: Observer{ (event) in
                switch event {
                case .next(let value):
                    do{
                        try observer.on(event: .next(transform(value)))
                    }
                    catch{
                        observer.on(event: .error(error))
                    }
                case .error(let error):
                    observer.on(event: .error(error))
                case .complete:
                    observer.on(event: .complete)
                default:
                    break
                }
            })
        }
    }
}

let disposable = myobservable.map({$0 * 2}).subscrible(observer: myobserver)

from http://chuquan.me/2020/08/24/rxswift-core-implement/