CoreStore RxSwift List Observer

RxCoreStoreListObserver.swift

import Foundation
import RxSwift
import RxCocoa
import CoreStore


private class RxCoreStoreListObserver<B: FetchChainableBuilderType>: Disposable where B.ObjectType: NSManagedObject {
    
    private var monitor: ListMonitor<B.ObjectType>!
    
    private var retainSelf: Disposable?
    private var changeCallback: (([B.ObjectType]) -> Void)?

    fileprivate init(_ clauseChain: B,
                     onChange: (([B.ObjectType]) -> Void)? = nil) {
        changeCallback = onChange
        self.monitor = CoreStore.monitorList(clauseChain)
        monitor.addObserver(self)
        if let initialList = CoreStore.fetchAll(clauseChain) {
            changeCallback?(initialList)
        }
    }
    
    func dispose() {
        monitor.removeObserver(self)
        retainSelf = nil
    }
    
}


extension RxCoreStoreListObserver: ListObserver {
    
    typealias ListEntityType = B.ObjectType
    
    func listMonitorDidChange(_ monitor: ListMonitor<ListEntityType>) {
        changeCallback?(monitor.objectsInAllSections())
    }
    
    func listMonitorDidRefetch(_ monitor: ListMonitor<ListEntityType>) {
        changeCallback?(monitor.objectsInAllSections())
    }
    
}


extension CoreStore {

    static func observable<B: FetchChainableBuilderType>(_ clauseChain: B) -> Observable<[B.ObjectType]>
        where B.ObjectType: NSManagedObject
    {
        return Observable.create { (observer) -> Disposable in
            return RxCoreStoreListObserver<B>(clauseChain, onChange: { observer.onNext($0) })
        }.share(replay: 1, scope: .whileConnected)
    }

}