Shares a number of underlying observables by provided keys Removes the underlying observable from the cache if there are no binding subscriptions to it, or if it terminates

SharedObservableCache.kt

package streams

import rx.Observable
import rx.observers.Subscribers.from
import rx.subscriptions.Subscriptions
import java.util.*
import java.util.concurrent.locks.ReentrantLock
import kotlin.concurrent.withLock

/**
 * Shares a number of underlying observables
 * Unsubscribes from and removes the source observable from the cache if there are no binding subscriptions to it,
 * or if the source observable terminates
 */
class SharedObservableCache<K, T>(private val factory: (K) -> Observable<T>) {

    private val sharedObservables = HashMap<K, Observable<T>>()
    private val lock = ReentrantLock()

    operator fun get(key: K): Observable<T> {
        return Observable.create<T>(
                { subscriber ->

                    val bindingSubscriber = from(subscriber)
                    lock.withLock {
                        getSharedObservable(key).subscribe(bindingSubscriber)
                    }

                    subscriber.add(Subscriptions.create(
                            {
                                lock.withLock {
                                    bindingSubscriber.unsubscribe()
                                }
                            }))
                })
    }

    private fun getSharedObservable(key: K): Observable<T> {
        return sharedObservables.getOrPut(key,
                {
                    factory.invoke(key)
                            // Occurs before calls to onComplete/onError to subscribers
                            // Removed at this point so as not to allow a subscriber to subscribe to a terminating observable
                            .doOnTerminate { lock.withLock { sharedObservables.remove(key) } }
                            // Locked by unsubscription of binding subscriber
                            .doOnUnsubscribe { sharedObservables.remove(key) }
                            .share()
                })
    }
}