MutableVariable.swift

// swift 4.2

public class MutableVariable<A>: Variable<A> {
    public typealias Transform = (inout A) -> Void
    
    internal typealias Mutate = (@escaping Transform) -> Void
    
    override public class func threadSafe(_ initial: A) -> MutableVariable<A> {
        return MutableVariable(initial, queue: DispatchQueue(label: "com.shawnthroop.sting.\(self)", attributes: .concurrent))
    }
    
    private let mutate: Mutate
    
    public convenience init(_ initial: A, queue: DispatchQueue? = nil) {
        var observers = Observers()
        var value: A = initial {
            didSet {
                observers.notifyObserversOfChange(from: oldValue, to: value)
            }
        }
        
        self.init(queue: queue, get: { value }, mutate: { $0(&value) }) { observer in
            let id = observers.nextIdentifier()
            observers[id] = observer
            
            return Disposable {
                observers[id] = nil
            }
        }
    }
    
    private convenience init(queue: DispatchQueue?, get: @escaping Get, mutate: @escaping Mutate, append: @escaping AppendObserver) {
        if let q = queue {
            self.init(queue: q, get: get, mutate: mutate, append: append)
        } else {
            self.init(get: get, mutate: mutate, append: append)
        }
    }
    
    internal init(queue: DispatchQueue, get: @escaping Get, mutate: @escaping Mutate, append: @escaping AppendObserver) {
        self.mutate = { transform in queue.async(flags: .barrier) { mutate(transform) }}
        super.init(queue: queue, get: get, append: append)
    }
    
    internal init(get: @escaping Get, mutate: @escaping Mutate, append: @escaping AppendObserver) {
        self.mutate = mutate
        super.init(get: get, append: append)
    }
    
    
    public func mutate(by transform: @escaping Transform) {
        mutate({ transform(&$0) })
    }
    
    
    /// Subscripting
    
    public subscript<B>(keyPath: WritableKeyPath<A,B>) -> MutableVariable<B> {
        let g: MutableVariable<B>.Get = { self.value[keyPath: keyPath] }
        let m: MutableVariable<B>.Mutate = { transform in
            self.mutate(by: { transform(&$0[keyPath: keyPath]) })
        }
        
        return MutableVariable<B>(get: g, mutate: m) { observer in
            self.addObserver { new, old in
                observer(new[keyPath: keyPath], old[keyPath: keyPath])
            }
        }
    }
}

Variable.swift

// swift 4.2

/// A reference type that allows observing value types.
///
/// - Note:
///     Orignal concept from [Swift Talk - S01E63: Mutable Shared Structs](https://talk.objc.io/episodes/S01E63-mutable-shared-structs-part-2)

public class Variable<A> {
    public typealias Observer = (_ new: A, _ old: A) -> Void
    
    internal typealias Get = () -> A
    internal typealias AppendObserver = (@escaping Observer) -> Disposable
    
    internal class Observers {
        typealias Identifier = Int
        
        private var observers = [Identifier: Observer]()
        private var identifiers = (0...).makeIterator()

        func nextIdentifier() -> Identifier {
            return identifiers.next()!
        }
        
        subscript(identifier: Identifier) -> Observer? {
            get {
                return observers[identifier]
            }
            set {
                observers[identifier] = newValue
            }
        }
        
        func notifyObserversOfChange(from old: A, to new: A) {
            for observer in observers.values {
                observer(new, old)
            }
        }
    }
    
    public class func threadSafe(_ initial: A) -> Variable<A> {
        return Variable(initial, queue: DispatchQueue(label: "com.throop.sting.\(self))", attributes: .concurrent))
    }
    
    private let get: Get
    private let append: AppendObserver
    
    public convenience init(_ initial: A, queue: DispatchQueue? = nil) {
        var observers = Observers()
        var value: A = initial {
            didSet {
                observers.notifyObserversOfChange(from: oldValue, to: value)
            }
        }
        
        self.init(queue: queue, get: { value }) { observer in
            let id = observers.nextIdentifier()
            observers[id] = observer
            
            return Disposable {
                observers[id] = nil
            }
        }
    }
    
    private convenience init(queue: DispatchQueue?, get: @escaping Get, append: @escaping AppendObserver) {
        if let q = queue {
            self.init(queue: q, get: get, append: append)
        } else {
            self.init(get: get, append: append)
        }
    }
    
    internal init(queue: DispatchQueue, get: @escaping Get, append: @escaping AppendObserver) {
        self.get = { queue.sync { get() }}
        self.append = { observer in queue.sync { append(observer) }}
    }
    
    internal init(get: @escaping Get, append: @escaping AppendObserver) {
        self.get = get
        self.append = append
    }
    
    
    public func addObserver(_ observer: @escaping Observer) -> Disposable {
        return append(observer)
    }
    
    public var value: A {
        return get()
    }
    
    
    /// Subscripting
    
    public subscript<B>(keyPath: KeyPath<A,B>) -> Variable<B> {
        return Variable<B>(get: { self.value[keyPath: keyPath] }) { observer in
            self.append { new, old in
                observer(new[keyPath: keyPath], old[keyPath: keyPath])
            }
        }
    }
    
    
    /// Predicated Observers
    
    public typealias Predicate = (_ new: A, _ old: A) -> Bool
    
    public func addObserver(where predicate: @escaping Predicate, observer: @escaping Observer) -> Disposable {
        return addObserver { new, old in
            guard predicate(new, old) else {
                return
            }
            
            observer(new, old)
        }
    }
}

extension Variable where A: Equatable {
    public func addChangeObserver(_ observer: @escaping Observer) -> Disposable {
        return addObserver(where: !=, observer: observer)
    }
}