greg/helenus
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Formatting and fixes to use MapCache in the UOW.

Browse source
This commit is contained in:
Greg Burd 2018-02-10 12:32:54 -05:00
parent b023ec359b
commit af4156079d
4 changed files with 363 additions and 390 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
src/main/java/net/helenus/core/AbstractUnitOfWork.java
View file

@ -24,12 +24,13 @@ import com.google.common.collect.TreeTraverser;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import javax.cache.Cache;
import javax.cache.CacheManager;
import javax.cache.integration.CacheLoader;
import javax.cache.integration.CacheLoaderException;
import net.helenus.core.cache.CacheUtil;
import net.helenus.core.cache.Facet;
import net.helenus.core.cache.MapCache;
@ -53,7 +54,7 @@ public abstract class AbstractUnitOfWork<E extends Exception>
private final HelenusSession session;
public final AbstractUnitOfWork<E> parent;
private final Table<String, String, Either<Object, List<Facet>>> cache = HashBasedTable.create();
private final MapCache<String, Object> statementCache = new MapCache<String, Object>(null, "UOW(" + hashCode() + ")", this);
private final MapCache<String, Object> statementCache;
protected String purpose;
protected List<String> nestedPurposes = new ArrayList<String>();
protected String info;
@ -76,6 +77,31 @@ public abstract class AbstractUnitOfWork<E extends Exception>
this.session = session;
this.parent = parent;
CacheLoader cacheLoader = null;
if (parent != null) {
cacheLoader =
new CacheLoader<String, Object>() {
Cache<String, Object> cache = parent.getCache();
@Override
public Object load(String key) throws CacheLoaderException {
return cache.get(key);
}
@Override
public Map<String, Object> loadAll(Iterable<? extends String> keys)
throws CacheLoaderException {
Map<String, Object> kvp = new HashMap<String, Object>();
for (String key : keys) {
kvp.put(key, cache.get(key));
}
return kvp;
}
};
}
this.statementCache =
new MapCache<String, Object>(null, "UOW(" + hashCode() + ")", cacheLoader, true);
}
@Override
@ -378,7 +404,8 @@ public abstract class AbstractUnitOfWork<E extends Exception>
// Merge our statement cache into the session cache if it exists.
CacheManager cacheManager = session.getCacheManager();
if (cacheManager != null) {
for (Map.Entry<String, Object> entry : (Set<Map.Entry<String, Object>>)statementCache.<Map>unwrap(Map.class).entrySet()) {
for (Map.Entry<String, Object> entry :
(Set<Map.Entry<String, Object>>) statementCache.<Map>unwrap(Map.class).entrySet()) {
String[] keyParts = entry.getKey().split("\\.");
if (keyParts.length == 2) {
String cacheName = keyParts[0];

2
src/main/java/net/helenus/core/UnitOfWork.java
View file

@ -20,9 +20,7 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;
import javax.cache.Cache;
import net.helenus.core.cache.Facet;
import net.helenus.core.operation.AbstractOperation;

198
src/main/java/net/helenus/core/cache/MapCache.java vendored
View file

@ -1,6 +1,5 @@
package net.helenus.core.cache;
import static net.helenus.core.HelenusSession.deleted;
import java.util.HashMap;
import java.util.HashSet;
@ -8,7 +7,6 @@ import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import javax.cache.Cache;
import javax.cache.CacheManager;
import javax.cache.configuration.CacheEntryListenerConfiguration;
@ -21,13 +19,9 @@ import javax.cache.processor.EntryProcessorException;
import javax.cache.processor.EntryProcessorResult;
import javax.cache.processor.MutableEntry;
import net.helenus.core.AbstractUnitOfWork;
import net.helenus.core.UnitOfWork;
public class MapCache<K, V> implements Cache<K, V> {
private final CacheManager manager;
private final String name;
private final UnitOfWork uow;
private Map<K, V> map = new ConcurrentHashMap<K, V>();
private Set<CacheEntryRemovedListener> cacheEntryRemovedListeners = new HashSet<>();
private CacheLoader<K, V> cacheLoader = null;
@ -36,50 +30,40 @@ public class MapCache<K, V> implements Cache<K, V> {
private static class MapConfiguration<K, V> implements Configuration<K, V> {
@Override public Class<K> getKeyType() {
@Override
public Class<K> getKeyType() {
return null;
}
@Override public Class<V> getValueType() {
@Override
public Class<V> getValueType() {
return null;
}
@Override public boolean isStoreByValue() {
@Override
public boolean isStoreByValue() {
return false;
}
}
public MapCache(CacheManager manager, String name, UnitOfWork uow) {
public MapCache(
CacheManager manager, String name, CacheLoader<K, V> cacheLoader, boolean isReadThrough) {
this.manager = manager;
this.name = name;
this.uow = uow;
this.cacheLoader = cacheLoader;
this.isReadThrough = isReadThrough;
}
private V map_get(K key) {
V value = null;
AbstractUnitOfWork uow = (AbstractUnitOfWork)this.uow;
do {
V result = (V) uow.getCache().get(key);
if (result != null) {
return result == deleted ? null : result;
}
uow = uow.parent;
} while (uow != null);
return null;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public V get(K key) {
V value = null;
synchronized (map) {
value = map_get(key);
value = map.get(key);
if (value == null && isReadThrough && cacheLoader != null) {
V loadedValue = cacheLoader.load(key);
if (loadedValue != null) {
map.put(key, value);
map.put(key, loadedValue);
value = loadedValue;
}
}
@ -87,16 +71,14 @@ public class MapCache<K, V> implements Cache<K, V> {
return value;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public Map<K, V> getAll(Set<? extends K> keys) {
Map<K, V> result = null;
synchronized (map) {
result = new HashMap<K, V>(keys.size());
for (K key : keys) {
V value = map_get(key);
V value = map.get(key);
if (value != null) {
result.put(key, value);
keys.remove(key);
@ -119,19 +101,16 @@ public class MapCache<K, V> implements Cache<K, V> {
return result;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public boolean containsKey(K key) {
return map.containsKey(key);
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public void loadAll(Set<? extends K> keys, boolean replaceExistingValues, CompletionListener completionListener) {
public void loadAll(
Set<? extends K> keys, boolean replaceExistingValues, CompletionListener completionListener) {
if (cacheLoader != null) {
try {
synchronized (map) {
@ -161,22 +140,18 @@ public class MapCache<K, V> implements Cache<K, V> {
}
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public void put(K key, V value) {
map.put(key, value);
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public V getAndPut(K key, V value) {
V result = null;
synchronized (map) {
result = map_get(key);
result = map.get(key);
if (value == null && isReadThrough && cacheLoader != null) {
V loadedValue = cacheLoader.load(key);
if (loadedValue != null) {
@ -189,9 +164,7 @@ public class MapCache<K, V> implements Cache<K, V> {
return result;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public void putAll(Map<? extends K, ? extends V> map) {
synchronized (map) {
@ -201,9 +174,7 @@ public class MapCache<K, V> implements Cache<K, V> {
}
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public boolean putIfAbsent(K key, V value) {
synchronized (map) {
@ -216,9 +187,7 @@ public class MapCache<K, V> implements Cache<K, V> {
}
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public boolean remove(K key) {
boolean removed = false;
@ -229,9 +198,7 @@ public class MapCache<K, V> implements Cache<K, V> {
return removed;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public boolean remove(K key, V oldValue) {
synchronized (map) {
@ -245,9 +212,7 @@ public class MapCache<K, V> implements Cache<K, V> {
return false;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public V getAndRemove(K key) {
synchronized (map) {
@ -259,9 +224,7 @@ public class MapCache<K, V> implements Cache<K, V> {
}
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public boolean replace(K key, V oldValue, V newValue) {
synchronized (map) {
@ -274,9 +237,7 @@ public class MapCache<K, V> implements Cache<K, V> {
return false;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public boolean replace(K key, V value) {
synchronized (map) {
@ -288,9 +249,7 @@ public class MapCache<K, V> implements Cache<K, V> {
return false;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public V getAndReplace(K key, V value) {
synchronized (map) {
@ -303,9 +262,7 @@ public class MapCache<K, V> implements Cache<K, V> {
return null;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public void removeAll(Set<? extends K> keys) {
synchronized (map) {
@ -320,9 +277,7 @@ public class MapCache<K, V> implements Cache<K, V> {
notifyRemovedListeners(keys);
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public void removeAll() {
synchronized (map) {
@ -332,17 +287,13 @@ public class MapCache<K, V> implements Cache<K, V> {
}
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public void clear() {
map.clear();
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public <C extends Configuration<K, V>> C getConfiguration(Class<C> clazz) {
if (!MapConfiguration.class.isAssignableFrom(clazz)) {
@ -351,18 +302,14 @@ public class MapCache<K, V> implements Cache<K, V> {
return null;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public <T> T invoke(K key, EntryProcessor<K, V, T> entryProcessor, Object... arguments)
throws EntryProcessorException {
return null;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public <T> Map<K, EntryProcessorResult<T>> invokeAll(
Set<? extends K> keys, EntryProcessor<K, V, T> entryProcessor, Object... arguments) {
@ -370,12 +317,15 @@ public class MapCache<K, V> implements Cache<K, V> {
for (K key : keys) {
V value = map.get(key);
if (value != null) {
entryProcessor.process(new MutableEntry<K, V>() {
@Override public boolean exists() {
entryProcessor.process(
new MutableEntry<K, V>() {
@Override
public boolean exists() {
return map.containsKey(key);
}
@Override public void remove() {
@Override
public void remove() {
synchronized (map) {
V value = map.get(key);
if (value != null) {
@ -385,86 +335,74 @@ public class MapCache<K, V> implements Cache<K, V> {
}
}
@Override public K getKey() {
@Override
public K getKey() {
return key;
}
@Override public V getValue() {
@Override
public V getValue() {
return map.get(value);
}
@Override public <T> T unwrap(Class<T> clazz) {
@Override
public <T> T unwrap(Class<T> clazz) {
return null;
}
@Override public void setValue(V value) {
@Override
public void setValue(V value) {
map.put(key, value);
}
}, arguments);
},
arguments);
}
}
}
return null;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public String getName() {
return name;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public CacheManager getCacheManager() {
return manager;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public void close() {
}
public void close() {}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public boolean isClosed() {
return false;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public <T> T unwrap(Class<T> clazz) {
return (T) map;
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public void registerCacheEntryListener(
CacheEntryListenerConfiguration<K, V> cacheEntryListenerConfiguration) {
//cacheEntryRemovedListeners.add(cacheEntryListenerConfiguration.getCacheEntryListenerFactory().create());
}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public void deregisterCacheEntryListener(
CacheEntryListenerConfiguration<K, V> cacheEntryListenerConfiguration) {}
/**
* {@inheritDoc}
*/
/** {@inheritDoc} */
@Override
public Iterator<Entry<K, V>> iterator() {
synchronized (map) {
@ -484,15 +422,18 @@ public class MapCache<K, V> implements Cache<K, V> {
K key = entry.getKey();
V value = entry.getValue();
@Override public K getKey() {
@Override
public K getKey() {
return key;
}
@Override public V getValue() {
@Override
public V getValue() {
return value;
}
@Override public <T> T unwrap(Class<T> clazz) {
@Override
public <T> T unwrap(Class<T> clazz) {
return null;
}
};
@ -512,8 +453,5 @@ public class MapCache<K, V> implements Cache<K, V> {
// }
}
private void notifyRemovedListeners(Set<? extends K> keys) {
}
private void notifyRemovedListeners(Set<? extends K> keys) {}
}

10
src/test/java/net/helenus/test/integration/core/unitofwork/UnitOfWorkTest.java
View file

@ -143,6 +143,8 @@ public class UnitOfWorkTest extends AbstractEmbeddedCassandraTest {
Widget w1, w1a, w2, w3, w4;
UUID key1 = UUIDs.timeBased();
UUID key2 = UUIDs.timeBased();
String cacheKey1 = MappingUtil.getTableName(Widget.class, false) + "." + key1.toString();
String cacheKey2 = MappingUtil.getTableName(Widget.class, false) + "." + key2.toString();
// This should inserted Widget, and not cache it in uow1.
try (UnitOfWork uow1 = session.begin()) {
@ -156,6 +158,8 @@ public class UnitOfWorkTest extends AbstractEmbeddedCassandraTest {
.value(widget::c, RandomString.make(10))
.value(widget::d, RandomString.make(10))
.sync(uow1);
uow1.getCache().put(cacheKey1, w1);
Assert.assertEquals(w1, uow1.getCache().get(cacheKey1));
try (UnitOfWork uow2 = session.begin(uow1)) {
@ -180,6 +184,7 @@ public class UnitOfWorkTest extends AbstractEmbeddedCassandraTest {
.orElse(null);
Assert.assertEquals(w1, w2);
uow2.getCache().put(cacheKey2, w2);
w3 =
session
@ -192,6 +197,8 @@ public class UnitOfWorkTest extends AbstractEmbeddedCassandraTest {
.value(widget::d, RandomString.make(10))
.sync(uow2);
Assert.assertEquals(w1, uow2.getCache().get(cacheKey1));
Assert.assertEquals(w2, uow2.getCache().get(cacheKey2));
uow2.commit()
.andThen(
() -> {
@ -378,6 +385,7 @@ public class UnitOfWorkTest extends AbstractEmbeddedCassandraTest {
Widget w1, w2, w3, w4, w5, w6;
Long committedAt = 0L;
UUID key = UUIDs.timeBased();
String cacheKey = MappingUtil.getTableName(Widget.class, false) + "." + key.toString();
try (UnitOfWork uow = session.begin()) {
w1 =
@ -392,6 +400,7 @@ public class UnitOfWorkTest extends AbstractEmbeddedCassandraTest {
.batch(uow);
Assert.assertTrue(0L == w1.writtenAt(widget::name));
Assert.assertTrue(0 == w1.ttlOf(widget::name));
uow.getCache().put(cacheKey, w1);
w2 =
session
.<Widget>update(w1)
@ -424,6 +433,7 @@ public class UnitOfWorkTest extends AbstractEmbeddedCassandraTest {
.value(widget::d, RandomString.make(10))
.batch(uow);
uow.getCache().put(cacheKey, w1);
uow.commit();
committedAt = uow.committedAt();
Date d = new Date(committedAt * 1000);