RxCache.swift

//
//  RxCache.swift
//
//  Created by Daniel Tartaglia on 12/20/18.
//  Copyright © 2018 Daniel Tartaglia. MIT License.
//

import RxSwift

final class Cache<Key: Hashable, State> {
    init(qos: DispatchQoS, source: @escaping (Key) -> Single<State>) {
        scheduler = SerialDispatchQueueScheduler(qos: qos)
        getState = source
    }

    func data(for key: Key) -> Single<State> {
        lock.lock(); defer { lock.unlock() }
        guard let state = cache[key] else {
            let state = ReplaySubject<State>.create(bufferSize: 1)
            getState(key)
                .observeOn(scheduler)
                .subscribe(onSuccess: { state.onNext($0) })
                .disposed(by: bag)
            cache[key] = state
            return state.asSingle()
        }

        return state.asSingle()
    }

    private var cache: [Key: ReplaySubject<State>] = [:]
    private let scheduler: SerialDispatchQueueScheduler
    private let lock = NSRecursiveLock()
    private let getState: (Key) -> Single<State>
    private let bag = DisposeBag()
}