Rx Sample

RxSample.swift

//
//  RxSample.swift
//  Rx Sample
//
//  Created by Svetoslav on 2018/12/5.
//  Copyright © 2018. All rights reserved.
//
import UIKit

import Model
import RxSwift
import RxCocoa

class NormalSummary: NSObject {
    let disposeBag = DisposeBag()
    let person: Person = Person()
    
    let subject = BehaviorRelay<String>(value: "")
    
    //MARK: 组合Map FlatMap FlatMapLatest
    func kvo_rx() {
        
        let observer = AnyObserver<Int> { (event) in
            print(event)
            //            print(event.element ?? "aaaaa")
        }
        
        observer.onNext(2)
        observer.onNext(3)
        
        Observable<Int>.just(1).subscribe(observer).disposed(by: disposeBag)
        
    }
}

//MARK: Observable变体 Single Completable driver controlEvent
extension NormalSummary {
    
    func rx_observable() {
        let observable = Observable<Int>.create { (observer) -> Disposable in
            print("/.......")
            observer.onNext(1)
            return Disposables.create {
                print("dispose........")
            }
        }.share(replay: 1)

        observable.subscribe(onNext: {
            print("1111111")
            print($0)
        }).disposed(by: self.disposeBag)
        
        observable.subscribe(onNext: {
            print("2222222")
            print($0)
        }).disposed(by: self.disposeBag)
    }
    
    func rx_single() {
        
        let single = Single<Int>.create { single in
            print("........发送信号")
            single(.success(1))
            
            return Disposables.create()
        }
        
        single.subscribe { (event) in
            switch event {
            case .success(let element):
                print(element)
                break
            case .error(_):
                break
            }
            }.disposed(by: disposeBag)
        
        single.subscribe { (event) in
            switch event {
            case .success(let element):
                print(element)
                break
            case .error(_):
                break
            }
            }.disposed(by: disposeBag)
    }
    
    /// completed error
    func rx_completable() {
        let completable = Completable.create { (completable) -> Disposable in
            completable(.completed)
            return Disposables.create()
        }
        
        completable.subscribe(onCompleted: {
            print("completed")
        }) { (error) in
            print(error)
            }.disposed(by: disposeBag)
    }
    
    func rx_maybe() {
        let success = true
        let maybe = Maybe<Int>.create { (maybe) -> Disposable in
            
            if success {
                maybe(.success(1))
            } else {
                //                maybe(.error(nil))
                maybe(.completed)
            }
            
            return Disposables.create()
        }
        
        maybe.subscribe { (maybe) in
            switch maybe {
            case .success(let element):
                print(element)
                break
                
            case .error(let error):
                print(error)
                break
                
            case .completed:
                break
            }
            }.disposed(by: disposeBag)
    }
    
    func rx_driver() {
        Observable<Int>.just(1).asDriver(onErrorJustReturn: -1).drive(onNext: { (element) in
            print(element)
        }).disposed(by: self.disposeBag)
    }
}

//MARK: AnyObserver Binder
extension NormalSummary {
    func rx_anyObserver() {
        let observer = AnyObserver<Int>.init { (event) in
            switch event {
            case .next(let element):
                print(element)
                break
            case .error(_):
                break
                
            case .completed:
                break
            }
        }
        
        Single<Int>.create { (single) -> Disposable in
            single(.success(1))
            return Disposables.create()
            }.asObservable().subscribe(observer).disposed(by: self.disposeBag)
    }
    
    func rx_Binder() {
        let observer: Binder<Int> = Binder<Int>.init(person) { (person, value) in
            person.age = "\(value)"
        }
        
        Observable.of(1, 2, 3).asDriver(onErrorJustReturn: -1).drive(observer).disposed(by: self.disposeBag)
    }
}

//MARK: Subject AsyncSubject,PublishSubject,ReplaySubject,BehaviorSubject,Variable,ControlProperty
extension NormalSummary {
    func rx_AsyncSubject() {
        let asyncSubject = AsyncSubject<Int>()
        asyncSubject.subscribe(onNext: { (element) in
            print("thread:\(Thread.current) element:\(element)")
        }).disposed(by: self.disposeBag)
        
        asyncSubject.onNext(1)
        asyncSubject.onNext(2)
        asyncSubject.onCompleted()
        
    }
    
