RxSwift changes trigger "Warning: recursive call or sync error!"

Question

RxSwift changes trigger "Warning: recursive call or sync error!"

I inherited some Swift 3 code that uses RxSwift to manage storage. Main class layout:

class UserActivityStore {
    fileprivate lazy var internalDataCache: Variable<Set<NSUserActivity>?> = Variable(nil)

    func addAction(_ actionToAdd: NSUserActivity) {
        var content = internalDataCache.value ?? Set<NSUserActivity>()
        content.insert(actionToAdd)
        internalDataCache.value = content
    }

    func resolveAction(_ action: NSUserActivity) {
        var content = internalDataCache.value
        _ = content?.remove(action)
        internalDataCache.value = content
    }

    func expectActions(_ filter: @escaping ((NSUserActivity) -> Bool)) -> Observable<NSUserActivity> {
        let observable = Observable<NSUserActivity>.create { (observer) -> Disposable in
            return self.internalDataCache.asObservable().subscribeNext { (newActions) in
                guard let newActions = newActions else { return }
                for eachAction in newActions where filter(eachAction) {
                    observer.onNext(eachAction)
                    self.resolveAction(eachAction)
                }
            }
        }
        return observable
    }
}

When an action is added to this, it correctly adds the item to the set. However, observer (c expectActions

) catches this change and solves it. Since this is all in one thread, the error "Warning: recursive call or sync error!" starts up.

I think this is a perfectly legitimate error and that RxSwift is probably correct in handling it. However, I can't help but think that this is a bad model. The code essentially processes a queue of objects NSUserActivity

.

Is this really a RxSwift modeling / abuse bug, or my limited understanding of RxSwift a misunderstanding? As a hacky solution, I tried to replace the function resolveAction

with one line internalDataCache.value?.remove(action)

, but that still throws an observable and hence an error.