MessageQueueObservableCacheFactory implements ObservableCacheFactory<K, V> { @Override public Mache<K, V> createCache(final Mache<K, V> underlyingCache) { final ObservableMap<K, V> observable = new ObservableMap<>(communicationFactory, underlyingCache, uuidUtils); observable.registerListener(communicationFactory.getProducer(configuration)); try { BaseCoordinationEntryEventConsumer<K> consumer = communicationFactory.getConsumer(configuration); consumer.registerEventListener(new RemoteCacheEntryListener<K>() { @Override public void onCreated(Iterable<CoordinationEntryEvent<K>> events) throws CacheEntryListenerException { handle(events); } @Override public void onInvalidate(Iterable<CoordinationEntryEvent<K>> events) throws CacheEntryListenerException { handle(events); } @Override public void onRemoved(Iterable<CoordinationEntryEvent<K>> events) throws CacheEntryListenerException { handle(events); } @Override public void onUpdated(Iterable<CoordinationEntryEvent<K>> events) throws CacheEntryListenerException { handle(events); } public void handle(Iterable<CoordinationEntryEvent<K>> events) { for (final CoordinationEntryEvent<K> e : events) { if (e.getEntityName().equals(underlyingCache.getName()) && !e.getCacheId().equals(underlyingCache.getId())) { K key = e.getKey(); underlyingCache.invalidate(key); } } } }); consumer.beginSubscriptionThread(); } catch (Exception e) { throw new RuntimeException("Error creating cache consumer from underlying cache " + underlyingCache, e); } return observable; } MessageQueueObservableCacheFactory(final MQFactory<K> communicationFactory,
                                              final MQConfiguration configuration,
                                              final UuidUtils uuidUtils); @Override Mache<K, V> createCache(final Mache<K, V> underlyingCache);  }
@Test public void shouldProperlySetupCachesUsingSameCacheLoader() throws ExecutionException, InterruptedException { Mache<String, TestEntity> cache1 = observableCacheFactory1.createCache( macheFactory.create(String.class, TestEntity.class, cacheLoader1)); cache1.put(testValue.pkey, testValue); Mache<String, TestEntity> cache2 = observableCacheFactory2.createCache( macheFactory.create(String.class, TestEntity.class, cacheLoader1)); assertEquals(testValue, cache2.get(testValue.pkey)); }
@SuppressWarnings("unchecked") @Test public void shouldProperlyInvalidateFromAnotherCacheWhenItemPut() throws Exception { Mache<String, TestEntity> cache2 = observableCacheFactory2.createCache( macheFactory.create(String.class, TestEntity.class, cacheLoader2)); reset(spiedCache1); cache2.put(testValue2.pkey, testValue2); verify(spiedCache1).invalidate(testValue2.pkey); }
@Test public void shouldProperlyPropagateValues() throws Exception { MacheLoader<String, TestEntity2> cacheLoader = new HashMapCacheLoader<>(TestEntity2.class); ObservableCacheFactory<String, TestEntity2> observableCacheFactory1 = new MessageQueueObservableCacheFactory<>(mqFactory1, mqConfiguration, new UuidUtils()); ObservableCacheFactory<String, TestEntity2> observableCacheFactory2 = new MessageQueueObservableCacheFactory<>(mqFactory2, mqConfiguration, new UuidUtils()); Mache<String, TestEntity2> cache1 = observableCacheFactory1.createCache( macheFactory2.create(String.class, TestEntity2.class, cacheLoader)); Mache<String, TestEntity2> cache2 = observableCacheFactory2.createCache( macheFactory2.create(String.class, TestEntity2.class, cacheLoader)); final String key1 = "X1"; final String val1 = "someValue1"; cache1.put(key1, new TestEntity2(key1, val1)); assertEquals(val1, cache2.get(key1).otherValue); final String val2 = "someValue2"; cache1.put(key1, new TestEntity2(key1, val2)); assertEquals(val2, cache2.get(key1).otherValue); }
@SuppressWarnings("unchecked") @Test public void shouldNotInvalidateFromAnotherCacheWhenItemFetched() throws Exception { Mache<String, TestEntity> cache1 = observableCacheFactory1.createCache( macheFactory.create(String.class, TestEntity.class, cacheLoader1)); cache1.put(testValue2.pkey, testValue2); assertNotNull(cache1.get(testValue2.pkey)); reset(spiedCache1); Mache<String, TestEntity> cache2 = observableCacheFactory2.createCache( macheFactory.create(String.class, TestEntity.class, cacheLoader1)); assertNotNull(cache2.get(testValue2.pkey)); verify(spiedCache1, never()).invalidate(testValue2.pkey); }