Cleaning up observables

Question

I am using rxSwift and I have a dictionary of observables which can be subscribed to.

var observables: [String: Observable<Bool>] = [:]

At some point I have to clean up some of those observables. I do it as follows

observables.removeValue(forKey: someKey)

Is it enough to clean up the observables with the above line? Is the observable also killed (or how do I "kill" it)? Someone might already be subscribed to the observable and then even after removing it from the dictionary it would still be alive and might fire, right? Or is the observable gone the moment I remove it because nobody holds a strong reference to it? What happens in the moment the observable is removed to potential subsribers?

I do not have access to the subscribers from the class where the dictionary with the observables is kept.

Answer 1

You can use takeUntil operator. It will send a completed signal to the observable, so the subscriber will release the retained resources.

For example, you can setup a PublishSubject where you send the observable identifier to complete that observable.

var observables: [String: Observable<Bool>] = [:]
let finishObservable = PublishSubject<String>()
func addObservable(observable: Observable<Bool>, withId identifier: String) -> Observable<Bool> {
    let condition = finishObservable.filter { $0 == identifier }
    let newObservable = observable.takeUntil(condition)
    observables[identifier] = newObservable
    return newObservable
}

This way, to clean an observable, you send the observable identifier and then you can remove the completed sequence from the dictionary.

func removeObservable(identifier: String) {
    // Complete the observable so it stops sending events and subscriber releases resources
    finishObservable.onNext(identifier)
    observables.removeValue(forKey: identifier)
}

---

If you're planning to share subscription between observers, you can also use a ConnectableObservable. I've used this kind of observables when subscribers come and go but you want to share the same subscription. It's usefull if the observable fetches network resources for example.

var disposables: [String: Disposable] = [:]
func addObservable(observable: Observable<Bool>, withId identifier: String) -> Observable<Bool> {
    let newObservable: ConnectableObservable = observable.replay(1)
    disposables[identifier] = newObservable.connect() // This call triggers the subscription, so you can call it later
    return newObservable
}

func removeObservable(identifier: String) {
    if let disposable = disposables.removeValue(forKey: identifier) {
        disposable.dispose()
    }
}