WithPrevious.swift

//
//  WithPrevious.swift
//
//  Created by ha1f on 2018/06/25.
//  Copyright © 2018年 ha1f. All rights reserved.
//

import Foundation
import RxCocoa
import RxSwift

extension ObservableType {
    /// (今の値, 前の値)を流す
    func withPrevious() -> Observable<(Element, Element?)> {
        return WithPrevious(source: self.asObservable()).asObservable()
    }
}

private final class WithPreviousSink<Element, O: ObserverType>: ObserverType, Disposable where O.Element == (Element, Element?) {
    private let _observer: O
    private var _disposed: Bool

    private var _previousValue: Element?

    init(observer: O) {
        _observer = observer
        _disposed = false
    }

    func on(_ event: RxSwift.Event<Element>) {
        guard !_disposed else {
            return
        }

        switch event {
        case let .next(value):
            _observer.onNext((value, _previousValue))
            _previousValue = value
        case let .error(error):
            _observer.onError(error)
            dispose()
        case .completed:
            _observer.onCompleted()
            dispose()
        }
    }

    func dispose() {
        _disposed = true
    }
}

private final class WithPrevious<T>: ObservableType {
    typealias Element = (T, T?)

    private let _source: Observable<T>

    init(source: Observable<T>) {
        _source = source
    }

    func subscribe<O>(_ observer: O) -> Disposable where O: ObserverType, Element == O.Element {
        let sink = WithPreviousSink(observer: observer)
        let subscription = _source.subscribe(sink)
        return Disposables.create(subscription, sink)
    }
}

RxSwift 5.xならElement、それ以下ならEに変更する必要あり