    func rx_publishSubject() {
        let publishSubject = PublishSubject<Int>()
        publishSubject.onNext(1)
        
        publishSubject.subscribe(onNext: { (element) in
            print(element)
        }).disposed(by: self.disposeBag)
        
        publishSubject.onNext(2)
        publishSubject.onNext(3)
    }
    
    func rx_replaySubject() {
        let replaySubject = ReplaySubject<Int>.create(bufferSize: 2)
        
        replaySubject.onNext(1)
        replaySubject.onNext(2)
        replaySubject.onNext(3)
        replaySubject.onNext(4)
        
        replaySubject.subscribe(onNext: { (element) in
            print(element)
        }).disposed(by: self.disposeBag)
        
        replaySubject.onNext(5)
        replaySubject.onNext(6)
    }
    
    func rx_behaviorSubject() {
        let behaviorSubject = BehaviorSubject<Int>.init(value: 0)
        
        behaviorSubject.onNext(1)
        behaviorSubject.onNext(2)
        
        behaviorSubject.subscribe(onNext: { (element) in
            print(element)
        }).disposed(by: self.disposeBag)
        
        behaviorSubject.onNext(3)
    }
    
    func rx_variable() {
        let variable = Variable<Person>(person)
        
        variable.asObservable().subscribe(onNext: { (person) in
            print(person.name ?? "无名氏")
        }).disposed(by: self.disposeBag)
        
        variable.value = Person(id: "1", name: "jack", age: "10", phone: "18516518602")
    }
    
    func rx_behaviorReply() {
        let replay: BehaviorRelay<Int> = BehaviorRelay<Int>(value: 0)
        replay.skip(1).subscribe(onNext: {
            print($0.isRxSubjectDefaultSingle, $0)
        }).disposed(by: self.disposeBag)
        
        replay.accept(2)
        replay.accept(3)
        replay.accept(4)
        replay.accept(5)
    }
    
    func rx_controlProperty() {
        let observable = Observable.just(1)
        observable.subscribe(onNext: { (el) in
            print("observable:\(el)")
        }).disposed(by: self.disposeBag)
        
        let observer   = Binder<Int>.init(person) { (person, value) in
            print("observer:\(value)")
        }
        
        let controlProperty = ControlProperty<Int>.init(values: observable, valueSink: observer)
        controlProperty.subscribe(onNext: { (el) in
            print("-----control property:\(el)")
        }).disposed(by: self.disposeBag)
        
        //        controlProperty.asDriver().drive(observer).disposed(by: self.disposeBag)
        observer.onNext(2)
        //        controlProperty.onNext(2)
    }
}

extension Person {
    
}

//MARK: operator 错误处理Error Handling
extension NormalSummary {
    
    private enum MyError: Error {
        case code(Int)
    }
    
    func rx_errorHandle() {
        Observable<Int>.create { (observer) -> Disposable in
            
            observer.onNext(1)
            observer.onNext(2)
            observer.onNext(3)
            
            observer.onError(MyError.code(2000))
            
            observer.onNext(4)
            observer.onNext(5)
            
            observer.onCompleted()
            
            return Disposables.create()
            }.retry(2).subscribe(onNext: { (element) in
                print(element)
            }).disposed(by: self.disposeBag)
        
        let observable = Observable<String>.create { (observer) -> Disposable in
            return Disposables.create()
        }
        
        observable.retryWhen { (rxError: Observable<Error>) -> Observable<String> in
            return Observable<String>.just("abc")
            //            return Observable<String>.timer(2, scheduler: MainScheduler.instance)
            }.subscribe(onNext: { print($0) }).disposed(by: self.disposeBag)
    }
}

private var kIsRxSingleDetaulKey = "kIsRxSingleDetaulKey"
extension Int {
    var isRxSubjectDefaultSingle: Bool {
        get {
            if let value = objc_getAssociatedObject(self, &kIsRxSingleDetaulKey) as? Bool {
                return value
            }
            
            return true
        }
        
        set {
            objc_setAssociatedObject(self, &kIsRxSingleDetaulKey, newValue, .OBJC_ASSOCIATION_ASSIGN)
        }
    }
}