A Guava Cache-based reactive store

CacheStore.java

import com.google.common.cache.Cache;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import rx.Observable;
import rx.subjects.BehaviorSubject;
import rx.subjects.PublishSubject;
import rx.subjects.Subject;

/**
 * A Cache-based reactive Store.
 */
public class CacheStore<K, V> {
    private final Cache<K, V> cache;
    private final Map<K, Subject<V, V>> subjectMap = new ConcurrentHashMap<>();

    public CacheStore(Cache<K, V> cache) {
        this.cache = cache;
    }

    public void invalidate(K key) {
        cache.invalidate(key);
    }

    public void put(K key, V value) {
        cache.put(key, value);
        if (subjectMap.containsKey(key)) {
            subjectMap.get(key).onNext(value);
        }
    }

    public void putAll(Map<? extends K,? extends V> m) {
        cache.putAll(m);
        for(Map.Entry<? extends K, ? extends V> entry : m.entrySet()) {
            if (subjectMap.containsKey(entry.getKey())) {
                subjectMap.get(entry.getKey()).onNext(entry.getValue());
            }
        }
    }

    public Observable<V> getStream(K key) {
        if (!subjectMap.containsKey(key)) {
            subjectMap.put(key, PublishSubject.create());
        }
        V cachedValue = cache.getIfPresent(key);
        if (cachedValue != null) {
            // Give the last value we have to the subscriber immediately, via BehaviorSubject.
            final Subject<V, V> subject = BehaviorSubject.create(cachedValue);
            // Subscribe the subject we're returning to the one in the map.
            subjectMap.get(key).subscribe(subject);
            return subject;
        } else {
            // Return the subject that will give the subscriber all subsequent updates to the
            // cache value.
            return subjectMap.get(key);
        }
    }

    public boolean isInCache(K key) {
        return cache.getIfPresent(key) != null;
    }
}

Do you want to use this cache in multi-threaded environments? If yes, you may run into NPE right here:

        if (subjectMap.containsKey(key)) {
            subjectMap.get(key).onNext(value);
        }