Make CBLReplication to return Observable<Double> to monitor the progress

CBLReplication+Rx.swift

func sync(remote syncURL: String, for store: Store) -> Observable<Double> {
    // Make sure url is set
    guard !syncURL.isEmpty, let url = URL(string: syncURL) else {
        return Observable.error(RemoteSyncError.badConfiguration)
    }
    // Make sure the store contains a session
    guard let session = store.syncSession else {
        return Observable.error(RemoteSyncError.missingSyncSession)
    }
    // Make sure store is good
    guard session.isValid else {
        return Observable.error(RemoteSyncError.invalidSyncSession(message: "Invalid or expired sync session"))
    }
    return Observable<Double>.create { observer -> Disposable in
        // Create the pull replicaton
        let pull = self.database.createPullReplication(url)
        pull.setCookieNamed(session.cookieName, withValue: session.sessionID, path: "/", expirationDate: session.expires, secure: false)
        pull.continuous = false
        pull.channels = [store.id, "etc_\(store.id)"]
        // Configure the push
        let push = self.database.createPushReplication(url)
        push.setCookieNamed(session.cookieName, withValue: session.sessionID, path: "/", expirationDate: session.expires, secure: false)
        push.continuous = false
        push.filter = "remotePushFilter"
        push.filterParams = ["storeid": store.id]
        var disposableBag: DisposeBag? = DisposeBag()
        Observable.merge(
            NotificationCenter.default.rx.notification(.cblReplicationChange, object: pull).skip(1),
            NotificationCenter.default.rx.notification(.cblReplicationChange, object: push).skip(1))
            .subscribe(onNext: { _ in
                v("[REMOTESYNC] Pull Status \(pull.status.rawValue) /  Push Status \(pull.status.rawValue)")
                // HACK: nowhere on the Interner tell the reason why the 2 replications always start with offline statuses
                if pull.status == .offline && push.status == .offline {
                    observer.onNext(0)
                    return
                }
                // Still active, calculate and inform progress
                if pull.status == .active || push.status == .active {
                    let total = pull.changesCount + push.changesCount
                    let completed = pull.completedChangesCount + push.completedChangesCount
                    if total > 0 {
                        let progress = Double(completed) / Double(total)
                        v("[REMOTESYNC] Progress changed \(progress)")
                        observer.onNext(progress)
                    } else {
                        observer.onNext(0)
                    }
                } else { // All inactive, means done
                    i("[REMOTESYNC] Completed")
                    // Notify listener
                    observer.onCompleted()
                    disposableBag = nil
                }
            }).disposed(by: disposableBag!)

        // Start the replications
        pull.start()
        push.start()
        i("[REMOTESYNC] Started for Store '\(store.storeName)/\(store.id)'")

        // When disposed, it must stop the replications
        return Disposables.create {
            pull.stop()
            push.stop()
            disposableBag = nil
        }
    }
}