Refresh Observable in response to another

Question

I have an observable that emits a list of CNContacts, and I want to reload the list when there is a change to the Contacts database (.CNContactStoreDidChange).

So the observable should emit a value on subscription, and whenever the other observable (the notification) emits a value. That sounds like combining them with withLatestFrom, but it doesn't emit anything.

let myContactKeys = [
    CNContactIdentifierKey as CNKeyDescriptor,
    CNContactFormatter.descriptorForRequiredKeys(for: .fullName)
]

func fetchContacts(by identifiers: [String],
                 contactKeys: [CNKeyDescriptor]) -> Observable<Event<[CNContact]>> {

    return Observable<[String]>.just(identifiers)
        .withLatestFrom(NotificationCenter.default.rx.notification(Notification.Name.CNContactStoreDidChange)) { ids, _ in ids}
        .flatMap { ids in
            Observable<[CNContact]>.create { observer in
                let predicate = CNContact.predicateForContacts(withIdentifiers: ids)
                do {
                    let contacts = try CNContactStore().unifiedContacts(matching: predicate, keysToFetch: contactKeys)
                    observer.onNext(contacts)
                } catch {
                    observer.onError(error)
                }

                return Disposables.create()
            }
            .materialize()
        }
        .observeOn(MainScheduler.instance)
        .share(replay: 1)
        .debug()
}

fetchContacts(by: ["123"], contactKeys: myContactKeys)
    .subscribe(
        onNext: { contacts in
            contacts.forEach { print($0.fullName) }
        },
        onError: { error in
            print(error.localizedDescription)
        })
    .dispose(by: disposeBag)

Answer 1

The problem with your code is that you are starting with Observable<[String]>.just(identifiers) which will emit your identifiers and immediately complete. You don't want it to complete, you want it to continue to emit values whenever the notification comes in.

From your description, it sounds like you want something like the below. It emits whenever the notification fires, and starts with the contacts.

let myContactKeys = [
    CNContactIdentifierKey as CNKeyDescriptor,
    CNContactFormatter.descriptorForRequiredKeys(for: .fullName)
]

func fetchContacts(by identifiers: [String], contactKeys: [CNKeyDescriptor]) -> Observable<Event<[CNContact]>> {

    func update() throws -> [CNContact] {
        let predicate = CNContact.predicateForContacts(withIdentifiers: identifiers)
        return try CNContactStore().unifiedContacts(matching: predicate, keysToFetch: contactKeys)
    }
    return Observable.deferred {
        NotificationCenter.default.rx.notification(Notification.Name.CNContactStoreDidChange)
            .map { _ in }
            .map(update)
            .materialize()
        }
        .startWith({ () -> Event<[CNContact]> in
            do {
                return Event.next(try update())
            }
            catch {
                return Event.error(error)
            }
        }())
        .share(replay: 1)
        .debug()
}