CoreStore RxSwift Property Observer

RxCoreStorePropertyObserver.swift

import Foundation
import RxSwift
import RxCocoa
import CoreStore


private class RxCoreStorePropertyObserver<ObjectEntity: NSManagedObject, Property>: Disposable {
    
    private var monitor: ObjectMonitor<ObjectEntity>!
    private var propertyKeyPath: KeyPath<ObjectEntity, Property>!
    
    private var retainSelf: Disposable?
    private var changeCallback: ((Property) -> Void)?
    private var deleteCallback: (() -> Void)?

    fileprivate init(object: ObjectEntity,
                     keyPath propertyKeyPath: KeyPath<ObjectEntity, Property>,
                     onChange: ((Property) -> Void)? = nil,
                     onDelete: (() -> Void)? = nil) {
        self.propertyKeyPath = propertyKeyPath
        changeCallback = onChange
        deleteCallback = onDelete
        onChange?(object[keyPath: propertyKeyPath])
        monitor = CoreStore.monitorObject(object)
        monitor.addObserver(self)
    }
    
    func dispose() {
        monitor.removeObserver(self)
        retainSelf = nil
    }
    
}


extension RxCoreStorePropertyObserver: ObjectObserver {
    
    typealias ObjectEntityType = ObjectEntity
    
    func objectMonitor(_ monitor: ObjectMonitor<ObjectEntityType>,
                       didUpdateObject object: ObjectEntityType,
                       changedPersistentKeys: Set<KeyPathString>) {
        if changedPersistentKeys.contains(NSExpression(forKeyPath: propertyKeyPath).keyPath) {
            changeCallback?(object[keyPath: propertyKeyPath])
        }
    }
    
    func objectMonitor(_ monitor: ObjectMonitor<ObjectEntityType>,
                       didDeleteObject object: ObjectEntityType) {
        deleteCallback?()
    }

}


extension CoreStore {

    static func observable<ObjectEntity: NSManagedObject, Property>
        (for object: ObjectEntity, keyPath propertyKeyPath: KeyPath<ObjectEntity, Property>) -> Observable<Property>
    {
        return Observable.create { (observer) -> Disposable in
            return RxCoreStorePropertyObserver<ObjectEntity, Property>(object: object,
                                                  keyPath: propertyKeyPath,
                                                  onChange: { observer.onNext($0) },
                                                  onDelete: { observer.onCompleted() })
        }.share(replay: 1, scope: .whileConnected)
    }

}