StringKeyPathObservingPublisher.swift

import Foundation

extension NSObject {
    /// A publisher for observing key-value changes when static swift KeyPaths are not available, such as when observing
    /// `UserDefaults`. With this publisher only a `String` based keypath is required.
    ///
    /// Since this does not use static typed keypaths, we can not ensure that the value received from KVO is the type we
    /// expect. So values that can not be converted to the expected value are ignored.
    public struct StringKeyPathObservingPublisher<Value>: Publisher {
        public typealias Output = Value
        public typealias Failure = Never
        
        private let observed: NSObject
        private let keyPath: String
        
        public init(object: NSObject, keyPath: String) {
            self.observed = object
            self.keyPath = keyPath
        }
        
        public func receive<S: Subscriber>(subscriber: S) where Self.Failure == S.Failure, Self.Output == S.Input {
            let subscription = Subscription(subscriber: subscriber, observed: observed, keyPath: keyPath)
            subscriber.receive(subscription: subscription)
        }
        
        private final class Subscription<S: Subscriber, Value>: NSObject, Combine.Subscription where S.Input == Value, S.Failure == Never {
            private var subscriber: S?
            private var demand: Subscribers.Demand = .none
            
            private let observed: NSObject
            private let keyPath: String
            private var hasObservation = false
            /// This lock is used to synchronize access to the mutable state within this class: `requested`, `subscriber` and `hasObservation`.
            private let lock = NSLock()
            /// This lock is used to ensure that the downstream publisher chain is not called simulatenously from multiple threads.
            /// We use a recursive lock beause the downstream publisher chain may synchronously initiate another KVO notification.
            private let downstreamLock = NSRecursiveLock()
            
            init(subscriber: S, observed: NSObject, keyPath: String) {
                self.subscriber = subscriber
                self.observed = observed
                self.keyPath = keyPath
                super.init()
            }
            
            func request(_ demand: Subscribers.Demand) {
                lock.lock()
                
                self.demand += demand
                if !hasObservation {
                    hasObservation = true
                    observed.addObserver(self, forKeyPath: keyPath, options: [.new], context: &kvoContext)
                }
                
                lock.unlock()
                
                guard let value = observed.value(forKeyPath: keyPath) as? Value else { return }
                send(value: value)
            }
            
            override func observeValue(forKeyPath keyPath: String?, of object: Any?, change: [NSKeyValueChangeKey: Any]?, context: UnsafeMutableRawPointer?) {
                guard context == &kvoContext, let newValue = change?[.newKey] as? Value else { return }
                send(value: newValue)
            }
        
            private func send(value: Value) {
                lock.lock()
                guard let subscriber = subscriber, demand > .none else {
                    lock.unlock()
                    return
                }
                // update the demand to reflect that a new value is about to be sent
                demand -= .max(1)
                lock.unlock()
                
                // use the `downstreamLock`, rather than `lock` to guard this critical section. That way lengthy execution on the
                // downstream publisher chain will not needlessly block the execution on the rest of this subscription. This
                // downstream publisher chain may also synchrnously trigger an additional KVO observation here. By using a
                // recursive lock for this critical section we can avoid deadlock in this situation.
                downstreamLock.lock()
                let newDemand = subscriber.receive(value)
                downstreamLock.unlock()
                guard newDemand != .none else { return }
                
                lock.lock()
                demand += newDemand
                lock.unlock()
            }
            
            func cancel() {
                lock.lock()
                defer { lock.unlock() }
                
                if hasObservation {
                    observed.removeObserver(self, forKeyPath: keyPath)
                    hasObservation = false
                }
                subscriber = nil
            }
            
            deinit {
                if hasObservation {
                    observed.removeObserver(self, forKeyPath: keyPath)
                    hasObservation = false
                }
            }
        }
    }
    
    /// Publish values when the value identified by a `String` based keypath changes. Only use this if static swift
    /// `KeyPath` is not available for this type and keypath.
    func publisher<T>(forKeyPath keyPath: String) -> StringKeyPathObservingPublisher<T> {
        StringKeyPathObservingPublisher<T>(object: self, keyPath: keyPath)
    }
}

/// Used as the context pointer for the KVO observation
private var kvoContext = 0