Observable

Observable.swift

import Foundation

class Observable<T>: NSObject {
    private var subscriptions: Atomic<[WeakBox<Subscription>]> = Atomic([])

    fileprivate func fire(_ payload: T) {
        let targets = subscriptions.modify { value -> [Subscription] in
            value = value.filter { $0.value != nil }

            return value.compactMap { $0.value }
        }

        for target in targets {
            target.work(payload)
        }
    }

    // Keep reference to returned Token as long as you want to keep your subscription
    typealias Token = NSObject
    func observe(_ work: @escaping (T) -> Void) -> Token {
        return subscriptions.modify { value -> Token in
            let subscription = Subscription(work)
            value.append(WeakBox(value: subscription))
            subscription.lifetime += self // As long as at least one subscription survive, keep Observable alive
            return subscription
        }
    }

    class Publisher {
        fileprivate weak var observable: Observable?

        func fire(_ payload: T) {
            observable?.fire(payload)
        }
    }

    private class Subscription: NSObject {
        let work: (T) -> Void

        init(_ work: @escaping (T) -> Void) {
            self.work = work
        }
    }

    static func pipe() -> (Observable, Publisher) {
        let observable = Observable()
        let publisher = Publisher()
        publisher.observable = observable

        return (observable, publisher)
    }
}

// MARK: - Extensions

extension Observable {
    func asyncOnMainIfNeeded() -> Observable {
        let observable = Observable()

        observable.lifetime += observe { [weak observable] value in
            DispatchQueue.main.asyncIfNeeded {
                observable?.fire(value)
            }
        }

        return observable
    }
}

extension Observable where T: OptionalProtocol {
    func skipNil() -> Observable<T.Wrapped> {
        let observable = Observable<T.Wrapped>()

        observable.lifetime += observe { [weak observable] value in
            guard let observable = observable, let wrapped = value.optional else {
                return
            }

            observable.fire(wrapped)
        }

        return observable
    }
}

// MARK: - Misc

protocol OptionalProtocol {
    associatedtype Wrapped

    init(reconstructing value: Wrapped?)

    var optional: Wrapped? { get }
}

extension Optional: OptionalProtocol {
    var optional: Wrapped? {
        return self
    }

    init(reconstructing value: Wrapped?) {
        self = value
    }
}

private struct WeakBox<T: NSObjectProtocol> {
    weak var value: T?
}