Cache of shared observables (Rx)

Question

So the idea is to maintain a cache of shared Observables - accessible by a key - which if not present will produce one using the given factory.

The underlying shared Observable is accessed via an Observable created on each invocation of get - this is in order to be able to maintain ownership of the shared Observable and enforce when and how subscribe/unsubscribe is called on it.

When a subscriber is subscribed to an external Observable, then it looks up or created the underlying Observable, and subscribes to it.

When a subscriber is unsubscribed, if there are no more subscriptions on the underlying Observable, then it is removed from the cache.

If the underlying shared Observable terminates, then it is removed from the cache. I believe doOnTerminate is called before the onComplete/onError methods of subscribers, meaning that it will not be possible to subscribe to an Observable which has just terminated, and receive nothing in response.

It's written in Kotlin, but if need be I can port to Java.

I'd appreciate validation that what I've said above is true, and any edge cases I may have missed. Additionally, if there are any suggestions for making this non-blocking, without the chance of duplicating Observables for the same key, or having unmanaged references, then I'd be keen to listen.

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()
                })
    }
}

Answer 1

For your methods that only return a single expression, I would use the single expression syntax operator fun get(key: K) = Observable.create<T> (same for getSharedObservable). If you want to still have keep the explicit return type, you can do that, too.

For methods that take a lambda as the last parameter, the style guide recommend to put the lambda outside of the parans. I also would format it to have the first curly brace on the same line, resulting in

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

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

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

same for getSharedObservable

Answer 2

Turns out I've been over thinking it and as usual Google have provided the solution I'm really looking for in the Guava library.

package streams

import com.google.common.base.Function
import com.google.common.cache.CacheBuilder
import com.google.common.cache.CacheLoader
import rx.Observable

class ObservableCache<K, T>(private val factory: (K) -> Observable<T>) {

    private val cache = CacheBuilder
            .newBuilder()
            .softValues()
            .build(CacheLoader.from(Function<K, rx.Observable<T>> { factory.invoke(it).share() }))

    operator fun get(key: K) = cache.get(key)
}

Once all subscribers have been removed from the Observable in the cache, the Observable will disconnect. It will later be removed when the JVM decides it is a good time to do so. If it is required again, it will be recreated.

See:
http://docs.guava-libraries.googlecode.com/git/javadoc/com/google/common/cache/CacheBuilder.html#softValues()