From 96cc8d760591fc153a8398f2d26e29e4eab4f7fe Mon Sep 17 00:00:00 2001 From: rob bygrave Date: Fri, 15 Jun 2018 17:36:23 +1200 Subject: [PATCH 1/4] #1427 - QueryCache should be cleared, if one of a dependent bean is updated Initial work, does not include propagation across cluster. --- pom.xml | 2 +- .../java/io/ebean/cache/QueryCacheEntry.java | 54 +++++++++++++ .../ebean/cache/QueryCacheEntryValidate.java | 12 +++ .../io/ebean/cache/ServerCacheConfig.java | 65 +++++++++++++++ .../io/ebean/cache/ServerCacheFactory.java | 4 +- .../io/ebeaninternal/api/SpiTransaction.java | 5 ++ .../api/SpiTransactionProxy.java | 5 ++ .../ebeaninternal/api/TransactionEvent.java | 9 ++- .../server/cache/CacheChangeSet.java | 49 +++++++----- .../server/cache/CacheManagerOptions.java | 10 ++- .../server/cache/DefaultCacheHolder.java | 26 +++--- .../server/cache/DefaultServerCache.java | 72 ++++++----------- .../cache/DefaultServerCacheConfig.java | 80 +++++++++++++++++++ .../cache/DefaultServerCacheFactory.java | 14 ++-- .../server/cache/DefaultServerQueryCache.java | 41 ++++++++++ .../server/core/CacheOptions.java | 24 ++++++ .../server/core/DefaultContainer.java | 51 +----------- .../server/core/InternalConfiguration.java | 78 ++++++++++++++++-- .../server/core/OrmQueryRequest.java | 20 ++++- .../server/deploy/BeanDescriptor.java | 5 +- .../deploy/BeanDescriptorCacheHelp.java | 75 ++++++++++------- .../deploy/meta/DeployBeanDescriptor.java | 7 ++ .../server/deploy/parse/AnnotationClass.java | 14 +++- .../io/ebeaninternal/server/query/CQuery.java | 5 ++ .../server/query/CQueryEngine.java | 19 +++++ .../query/CQueryFetchSingleAttribute.java | 5 ++ .../server/query/CQueryPlan.java | 10 +++ .../server/query/CQueryRowCount.java | 5 ++ .../server/query/DefaultOrmQueryEngine.java | 27 +------ .../ebeaninternal/server/query/SqlTree.java | 10 +++ .../server/query/SqlTreeNode.java | 6 ++ .../server/query/SqlTreeNodeBean.java | 9 +++ .../server/query/SqlTreeNodeExtraJoin.java | 11 +++ .../server/query/SqlTreeNodeManyRoot.java | 1 + .../query/SqlTreeNodeManyWhereJoin.java | 6 ++ .../server/query/SqlTreeNodeRoot.java | 7 ++ .../ImplicitReadOnlyTransaction.java | 8 ++ .../server/transaction/JdbcTransaction.java | 12 ++- .../server/transaction/NoTransaction.java | 6 ++ .../transaction/PostCommitProcessing.java | 12 ++- .../server/transaction/TableModState.java | 51 ++++++++++++ .../transaction/TransactionManager.java | 21 +++-- .../TransactionManagerOptions.java | 6 +- .../server/cache/DefaultCacheHolderTest.java | 60 ++++++++------ ...faultCacheHolder_getCacheOptions_Test.java | 6 +- .../cache/DefaultServerCacheConfigTest.java | 52 ++++++++++++ .../server/cache/DefaultServerCacheTest.java | 44 ++-------- .../DefaultServerCache_RunEvictionTest.java | 5 +- .../server/transaction/TableModStateTest.java | 44 ++++++++++ .../cache/TestQueryCacheTableDependency.java | 59 ++++++++++++++ .../java/org/tests/model/basic/Address.java | 6 ++ 51 files changed, 952 insertions(+), 283 deletions(-) create mode 100644 src/main/java/io/ebean/cache/QueryCacheEntry.java create mode 100644 src/main/java/io/ebean/cache/QueryCacheEntryValidate.java create mode 100644 src/main/java/io/ebean/cache/ServerCacheConfig.java create mode 100644 src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheConfig.java create mode 100644 src/main/java/io/ebeaninternal/server/cache/DefaultServerQueryCache.java create mode 100644 src/main/java/io/ebeaninternal/server/transaction/TableModState.java create mode 100644 src/test/java/io/ebeaninternal/server/cache/DefaultServerCacheConfigTest.java create mode 100644 src/test/java/io/ebeaninternal/server/transaction/TableModStateTest.java create mode 100644 src/test/java/org/tests/cache/TestQueryCacheTableDependency.java diff --git a/pom.xml b/pom.xml index c67edab03..4e54f6d66 100644 --- a/pom.xml +++ b/pom.xml @@ -117,7 +117,7 @@ io.ebean ebean-annotation - 3.11 + 3.12 diff --git a/src/main/java/io/ebean/cache/QueryCacheEntry.java b/src/main/java/io/ebean/cache/QueryCacheEntry.java new file mode 100644 index 000000000..31f1134d9 --- /dev/null +++ b/src/main/java/io/ebean/cache/QueryCacheEntry.java @@ -0,0 +1,54 @@ +package io.ebean.cache; + +import java.util.Set; + +/** + * For query cache entries we additionally hold the dependent tables and timestamp for the query result. + *

+ * We use the dependent tables and timestamp to validate that tables the query joins to have not been + * modified since the query cache entry was cached. If any dependent tables have since been modified + * the query cache entry is treated as invalid. + *

+ */ +public class QueryCacheEntry { + + private final Object value; + + private final Set dependentTables; + + private final long timestamp; + + /** + * Create with dependent tables and timestamp. + * + * @param value The query result being cached + * @param dependentTables The extra tables the query is dependent on (joins to) + * @param timestamp The timestamp that the query uses to check for modifications + */ + public QueryCacheEntry(Object value, Set dependentTables, long timestamp) { + this.value = value; + this.dependentTables = dependentTables; + this.timestamp = timestamp; + } + + /** + * Return the actual query result. + */ + public Object getValue() { + return value; + } + + /** + * Return the tables the query result is dependent on. + */ + public Set getDependentTables() { + return dependentTables; + } + + /** + * Return the timestamp used to check for modifications on the dependent tables. + */ + public long getTimestamp() { + return timestamp; + } +} diff --git a/src/main/java/io/ebean/cache/QueryCacheEntryValidate.java b/src/main/java/io/ebean/cache/QueryCacheEntryValidate.java new file mode 100644 index 000000000..f94c07a7d --- /dev/null +++ b/src/main/java/io/ebean/cache/QueryCacheEntryValidate.java @@ -0,0 +1,12 @@ +package io.ebean.cache; + +/** + * Used to validate that a query cache entry is still valid based on dependent tables. + */ +public interface QueryCacheEntryValidate { + + /** + * Return true if the entry is still valid based on dependent tables. + */ + boolean isValid(QueryCacheEntry queryCacheEntry); +} diff --git a/src/main/java/io/ebean/cache/ServerCacheConfig.java b/src/main/java/io/ebean/cache/ServerCacheConfig.java new file mode 100644 index 000000000..a2a6d5c52 --- /dev/null +++ b/src/main/java/io/ebean/cache/ServerCacheConfig.java @@ -0,0 +1,65 @@ +package io.ebean.cache; + +import io.ebean.config.CurrentTenantProvider; + +/** + * Configuration used to create ServerCache instances. + */ +public class ServerCacheConfig { + + private final ServerCacheType type; + private final String cacheKey; + private final ServerCacheOptions cacheOptions; + private final CurrentTenantProvider tenantProvider; + private final QueryCacheEntryValidate queryCacheEntryValidate; + + public ServerCacheConfig(ServerCacheType type, String cacheKey, ServerCacheOptions cacheOptions, CurrentTenantProvider tenantProvider, QueryCacheEntryValidate queryCacheEntryValidate) { + this.type = type; + this.cacheKey = cacheKey; + this.cacheOptions = cacheOptions; + this.tenantProvider = tenantProvider; + this.queryCacheEntryValidate = queryCacheEntryValidate; + } + + /** + * Return the cache type. + */ + public ServerCacheType getType() { + return type; + } + + /** + * Return the name of the cache. + */ + public String getCacheKey() { + return cacheKey; + } + + /** + * Return the tuning options. + */ + public ServerCacheOptions getCacheOptions() { + return cacheOptions; + } + + /** + * Return the current tenant provider. + */ + public CurrentTenantProvider getTenantProvider() { + return tenantProvider; + } + + /** + * Return the service that provides validation for query cache entries. + */ + public QueryCacheEntryValidate getQueryCacheEntryValidate() { + return queryCacheEntryValidate; + } + + /** + * Return true if the cache is a query cache. + */ + public boolean isQueryCache() { + return type == ServerCacheType.QUERY; + } +} diff --git a/src/main/java/io/ebean/cache/ServerCacheFactory.java b/src/main/java/io/ebean/cache/ServerCacheFactory.java index 269cb9ae1..d59bfd3c4 100644 --- a/src/main/java/io/ebean/cache/ServerCacheFactory.java +++ b/src/main/java/io/ebean/cache/ServerCacheFactory.java @@ -1,7 +1,5 @@ package io.ebean.cache; -import io.ebean.config.CurrentTenantProvider; - /** * Defines method for constructing caches for beans and queries. */ @@ -10,6 +8,6 @@ public interface ServerCacheFactory { /** * Create the cache for the given type with options. */ - ServerCache createCache(ServerCacheType type, String cacheKey, CurrentTenantProvider tenantProvider, ServerCacheOptions cacheOptions); + ServerCache createCache(ServerCacheConfig config); } diff --git a/src/main/java/io/ebeaninternal/api/SpiTransaction.java b/src/main/java/io/ebeaninternal/api/SpiTransaction.java index e134986c5..0054757da 100644 --- a/src/main/java/io/ebeaninternal/api/SpiTransaction.java +++ b/src/main/java/io/ebeaninternal/api/SpiTransaction.java @@ -101,6 +101,11 @@ public interface SpiTransaction extends Transaction { */ String getId(); + /** + * Return the start timestamp for the transaction (JVM side). + */ + long getStartMillis(); + /** * Return true if this transaction has updateAllLoadedProperties set. * If null is returned the server default is used (set on ServerConfig). diff --git a/src/main/java/io/ebeaninternal/api/SpiTransactionProxy.java b/src/main/java/io/ebeaninternal/api/SpiTransactionProxy.java index 02057236c..ffbb3c727 100644 --- a/src/main/java/io/ebeaninternal/api/SpiTransactionProxy.java +++ b/src/main/java/io/ebeaninternal/api/SpiTransactionProxy.java @@ -30,6 +30,11 @@ public abstract class SpiTransactionProxy implements SpiTransaction { return transaction.translate(message, cause); } + @Override + public long getStartMillis() { + return transaction.getStartMillis(); + } + @Override public void setLabel(String label) { transaction.setLabel(label); diff --git a/src/main/java/io/ebeaninternal/api/TransactionEvent.java b/src/main/java/io/ebeaninternal/api/TransactionEvent.java index 44ba7cf4d..711fa49dd 100644 --- a/src/main/java/io/ebeaninternal/api/TransactionEvent.java +++ b/src/main/java/io/ebeaninternal/api/TransactionEvent.java @@ -27,6 +27,8 @@ public class TransactionEvent implements Serializable { */ private final transient boolean local; + private final long modificationTimestamp; + private TransactionEventTable eventTables; private transient TransactionEventBeans eventBeans; @@ -36,8 +38,9 @@ public class TransactionEvent implements Serializable { /** * Create the TransactionEvent, one per Transaction. */ - public TransactionEvent() { + public TransactionEvent(long modificationTimestamp) { this.local = true; + this.modificationTimestamp = modificationTimestamp; } public void addDeleteById(BeanDescriptor desc, Object id) { @@ -108,8 +111,8 @@ public class TransactionEvent implements Serializable { /** * Build and return the cache changeSet. */ - public CacheChangeSet buildCacheChanges(boolean viewInvalidation) { - CacheChangeSet changeSet = new CacheChangeSet(viewInvalidation); + public CacheChangeSet buildCacheChanges() { + CacheChangeSet changeSet = new CacheChangeSet(modificationTimestamp); if (eventBeans != null) { eventBeans.notifyCache(changeSet); } diff --git a/src/main/java/io/ebeaninternal/server/cache/CacheChangeSet.java b/src/main/java/io/ebeaninternal/server/cache/CacheChangeSet.java index f0c09f6cb..45956d181 100644 --- a/src/main/java/io/ebeaninternal/server/cache/CacheChangeSet.java +++ b/src/main/java/io/ebeaninternal/server/cache/CacheChangeSet.java @@ -17,22 +17,26 @@ public class CacheChangeSet { private final List entries = new ArrayList<>(); + private final Set touchedTables = new HashSet<>(); + private final Set> queryCaches = new HashSet<>(); private final Map manyChangeMap = new HashMap<>(); - /** - * Set of "base tables" modified used to invalidate entities based on views. - */ - private final Set viewInvalidation = new HashSet<>(); - - private final boolean viewEntityInvalidation; + private final long modificationTimestamp; /** * Construct specifying if we also need to process invalidation for entities based on views. */ - public CacheChangeSet(boolean viewEntityInvalidation) { - this.viewEntityInvalidation = viewEntityInvalidation; + public CacheChangeSet(long modificationTimestamp) { + this.modificationTimestamp = modificationTimestamp; + } + + /** + * Return the touched tables. + */ + public Set touchedTables() { + return touchedTables; } /** @@ -40,7 +44,7 @@ public class CacheChangeSet { *

* Return the set of table changes to process invalidation for entities based on views. */ - public Set apply() { + public void apply() { for (BeanDescriptor entry : queryCaches) { entry.clearQueryCache(); } @@ -50,7 +54,13 @@ public class CacheChangeSet { for (CacheChange entry : manyChangeMap.values()) { entry.apply(); } - return viewInvalidation; + } + + /** + * Add an entry to clear a query cache. + */ + public void addInvalidate(BeanDescriptor descriptor) { + touchedTables.add(descriptor.getBaseTable()); } /** @@ -85,29 +95,23 @@ public class CacheChangeSet { * On bean insert register table for view based entity invalidation. */ public void addBeanInsert(String baseTable) { - if (viewEntityInvalidation) { - viewInvalidation.add(baseTable); - } + touchedTables.add(baseTable); } /** * Remove a bean from the cache. */ public void addBeanRemove(BeanDescriptor desc, Object id) { + touchedTables.add(desc.getBaseTable()); entries.add(new CacheChangeBeanRemove(desc, id)); - if (viewEntityInvalidation) { - viewInvalidation.add(desc.getBaseTable()); - } } /** * Update a bean entry. */ public void addBeanUpdate(BeanDescriptor desc, Object id, Map changes, boolean updateNaturalKey, long version) { + touchedTables.add(desc.getBaseTable()); entries.add(new CacheChangeBeanUpdate(desc, id, changes, updateNaturalKey, version)); - if (viewEntityInvalidation) { - viewInvalidation.add(desc.getBaseTable()); - } } /** @@ -125,6 +129,13 @@ public class CacheChangeSet { return manyChangeMap.computeIfAbsent(key, ManyChange::new); } + /** + * Return the modification timestamp for these changes. + */ + public long modificationTimestamp() { + return modificationTimestamp; + } + /** * Changes for a specific many property. */ diff --git a/src/main/java/io/ebeaninternal/server/cache/CacheManagerOptions.java b/src/main/java/io/ebeaninternal/server/cache/CacheManagerOptions.java index b61b48080..36a7ed571 100644 --- a/src/main/java/io/ebeaninternal/server/cache/CacheManagerOptions.java +++ b/src/main/java/io/ebeaninternal/server/cache/CacheManagerOptions.java @@ -1,5 +1,6 @@ package io.ebeaninternal.server.cache; +import io.ebean.cache.QueryCacheEntryValidate; import io.ebean.cache.ServerCacheFactory; import io.ebean.cache.ServerCacheOptions; import io.ebean.config.CurrentTenantProvider; @@ -19,6 +20,8 @@ public class CacheManagerOptions { private CurrentTenantProvider currentTenantProvider; + private QueryCacheEntryValidate queryCacheEntryValidate; + private ServerCacheFactory cacheFactory = new DefaultServerCacheFactory(); private ServerCacheOptions beanDefault = new ServerCacheOptions(); private ServerCacheOptions queryDefault = new ServerCacheOptions(); @@ -45,8 +48,9 @@ public class CacheManagerOptions { return this; } - public CacheManagerOptions with(ServerCacheFactory cacheFactory) { + public CacheManagerOptions with(ServerCacheFactory cacheFactory, QueryCacheEntryValidate queryCacheEntryValidate) { this.cacheFactory = cacheFactory; + this.queryCacheEntryValidate = queryCacheEntryValidate; return this; } @@ -82,4 +86,8 @@ public class CacheManagerOptions { public ClusterManager getClusterManager() { return clusterManager; } + + public QueryCacheEntryValidate getQueryCacheEntryValidate() { + return queryCacheEntryValidate; + } } diff --git a/src/main/java/io/ebeaninternal/server/cache/DefaultCacheHolder.java b/src/main/java/io/ebeaninternal/server/cache/DefaultCacheHolder.java index 092fe7b9a..20e966f6e 100644 --- a/src/main/java/io/ebeaninternal/server/cache/DefaultCacheHolder.java +++ b/src/main/java/io/ebeaninternal/server/cache/DefaultCacheHolder.java @@ -2,7 +2,9 @@ package io.ebeaninternal.server.cache; import io.ebean.annotation.CacheBeanTuning; import io.ebean.annotation.CacheQueryTuning; +import io.ebean.cache.QueryCacheEntryValidate; import io.ebean.cache.ServerCache; +import io.ebean.cache.ServerCacheConfig; import io.ebean.cache.ServerCacheFactory; import io.ebean.cache.ServerCacheOptions; import io.ebean.cache.ServerCacheType; @@ -33,22 +35,14 @@ class DefaultCacheHolder { private final CurrentTenantProvider tenantProvider; - DefaultCacheHolder(CacheManagerOptions builder) { - this(builder.getCacheFactory(), builder.getBeanDefault(), builder.getQueryDefault(), builder.getCurrentTenantProvider()); - } + private final QueryCacheEntryValidate queryCacheEntryValidate; - /** - * Create with a cache factory and default cache options. - * - * @param cacheFactory the factory for creating the cache - * @param beanDefault the default options for tuning bean caches - * @param queryDefault the default options for tuning query caches - */ - DefaultCacheHolder(ServerCacheFactory cacheFactory, ServerCacheOptions beanDefault, ServerCacheOptions queryDefault, CurrentTenantProvider tenantProvider) { - this.cacheFactory = cacheFactory; - this.beanDefault = beanDefault; - this.queryDefault = queryDefault; - this.tenantProvider = tenantProvider; + DefaultCacheHolder(CacheManagerOptions builder) { + this.cacheFactory = builder.getCacheFactory(); + this.beanDefault = builder.getBeanDefault(); + this.queryDefault = builder.getQueryDefault(); + this.tenantProvider = builder.getCurrentTenantProvider(); + this.queryCacheEntryValidate = builder.getQueryCacheEntryValidate(); } ServerCache getCache(Class beanType, String cacheKey, ServerCacheType type) { @@ -76,7 +70,7 @@ class DefaultCacheHolder { collectIdCaches.computeIfAbsent(beanType.getName(), s -> new ConcurrentSkipListSet<>()).add(key); } } - return cacheFactory.createCache(type, key, tenantProvider, options); + return cacheFactory.createCache(new ServerCacheConfig(type, key, options, tenantProvider, queryCacheEntryValidate)); } void clearAll() { diff --git a/src/main/java/io/ebeaninternal/server/cache/DefaultServerCache.java b/src/main/java/io/ebeaninternal/server/cache/DefaultServerCache.java index b8bee5cc4..33e85f34b 100644 --- a/src/main/java/io/ebeaninternal/server/cache/DefaultServerCache.java +++ b/src/main/java/io/ebeaninternal/server/cache/DefaultServerCache.java @@ -2,10 +2,8 @@ package io.ebeaninternal.server.cache; import io.ebean.BackgroundExecutor; import io.ebean.cache.ServerCache; -import io.ebean.cache.ServerCacheOptions; import io.ebean.cache.ServerCacheStatistics; import io.ebean.cache.TenantAwareKey; -import io.ebean.config.CurrentTenantProvider; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -15,7 +13,6 @@ import java.util.Comparator; import java.util.Iterator; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.LongAdder; @@ -65,47 +62,14 @@ public class DefaultServerCache implements ServerCache { protected TenantAwareKey tenantAwareKey; - /** - * Construct using a ConcurrentHashMap and cache options. - */ - public DefaultServerCache(String name, CurrentTenantProvider tenantProvider, ServerCacheOptions options) { - this(name, new ConcurrentHashMap<>(), tenantProvider, options); - } - - /** - * Construct passing in name, map and base eviction controls as ServerCacheOptions. - */ - public DefaultServerCache(String name, Map map, CurrentTenantProvider tenantProvider, ServerCacheOptions options) { - this(name, map, tenantProvider, options.getMaxSize(), options.getMaxIdleSecs(), options.getMaxSecsToLive(), options.getTrimFrequency()); - } - - /** - * Construct passing in name, map and base eviction controls. - */ - public DefaultServerCache(String name, Map map, CurrentTenantProvider tenantProvider, int maxSize, int maxIdleSecs, int maxSecsToLive, int trimFrequency) { - this.name = name; - this.map = map; - this.maxSize = maxSize; - this.tenantAwareKey = new TenantAwareKey(tenantProvider); - this.maxIdleSecs = maxIdleSecs; - this.maxSecsToLive = maxSecsToLive; - this.trimFrequency = determineTrim(maxIdleSecs, maxSecsToLive, trimFrequency); - } - - /** - * Determine a good trimFrequency as half of maxIdleSecs (or maxSecsToLive). - */ - int determineTrim(int maxIdleSecs, int maxSecsToLive, int trimFrequency) { - if (trimFrequency > 0) { - return trimFrequency; - } - if (maxIdleSecs > 0) { - return maxIdleSecs / 2 - 1; - } - if (maxSecsToLive > 0) { - return maxSecsToLive / 2 - 1; - } - return 0; + public DefaultServerCache(DefaultServerCacheConfig config) { + this.name = config.getName(); + this.map = config.getMap(); + this.maxSize = config.getMaxSize(); + this.tenantAwareKey = new TenantAwareKey(config.getTenantProvider()); + this.maxIdleSecs = config.getMaxIdleSecs(); + this.maxSecsToLive = config.getMaxSecsToLive(); + this.trimFrequency = config.determineTrimFrequency(); } public void periodicTrim(BackgroundExecutor executor) { @@ -193,7 +157,7 @@ public class DefaultServerCache implements ServerCache { /** * Return the tenant aware key. */ - private Object key(Object id) { + protected Object key(Object id) { return tenantAwareKey.key(id); } @@ -203,7 +167,7 @@ public class DefaultServerCache implements ServerCache { @Override public Object get(Object id) { - CacheEntry entry = map.get(key(id)); + CacheEntry entry = getCacheEntry(id); if (entry == null) { missCount.increment(); return null; @@ -212,10 +176,24 @@ public class DefaultServerCache implements ServerCache { // Important that hitCount.increment() MUST be low latency under concurrent // use hence must use LongAdder or better here hitCount.increment(); - return entry.getValue(); + return unwrapEntry(entry); } } + /** + * Unwrap the cache entry - override for query cache to unwrap to the query result. + */ + protected Object unwrapEntry(CacheEntry entry) { + return entry.getValue(); + } + + /** + * Get the cache entry - override for query cache to validate dependent tables. + */ + protected CacheEntry getCacheEntry(Object id) { + return map.get(key(id)); + } + @Override public void putAll(Map keyValues) { keyValues.forEach(this::put); diff --git a/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheConfig.java b/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheConfig.java new file mode 100644 index 000000000..44538beed --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheConfig.java @@ -0,0 +1,80 @@ +package io.ebeaninternal.server.cache; + +import io.ebean.cache.QueryCacheEntryValidate; +import io.ebean.cache.ServerCacheConfig; +import io.ebean.cache.ServerCacheOptions; +import io.ebean.config.CurrentTenantProvider; + +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +public class DefaultServerCacheConfig { + + private final ServerCacheConfig config; + + private int maxSize; + private int maxIdleSecs; + private int maxSecsToLive; + private int trimFrequency; + + private Map map; + + public DefaultServerCacheConfig(ServerCacheConfig config) { + this(config, new ConcurrentHashMap<>()); + } + + public DefaultServerCacheConfig(ServerCacheConfig config, Map map) { + this.config = config; + this.map = map; + + ServerCacheOptions options = config.getCacheOptions(); + this.maxIdleSecs = options.getMaxIdleSecs(); + this.maxSecsToLive = options.getMaxSecsToLive(); + this.trimFrequency = options.getTrimFrequency(); + this.maxSize = options.getMaxSize(); + } + + public CurrentTenantProvider getTenantProvider() { + return config.getTenantProvider(); + } + + public QueryCacheEntryValidate getQueryCacheEntryValidate() { + return config.getQueryCacheEntryValidate(); + } + + public String getName() { + return config.getCacheKey(); + } + + public Map getMap() { + return map; + } + + public int getMaxSize() { + return maxSize; + } + + public int getMaxIdleSecs() { + return maxIdleSecs; + } + + public int getMaxSecsToLive() { + return maxSecsToLive; + } + + /** + * Determine a good trimFrequency as half of maxIdleSecs (or maxSecsToLive). + */ + public int determineTrimFrequency() { + if (trimFrequency > 0) { + return trimFrequency; + } + if (maxIdleSecs > 0) { + return maxIdleSecs / 2 - 1; + } + if (maxSecsToLive > 0) { + return maxSecsToLive / 2 - 1; + } + return 0; + } +} diff --git a/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheFactory.java b/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheFactory.java index 197052dcb..a519ea469 100644 --- a/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheFactory.java +++ b/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheFactory.java @@ -2,10 +2,8 @@ package io.ebeaninternal.server.cache; import io.ebean.BackgroundExecutor; import io.ebean.cache.ServerCache; +import io.ebean.cache.ServerCacheConfig; import io.ebean.cache.ServerCacheFactory; -import io.ebean.cache.ServerCacheOptions; -import io.ebean.cache.ServerCacheType; -import io.ebean.config.CurrentTenantProvider; /** @@ -30,9 +28,15 @@ class DefaultServerCacheFactory implements ServerCacheFactory { } @Override - public ServerCache createCache(ServerCacheType type, String cacheKey, CurrentTenantProvider tenantProvider, ServerCacheOptions cacheOptions) { + public ServerCache createCache(ServerCacheConfig config) { - DefaultServerCache cache = new DefaultServerCache(cacheKey, tenantProvider, cacheOptions); + DefaultServerCache cache; + if (config.isQueryCache()) { + // use a server cache aware of extra validation and QueryCacheEntry + cache = new DefaultServerQueryCache(new DefaultServerCacheConfig(config)); + } else { + cache = new DefaultServerCache(new DefaultServerCacheConfig(config)); + } if (executor != null) { cache.periodicTrim(executor); } diff --git a/src/main/java/io/ebeaninternal/server/cache/DefaultServerQueryCache.java b/src/main/java/io/ebeaninternal/server/cache/DefaultServerQueryCache.java new file mode 100644 index 000000000..85f4b22ca --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/cache/DefaultServerQueryCache.java @@ -0,0 +1,41 @@ +package io.ebeaninternal.server.cache; + +import io.ebean.cache.QueryCacheEntry; +import io.ebean.cache.QueryCacheEntryValidate; + +/** + * Server cache for query caching. + *

+ * Entries in this cache contain QueryCacheEntry and we need to additionally + * validate the entries when hit for changes to dependent tables. + *

+ */ +public class DefaultServerQueryCache extends DefaultServerCache { + + private final QueryCacheEntryValidate queryCacheEntryValidate; + + public DefaultServerQueryCache(DefaultServerCacheConfig config) { + super(config); + this.queryCacheEntryValidate = config.getQueryCacheEntryValidate(); + } + + protected Object unwrapEntry(CacheEntry entry) { + return ((QueryCacheEntry) entry.getValue()).getValue(); + } + + @Override + protected CacheEntry getCacheEntry(Object id) { + Object key = key(id); + CacheEntry entry = map.get(key); + if (entry == null) { + return null; + } + QueryCacheEntry value = (QueryCacheEntry) entry.getValue(); + if (!queryCacheEntryValidate.isValid(value)) { + map.remove(key); + removeCount.increment(); + return null; + } + return entry; + } +} diff --git a/src/main/java/io/ebeaninternal/server/core/CacheOptions.java b/src/main/java/io/ebeaninternal/server/core/CacheOptions.java index 5cb7741a2..24d441828 100644 --- a/src/main/java/io/ebeaninternal/server/core/CacheOptions.java +++ b/src/main/java/io/ebeaninternal/server/core/CacheOptions.java @@ -12,6 +12,9 @@ public class CacheOptions { */ public static final CacheOptions NO_CACHING = new CacheOptions(); + public static final CacheOptions INVALIDATE_QUERY_CACHE = new CacheOptions(true); + + private final boolean invalidateQueryCache; private final boolean enableBeanCache; private final boolean enableQueryCache; private final boolean readOnly; @@ -21,6 +24,18 @@ public class CacheOptions { * Construct for no caching. */ private CacheOptions() { + invalidateQueryCache = false; + enableBeanCache = false; + enableQueryCache = false; + readOnly = false; + naturalKey = null; + } + + /** + * Construct for invalidateQueryCache. + */ + private CacheOptions(boolean invalidateQueryCache) { + this.invalidateQueryCache = invalidateQueryCache; enableBeanCache = false; enableQueryCache = false; readOnly = false; @@ -31,12 +46,21 @@ public class CacheOptions { * Construct with cache annotation. */ public CacheOptions(Cache cache, String[] naturalKey) { + invalidateQueryCache = false; enableBeanCache = cache.enableBeanCache(); enableQueryCache = cache.enableQueryCache(); readOnly = cache.readOnly(); this.naturalKey = naturalKey; } + /** + * Return true if this is InvalidateQueryCache. A Bean that itself isn't L2 + * cached but invalidates query cache entries that join to it. + */ + public boolean isInvalidateQueryCache() { + return invalidateQueryCache; + } + /** * Return true if bean caching is enabled. */ diff --git a/src/main/java/io/ebeaninternal/server/core/DefaultContainer.java b/src/main/java/io/ebeaninternal/server/core/DefaultContainer.java index a8cf20c08..0428a8893 100644 --- a/src/main/java/io/ebeaninternal/server/core/DefaultContainer.java +++ b/src/main/java/io/ebeaninternal/server/core/DefaultContainer.java @@ -138,11 +138,10 @@ public class DefaultContainer implements SpiContainer { // executor and l2 caching service setup early (used during server construction) SpiBackgroundExecutor executor = createBackgroundExecutor(serverConfig); - SpiCacheManager cacheManager = getCacheManager(online, serverConfig, executor); - InternalConfiguration c = new InternalConfiguration(clusterManager, cacheManager, executor, serverConfig, bootupClasses); + InternalConfiguration c = new InternalConfiguration(online, clusterManager, executor, serverConfig, bootupClasses); - DefaultServer server = new DefaultServer(c, c.cache()); + DefaultServer server = new DefaultServer(c, c.cacheManager()); // generate and run DDL if required // if there are any other tasks requiring action in their plugins, do them as well @@ -165,52 +164,6 @@ public class DefaultContainer implements SpiContainer { } } - /** - * Create and return the CacheManager. - */ - private SpiCacheManager getCacheManager(boolean online, ServerConfig serverConfig, BackgroundExecutor executor) { - - if (!online || serverConfig.isDisableL2Cache()) { - // use local only L2 cache implementation as placeholder - return new DefaultServerCacheManager(); - } - - // reasonable default settings are for a cache per bean type - ServerCacheOptions beanOptions = new ServerCacheOptions(); - beanOptions.setMaxSize(serverConfig.getCacheMaxSize()); - beanOptions.setMaxIdleSecs(serverConfig.getCacheMaxIdleTime()); - beanOptions.setMaxSecsToLive(serverConfig.getCacheMaxTimeToLive()); - - // reasonable default settings for the query cache per bean type - ServerCacheOptions queryOptions = new ServerCacheOptions(); - queryOptions.setMaxSize(serverConfig.getQueryCacheMaxSize()); - queryOptions.setMaxIdleSecs(serverConfig.getQueryCacheMaxIdleTime()); - queryOptions.setMaxSecsToLive(serverConfig.getQueryCacheMaxTimeToLive()); - - boolean localL2Caching = false; - ServerCachePlugin plugin = serverConfig.getServerCachePlugin(); - if (plugin == null) { - ServiceLoader cacheFactories = ServiceLoader.load(ServerCachePlugin.class); - Iterator iterator = cacheFactories.iterator(); - if (iterator.hasNext()) { - // use the cacheFactory (via classpath service loader) - plugin = iterator.next(); - logger.debug("using ServerCacheFactory {}", plugin.getClass()); - } else { - // use the built in default l2 caching which is local cache based - localL2Caching = true; - plugin = new DefaultServerCachePlugin(); - } - } - - ServerCacheFactory factory = plugin.create(serverConfig, executor); - - CacheManagerOptions builder = new CacheManagerOptions(clusterManager, serverConfig, localL2Caching) - .with(beanOptions, queryOptions) - .with(factory); - return new DefaultServerCacheManager(builder); - } - /** * Get the entities, scalarTypes, Listeners etc combining the class registered * ones with the already created instances. diff --git a/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java b/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java index 4ea1988a1..8d201d9c9 100644 --- a/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java +++ b/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java @@ -3,7 +3,10 @@ package io.ebeaninternal.server.core; import com.fasterxml.jackson.core.JsonFactory; import io.ebean.ExpressionFactory; import io.ebean.annotation.Platform; +import io.ebean.cache.ServerCacheFactory; import io.ebean.cache.ServerCacheManager; +import io.ebean.cache.ServerCacheOptions; +import io.ebean.cache.ServerCachePlugin; import io.ebean.config.ExternalTransactionManager; import io.ebean.config.ProfilingConfig; import io.ebean.config.ServerConfig; @@ -27,7 +30,10 @@ import io.ebeaninternal.api.SpiProfileHandler; import io.ebeaninternal.dbmigration.DbOffline; import io.ebeaninternal.server.autotune.AutoTuneService; import io.ebeaninternal.server.autotune.service.AutoTuneServiceFactory; +import io.ebeaninternal.server.cache.CacheManagerOptions; import io.ebeaninternal.server.cache.DefaultCacheAdapter; +import io.ebeaninternal.server.cache.DefaultServerCacheManager; +import io.ebeaninternal.server.cache.DefaultServerCachePlugin; import io.ebeaninternal.server.cache.SpiCacheManager; import io.ebeaninternal.server.changelog.DefaultChangeLogListener; import io.ebeaninternal.server.changelog.DefaultChangeLogPrepare; @@ -69,6 +75,7 @@ import io.ebeaninternal.server.transaction.ExplicitTransactionManager; import io.ebeaninternal.server.transaction.ExternalTransactionScopeManager; import io.ebeaninternal.server.transaction.JtaTransactionManager; import io.ebeaninternal.server.transaction.NoopProfileHandler; +import io.ebeaninternal.server.transaction.TableModState; import io.ebeaninternal.server.transaction.TransactionManager; import io.ebeaninternal.server.transaction.TransactionManagerOptions; import io.ebeaninternal.server.transaction.TransactionScopeManager; @@ -84,6 +91,7 @@ import org.slf4j.LoggerFactory; import javax.sql.DataSource; import java.util.ArrayList; +import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.ServiceLoader; @@ -96,6 +104,10 @@ public class InternalConfiguration { private static final Logger logger = LoggerFactory.getLogger(InternalConfiguration.class); + private final TableModState tableModState = new TableModState(); + + private final boolean online; + private final ServerConfig serverConfig; private final BootupClasses bootupClasses; @@ -124,6 +136,10 @@ public class InternalConfiguration { private final SpiCacheManager cacheManager; + private final ServerCachePlugin serverCachePlugin; + + private boolean localL2Caching; + private final ExpressionFactory expressionFactory; private final SpiBackgroundExecutor backgroundExecutor; @@ -141,17 +157,16 @@ public class InternalConfiguration { private final SpiLogManager logManager; - public InternalConfiguration(ClusterManager clusterManager, - SpiCacheManager cacheManager, SpiBackgroundExecutor backgroundExecutor, + public InternalConfiguration(boolean online, ClusterManager clusterManager, SpiBackgroundExecutor backgroundExecutor, ServerConfig serverConfig, BootupClasses bootupClasses) { + this.online = online; this.serverConfig = serverConfig; this.logManager = initLogManager(); this.docStoreFactory = initDocStoreFactory(serverConfig.service(DocStoreFactory.class)); this.jsonFactory = serverConfig.getJsonFactory(); this.clusterManager = clusterManager; this.backgroundExecutor = backgroundExecutor; - this.cacheManager = cacheManager; this.bootupClasses = bootupClasses; this.databasePlatform = serverConfig.getDatabasePlatform(); @@ -164,6 +179,9 @@ public class InternalConfiguration { this.deployCreateProperties = new DeployCreateProperties(typeManager); this.deployUtil = new DeployUtil(typeManager, serverConfig); + this.serverCachePlugin = initServerCachePlugin(); + this.cacheManager = initCacheManager(); + InternalConfigXmlRead xmlRead = new InternalConfigXmlRead(serverConfig); this.dtoBeanManager = new DtoBeanManager(typeManager, xmlRead.readDtoMapping()); @@ -400,7 +418,7 @@ public class InternalConfiguration { TransactionManagerOptions options = new TransactionManagerOptions(notifyL2CacheInForeground, serverConfig, scopeManager, clusterManager, backgroundExecutor, - indexUpdateProcessor, beanDescriptorManager, dataSource(), profileHandler(), logManager); + indexUpdateProcessor, beanDescriptorManager, dataSource(), profileHandler(), logManager, tableModState); if (serverConfig.isExplicitTransactionBeginMode()) { return new ExplicitTransactionManager(options); @@ -493,7 +511,7 @@ public class InternalConfiguration { return dataTimeZone; } - public ServerCacheManager cache() { + public ServerCacheManager cacheManager() { return new DefaultCacheAdapter(cacheManager); } @@ -537,4 +555,54 @@ public class InternalConfiguration { public SpiLogManager getLogManager() { return logManager; } + + private ServerCachePlugin initServerCachePlugin() { + + ServerCachePlugin plugin = serverConfig.getServerCachePlugin(); + if (plugin == null) { + ServiceLoader cacheFactories = ServiceLoader.load(ServerCachePlugin.class); + Iterator iterator = cacheFactories.iterator(); + if (iterator.hasNext()) { + // use the cacheFactory (via classpath service loader) + plugin = iterator.next(); + logger.debug("using ServerCacheFactory {}", serverCachePlugin.getClass()); + } else { + // use the built in default l2 caching which is local cache based + localL2Caching = true; + plugin = new DefaultServerCachePlugin(); + } + } + return plugin; + } + + /** + * Create and return the CacheManager. + */ + private SpiCacheManager initCacheManager() { + + if (!online || serverConfig.isDisableL2Cache()) { + // use local only L2 cache implementation as placeholder + return new DefaultServerCacheManager(); + } + + ServerCacheFactory factory = serverCachePlugin.create(serverConfig, backgroundExecutor); + + // reasonable default settings are for a cache per bean type + ServerCacheOptions beanOptions = new ServerCacheOptions(); + beanOptions.setMaxSize(serverConfig.getCacheMaxSize()); + beanOptions.setMaxIdleSecs(serverConfig.getCacheMaxIdleTime()); + beanOptions.setMaxSecsToLive(serverConfig.getCacheMaxTimeToLive()); + + // reasonable default settings for the query cache per bean type + ServerCacheOptions queryOptions = new ServerCacheOptions(); + queryOptions.setMaxSize(serverConfig.getQueryCacheMaxSize()); + queryOptions.setMaxIdleSecs(serverConfig.getQueryCacheMaxIdleTime()); + queryOptions.setMaxSecsToLive(serverConfig.getQueryCacheMaxTimeToLive()); + + CacheManagerOptions builder = new CacheManagerOptions(clusterManager, serverConfig, localL2Caching) + .with(beanOptions, queryOptions) + .with(factory, tableModState); + + return new DefaultServerCacheManager(builder); + } } diff --git a/src/main/java/io/ebeaninternal/server/core/OrmQueryRequest.java b/src/main/java/io/ebeaninternal/server/core/OrmQueryRequest.java index 799c5b9d2..264af7bc8 100644 --- a/src/main/java/io/ebeaninternal/server/core/OrmQueryRequest.java +++ b/src/main/java/io/ebeaninternal/server/core/OrmQueryRequest.java @@ -8,6 +8,7 @@ import io.ebean.Version; import io.ebean.bean.BeanCollection; import io.ebean.bean.EntityBean; import io.ebean.bean.PersistenceContext; +import io.ebean.cache.QueryCacheEntry; import io.ebean.common.BeanList; import io.ebean.common.CopyOnFirstWriteList; import io.ebean.event.BeanFindController; @@ -83,6 +84,8 @@ public final class OrmQueryRequest extends BeanRequest implements SpiOrmQuery private boolean inlineCountDistinct; + private Set dependentTables; + /** * Create the InternalQueryRequest. */ @@ -207,7 +210,7 @@ public final class OrmQueryRequest extends BeanRequest implements SpiOrmQuery /** * Prepare the query and calculate the query plan key. */ - public void prepareQuery() { + void prepareQuery() { beanDescriptor.prepareQuery(query); adapterPreQuery(); this.secondaryQueries = query.convertJoins(); @@ -680,8 +683,10 @@ public final class OrmQueryRequest extends BeanRequest implements SpiOrmQuery } } - public void putToQueryCache(Object queryResult) { - beanDescriptor.queryCachePut(cacheKey, queryResult); + public void putToQueryCache(Object result) { + // use transaction start where as query statement start would be better at READ_COMMITTED + long asOfTimestamp = transaction.getStartMillis(); + beanDescriptor.queryCachePut(cacheKey, new QueryCacheEntry(result, dependentTables, asOfTimestamp)); } /** @@ -752,4 +757,13 @@ public final class OrmQueryRequest extends BeanRequest implements SpiOrmQuery public boolean isInlineCountDistinct() { return inlineCountDistinct; } + + public void addDependentTables(Set tables) { + if (tables != null && !tables.isEmpty()) { + if (dependentTables == null) { + dependentTables = new LinkedHashSet<>(); + } + dependentTables.addAll(tables); + } + } } diff --git a/src/main/java/io/ebeaninternal/server/deploy/BeanDescriptor.java b/src/main/java/io/ebeaninternal/server/deploy/BeanDescriptor.java index ba0ecc52d..60b5ed482 100644 --- a/src/main/java/io/ebeaninternal/server/deploy/BeanDescriptor.java +++ b/src/main/java/io/ebeaninternal/server/deploy/BeanDescriptor.java @@ -11,6 +11,7 @@ import io.ebean.bean.BeanCollection; import io.ebean.bean.EntityBean; import io.ebean.bean.EntityBeanIntercept; import io.ebean.bean.PersistenceContext; +import io.ebean.cache.QueryCacheEntry; import io.ebean.config.EncryptKey; import io.ebean.config.ServerConfig; import io.ebean.config.dbplatform.IdType; @@ -1361,8 +1362,8 @@ public class BeanDescriptor implements BeanType, STreeType { /** * Put a query result into the query cache. */ - public void queryCachePut(Object id, Object queryResult) { - cacheHelp.queryCachePut(id, queryResult); + public void queryCachePut(Object id, QueryCacheEntry entry) { + cacheHelp.queryCachePut(id, entry); } /** diff --git a/src/main/java/io/ebeaninternal/server/deploy/BeanDescriptorCacheHelp.java b/src/main/java/io/ebeaninternal/server/deploy/BeanDescriptorCacheHelp.java index 1d11da087..07f326e6c 100644 --- a/src/main/java/io/ebeaninternal/server/deploy/BeanDescriptorCacheHelp.java +++ b/src/main/java/io/ebeaninternal/server/deploy/BeanDescriptorCacheHelp.java @@ -4,6 +4,7 @@ import io.ebean.bean.BeanCollection; import io.ebean.bean.EntityBean; import io.ebean.bean.EntityBeanIntercept; import io.ebean.bean.PersistenceContext; +import io.ebean.cache.QueryCacheEntry; import io.ebean.cache.ServerCache; import io.ebeaninternal.api.BeanCacheResult; import io.ebeaninternal.api.TransactionEventTable.TableIUD; @@ -54,6 +55,7 @@ final class BeanDescriptorCacheHelp { * Flag indicating this bean has no relationships. */ private final boolean cacheSharableBeans; + private final boolean invalidateQueryCache; private final Class beanType; @@ -84,6 +86,7 @@ final class BeanDescriptorCacheHelp { this.cacheName = beanType.getSimpleName(); this.cacheManager = cacheManager; this.cacheOptions = cacheOptions; + this.invalidateQueryCache = cacheOptions.isInvalidateQueryCache(); this.cacheSharableBeans = cacheSharableBeans; this.propertiesOneImported = propertiesOneImported; this.naturalKey = cacheOptions.getNaturalKey(); @@ -111,7 +114,7 @@ final class BeanDescriptorCacheHelp { * Derive the cache notify flags. */ void deriveNotifyFlags() { - cacheNotifyOnAll = (beanCache != null || queryCache != null); + cacheNotifyOnAll = (invalidateQueryCache || beanCache != null || queryCache != null); cacheNotifyOnDelete = !cacheNotifyOnAll && isNotifyOnDeletes(); if (logger.isDebugEnabled()) { @@ -225,14 +228,14 @@ final class BeanDescriptorCacheHelp { /** * Put a query result into the query cache. */ - void queryCachePut(Object id, Object queryResult) { + void queryCachePut(Object id, QueryCacheEntry entry) { if (queryCache == null) { throw new IllegalStateException("No query cache enabled on " + desc + ". Need explicit @Cache(enableQueryCache=true)"); } if (queryLog.isDebugEnabled()) { queryLog.debug(" PUT {}({})", cacheName, id); } - queryCache.put(id, queryResult); + queryCache.put(id, entry); } @@ -723,30 +726,42 @@ final class BeanDescriptorCacheHelp { * Add appropriate cache changes to support delete by id. */ void handleDelete(Object id, CacheChangeSet changeSet) { - if (beanCache != null) { - changeSet.addBeanRemove(desc, id); + if (invalidateQueryCache) { + changeSet.addInvalidate(desc); + } else { + if (beanCache != null) { + changeSet.addBeanRemove(desc, id); + } + cacheDeleteImported(true, null, changeSet); } - cacheDeleteImported(true, null, changeSet); } /** * Add appropriate cache changes to support delete bean. */ void handleDelete(Object id, PersistRequestBean deleteRequest, CacheChangeSet changeSet) { - queryCacheClear(changeSet); - if (beanCache != null) { - changeSet.addBeanRemove(desc, id); + if (invalidateQueryCache) { + changeSet.addInvalidate(desc); + } else { + queryCacheClear(changeSet); + if (beanCache != null) { + changeSet.addBeanRemove(desc, id); + } + cacheDeleteImported(true, deleteRequest.getEntityBean(), changeSet); } - cacheDeleteImported(true, deleteRequest.getEntityBean(), changeSet); } /** * Add appropriate cache changes to support insert. */ void handleInsert(PersistRequestBean insertRequest, CacheChangeSet changeSet) { - queryCacheClear(changeSet); - cacheDeleteImported(false, insertRequest.getEntityBean(), changeSet); - changeSet.addBeanInsert(desc.getBaseTable()); + if (invalidateQueryCache) { + changeSet.addInvalidate(desc); + } else { + queryCacheClear(changeSet); + cacheDeleteImported(false, insertRequest.getEntityBean(), changeSet); + changeSet.addBeanInsert(desc.getBaseTable()); + } } private void cacheDeleteImported(boolean clear, EntityBean entityBean, CacheChangeSet changeSet) { @@ -759,28 +774,30 @@ final class BeanDescriptorCacheHelp { * Add appropriate changes to support update. */ void handleUpdate(Object id, PersistRequestBean updateRequest, CacheChangeSet changeSet) { + if (invalidateQueryCache) { + changeSet.addInvalidate(desc); - queryCacheClear(changeSet); + } else { + queryCacheClear(changeSet); + if (beanCache == null) { + // query caching only + return; + } - if (beanCache == null) { - // query caching only - return; - } - - List> manyCollections = updateRequest.getUpdatedManyCollections(); - if (manyCollections != null) { - for (BeanPropertyAssocMany many : manyCollections) { - if (!many.isElementCollection()) { - Object details = many.getValue(updateRequest.getEntityBean()); - CachedManyIds entry = createManyIds(many, details); - if (entry != null) { - changeSet.addManyPut(desc, many.getName(), id, entry); + List> manyCollections = updateRequest.getUpdatedManyCollections(); + if (manyCollections != null) { + for (BeanPropertyAssocMany many : manyCollections) { + if (!many.isElementCollection()) { + Object details = many.getValue(updateRequest.getEntityBean()); + CachedManyIds entry = createManyIds(many, details); + if (entry != null) { + changeSet.addManyPut(desc, many.getName(), id, entry); + } } } } + updateRequest.addBeanUpdate(changeSet); } - - updateRequest.addBeanUpdate(changeSet); } /** diff --git a/src/main/java/io/ebeaninternal/server/deploy/meta/DeployBeanDescriptor.java b/src/main/java/io/ebeaninternal/server/deploy/meta/DeployBeanDescriptor.java index 199b7a90b..24eca35eb 100644 --- a/src/main/java/io/ebeaninternal/server/deploy/meta/DeployBeanDescriptor.java +++ b/src/main/java/io/ebeaninternal/server/deploy/meta/DeployBeanDescriptor.java @@ -440,6 +440,13 @@ public class DeployBeanDescriptor { this.inheritInfo = inheritInfo; } + /** + * Set that this type invalidates query caches. + */ + public void setInvalidateQueryCache() { + this.cacheOptions = CacheOptions.INVALIDATE_QUERY_CACHE; + } + /** * Enable L2 bean and query caching based on Cache annotation. */ diff --git a/src/main/java/io/ebeaninternal/server/deploy/parse/AnnotationClass.java b/src/main/java/io/ebeaninternal/server/deploy/parse/AnnotationClass.java index 208b0bdf3..f28f5e303 100644 --- a/src/main/java/io/ebeaninternal/server/deploy/parse/AnnotationClass.java +++ b/src/main/java/io/ebeaninternal/server/deploy/parse/AnnotationClass.java @@ -7,6 +7,7 @@ import io.ebean.annotation.Draftable; import io.ebean.annotation.DraftableElement; import io.ebean.annotation.History; import io.ebean.annotation.Index; +import io.ebean.annotation.InvalidateQueryCache; import io.ebean.annotation.ReadAudit; import io.ebean.annotation.UpdateMode; import io.ebean.annotation.View; @@ -181,9 +182,16 @@ public class AnnotationClass extends AnnotationParser { descriptor.setUpdateChangesOnly(updateMode.updateChangesOnly()); } - Cache cache = AnnotationUtil.findAnnotationRecursive(cls, Cache.class); - if (cache != null && !disableL2Cache) { - descriptor.setCache(cache); + if (!disableL2Cache) { + Cache cache = AnnotationUtil.findAnnotationRecursive(cls, Cache.class); + if (cache != null) { + descriptor.setCache(cache); + } else { + InvalidateQueryCache invalidateQueryCache = AnnotationUtil.findAnnotationRecursive(cls, InvalidateQueryCache.class); + if (invalidateQueryCache != null) { + descriptor.setInvalidateQueryCache(); + } + } } Set namedQueries = AnnotationUtil.findAnnotationsRecursive(cls, NamedQuery.class); diff --git a/src/main/java/io/ebeaninternal/server/query/CQuery.java b/src/main/java/io/ebeaninternal/server/query/CQuery.java index 9a40afaaf..cdc0a7138 100644 --- a/src/main/java/io/ebeaninternal/server/query/CQuery.java +++ b/src/main/java/io/ebeaninternal/server/query/CQuery.java @@ -38,6 +38,7 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.NoSuchElementException; +import java.util.Set; /** * An object that represents a SqlSelect statement. @@ -811,4 +812,8 @@ public class CQuery implements DbReadContext, CancelableQuery, SpiProfileTran PreparedStatement getPstmt() { return pstmt; } + + public Set getDependentTables() { + return queryPlan.getDependentTables(); + } } diff --git a/src/main/java/io/ebeaninternal/server/query/CQueryEngine.java b/src/main/java/io/ebeaninternal/server/query/CQueryEngine.java index a38a9e61a..05e64658a 100644 --- a/src/main/java/io/ebeaninternal/server/query/CQueryEngine.java +++ b/src/main/java/io/ebeaninternal/server/query/CQueryEngine.java @@ -25,6 +25,8 @@ import org.slf4j.LoggerFactory; import javax.persistence.PersistenceException; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.ArrayList; +import java.util.Collections; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -111,6 +113,15 @@ public class CQueryEngine { if (request.isLogSummary()) { request.getTransaction().logSummary(rcQuery.getSummary()); } + if (request.isQueryCachePut() && !list.isEmpty()) { + request.addDependentTables(rcQuery.getDependentTables()); + + list = Collections.unmodifiableList(list); + request.putToQueryCache(list); + if (Boolean.FALSE.equals(request.getQuery().isReadOnly())) { + list = new ArrayList<>(list); + } + } return list; } catch (SQLException e) { @@ -173,6 +184,11 @@ public class CQueryEngine { request.getTransaction().end(); } + if (request.isQueryCachePut()) { + request.addDependentTables(rcQuery.getDependentTables()); + request.putToQueryCache(count); + } + return count; } catch (SQLException e) { @@ -385,6 +401,9 @@ public class CQueryEngine { } request.executeSecondaryQueries(false); + if (request.isQueryCachePut()) { + request.addDependentTables(cquery.getDependentTables()); + } return beanCollection; diff --git a/src/main/java/io/ebeaninternal/server/query/CQueryFetchSingleAttribute.java b/src/main/java/io/ebeaninternal/server/query/CQueryFetchSingleAttribute.java index b5d2d05b1..3b8b8ef01 100644 --- a/src/main/java/io/ebeaninternal/server/query/CQueryFetchSingleAttribute.java +++ b/src/main/java/io/ebeaninternal/server/query/CQueryFetchSingleAttribute.java @@ -17,6 +17,7 @@ import java.sql.PreparedStatement; import java.sql.SQLException; import java.util.ArrayList; import java.util.List; +import java.util.Set; /** * Base compiled query request for single attribute queries. @@ -188,4 +189,8 @@ class CQueryFetchSingleAttribute implements SpiProfileTransactionEvent { .profileStream() .addQueryEvent(query.profileEventId(), profileOffset, desc.getProfileId(), rowCount, query.getProfileId()); } + + Set getDependentTables() { + return queryPlan.getDependentTables(); + } } diff --git a/src/main/java/io/ebeaninternal/server/query/CQueryPlan.java b/src/main/java/io/ebeaninternal/server/query/CQueryPlan.java index a0d9a46b1..7b4b27e18 100644 --- a/src/main/java/io/ebeaninternal/server/query/CQueryPlan.java +++ b/src/main/java/io/ebeaninternal/server/query/CQueryPlan.java @@ -24,6 +24,8 @@ import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.Collections; +import java.util.Set; /** * Represents a query for a given SQL statement. @@ -88,6 +90,8 @@ public class CQueryPlan { */ private volatile String auditQueryHash; + private final Set dependentTables; + /** * Create a query plan based on a OrmQueryRequest. */ @@ -110,6 +114,7 @@ public class CQueryPlan { this.logWhereSql = logWhereSql; this.encryptedProps = sqlTree.getEncryptedProps(); this.stats = new CQueryPlanStats(this, server.isCollectQueryOrigins()); + this.dependentTables = sqlTree.dependentTables(); } /** @@ -134,6 +139,7 @@ public class CQueryPlan { this.logWhereSql = logWhereSql; this.encryptedProps = sqlTree.getEncryptedProps(); this.stats = new CQueryPlanStats(this, server.isCollectQueryOrigins()); + this.dependentTables = (rawSql) ? Collections.emptySet() : sqlTree.dependentTables(); } private String location() { @@ -154,6 +160,10 @@ public class CQueryPlan { return beanType; } + public Set getDependentTables() { + return dependentTables; + } + public ProfileLocation getProfileLocation() { return profileLocation; } diff --git a/src/main/java/io/ebeaninternal/server/query/CQueryRowCount.java b/src/main/java/io/ebeaninternal/server/query/CQueryRowCount.java index 5ddfe7b02..e0f9c1dbb 100644 --- a/src/main/java/io/ebeaninternal/server/query/CQueryRowCount.java +++ b/src/main/java/io/ebeaninternal/server/query/CQueryRowCount.java @@ -12,6 +12,7 @@ import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; +import java.util.Set; /** * Executes the select row count query. @@ -154,4 +155,8 @@ class CQueryRowCount implements SpiProfileTransactionEvent { .profileStream() .addQueryEvent(query.profileEventId(), profileOffset, desc.getProfileId(), rowCount, query.getProfileId()); } + + Set getDependentTables() { + return queryPlan.getDependentTables(); + } } diff --git a/src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java b/src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java index c6c042c92..d72a2f688 100644 --- a/src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java +++ b/src/main/java/io/ebeaninternal/server/query/DefaultOrmQueryEngine.java @@ -88,41 +88,20 @@ public class DefaultOrmQueryEngine implements OrmQueryEngine { public int findCount(OrmQueryRequest request) { flushJdbcBatchOnQuery(request); - int result = queryEngine.findCount(request); - if (request.getQuery().getUseQueryCache().isPut()) { - request.putToQueryCache(result); - } - return result; + return queryEngine.findCount(request); } @Override public List findIds(OrmQueryRequest request) { flushJdbcBatchOnQuery(request); - List result = queryEngine.findIds(request); - if (request.getQuery().getUseQueryCache().isPut()) { - result = Collections.unmodifiableList(result); - request.putToQueryCache(result); - if (Boolean.FALSE.equals(request.getQuery().isReadOnly())) { - result = new ArrayList<>(result); - } - } - return result; + return queryEngine.findIds(request); } @Override public List findSingleAttributeList(OrmQueryRequest request) { flushJdbcBatchOnQuery(request); - List result = queryEngine.findSingleAttributeList(request); - if (!result.isEmpty() && request.getQuery().getUseQueryCache().isPut()) { - // load the query result into the query cache - result = Collections.unmodifiableList(result); - request.putToQueryCache(result); - if (Boolean.FALSE.equals(request.getQuery().isReadOnly())) { - result = new ArrayList<>(result); - } - } - return result; + return queryEngine.findSingleAttributeList(request); } @Override diff --git a/src/main/java/io/ebeaninternal/server/query/SqlTree.java b/src/main/java/io/ebeaninternal/server/query/SqlTree.java index b6ee3c980..91e71fb0d 100644 --- a/src/main/java/io/ebeaninternal/server/query/SqlTree.java +++ b/src/main/java/io/ebeaninternal/server/query/SqlTree.java @@ -3,6 +3,7 @@ package io.ebeaninternal.server.query; import io.ebeaninternal.api.SpiQuery; import java.util.ArrayList; +import java.util.LinkedHashSet; import java.util.List; import java.util.Set; @@ -163,4 +164,13 @@ class SqlTree { boolean isSingleProperty() { return rootNode.isSingleProperty(); } + + /** + * Return the tables that are joined in this query. + */ + Set dependentTables() { + Set tables = new LinkedHashSet<>(); + rootNode.dependentTables(tables); + return tables; + } } diff --git a/src/main/java/io/ebeaninternal/server/query/SqlTreeNode.java b/src/main/java/io/ebeaninternal/server/query/SqlTreeNode.java index c503581d8..39e0bff28 100644 --- a/src/main/java/io/ebeaninternal/server/query/SqlTreeNode.java +++ b/src/main/java/io/ebeaninternal/server/query/SqlTreeNode.java @@ -9,6 +9,7 @@ import io.ebeaninternal.server.type.ScalarType; import java.sql.SQLException; import java.util.List; +import java.util.Set; interface SqlTreeNode { @@ -87,4 +88,9 @@ interface SqlTreeNode { * Return true if the query is known to only have a single property selected. */ boolean isSingleProperty(); + + /** + * Add dependent tables to the given set. + */ + void dependentTables(Set tables); } diff --git a/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeBean.java b/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeBean.java index 339b08a95..9bafd1317 100644 --- a/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeBean.java +++ b/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeBean.java @@ -21,6 +21,7 @@ import java.sql.Timestamp; import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Set; /** * Normal bean included in the query. @@ -578,6 +579,14 @@ class SqlTreeNodeBean implements SqlTreeNode { } } + @Override + public void dependentTables(Set tables) { + tables.add(nodeBeanProp.target().getBaseTable(temporalMode)); + for (SqlTreeNode child : children) { + child.dependentTables(tables); + } + } + /** * Join to base table for this node. This includes a join to the intersection * table if this is a ManyToMany node. diff --git a/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeExtraJoin.java b/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeExtraJoin.java index b517965e4..6dfd1fd7b 100644 --- a/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeExtraJoin.java +++ b/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeExtraJoin.java @@ -11,6 +11,7 @@ import io.ebeaninternal.server.type.ScalarType; import java.util.ArrayList; import java.util.List; +import java.util.Set; /** * The purpose is to add an extra join to the query. @@ -100,6 +101,16 @@ class SqlTreeNodeExtraJoin implements SqlTreeNode { children.add(child); } + @Override + public void dependentTables(Set tables) { + tables.add(assocBeanProperty.target().getBaseTable(SpiQuery.TemporalMode.CURRENT)); + if (children != null) { + for (SqlTreeNode child : children) { + child.dependentTables(tables); + } + } + } + @Override public void appendFrom(DbSqlContext ctx, SqlJoinType joinType) { diff --git a/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeManyRoot.java b/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeManyRoot.java index edeff9fce..ec8f6aa47 100644 --- a/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeManyRoot.java +++ b/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeManyRoot.java @@ -7,6 +7,7 @@ import io.ebeaninternal.server.deploy.DbSqlContext; import java.sql.SQLException; import java.util.List; +import java.util.Set; final class SqlTreeNodeManyRoot extends SqlTreeNodeBean { diff --git a/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeManyWhereJoin.java b/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeManyWhereJoin.java index c31ea619c..a5cd0e232 100644 --- a/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeManyWhereJoin.java +++ b/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeManyWhereJoin.java @@ -10,6 +10,7 @@ import io.ebeaninternal.server.deploy.TableJoin; import io.ebeaninternal.server.type.ScalarType; import java.util.List; +import java.util.Set; /** * Join to Many (or child of a many) to support where clause predicates on many properties. @@ -109,6 +110,11 @@ class SqlTreeNodeManyWhereJoin implements SqlTreeNode { } } + @Override + public void dependentTables(Set tables) { + tables.add(nodeBeanProp.target().getBaseTable(SpiQuery.TemporalMode.CURRENT)); + } + @Override public void buildRawSqlSelectChain(List selectChain) { // nothing to add diff --git a/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeRoot.java b/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeRoot.java index 74be7c415..2a69a0c16 100644 --- a/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeRoot.java +++ b/src/main/java/io/ebeaninternal/server/query/SqlTreeNodeRoot.java @@ -5,6 +5,7 @@ import io.ebeaninternal.server.deploy.DbSqlContext; import io.ebeaninternal.server.deploy.TableJoin; import java.util.List; +import java.util.Set; /** * Represents the root node of the Sql Tree. @@ -77,4 +78,10 @@ final class SqlTreeNodeRoot extends SqlTreeNodeBean { return joinType; } + @Override + public void dependentTables(Set tables) { + for (SqlTreeNode child : children) { + child.dependentTables(tables); + } + } } diff --git a/src/main/java/io/ebeaninternal/server/transaction/ImplicitReadOnlyTransaction.java b/src/main/java/io/ebeaninternal/server/transaction/ImplicitReadOnlyTransaction.java index 066a7504e..1ccc44dbd 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/ImplicitReadOnlyTransaction.java +++ b/src/main/java/io/ebeaninternal/server/transaction/ImplicitReadOnlyTransaction.java @@ -67,6 +67,7 @@ class ImplicitReadOnlyTransaction implements SpiTransaction, TxnProfileEventCode private Map userObjects; private long startNanos; + private long startMillis; /** * Create without a tenantId. @@ -79,6 +80,7 @@ class ImplicitReadOnlyTransaction implements SpiTransaction, TxnProfileEventCode this.connection = connection; this.persistenceContext = new DefaultPersistenceContext(); this.startNanos = System.nanoTime(); + this.startMillis = manager.clockNowEpoch(); } /** @@ -89,6 +91,12 @@ class ImplicitReadOnlyTransaction implements SpiTransaction, TxnProfileEventCode this.tenantId = tenantId; } + @Override + public long getStartMillis() { + // not used on read only transaction + return startMillis; + } + @Override public void setLabel(String label) { // do nothing diff --git a/src/main/java/io/ebeaninternal/server/transaction/JdbcTransaction.java b/src/main/java/io/ebeaninternal/server/transaction/JdbcTransaction.java index 4f6ec5bd1..da8e53781 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/JdbcTransaction.java +++ b/src/main/java/io/ebeaninternal/server/transaction/JdbcTransaction.java @@ -187,6 +187,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes { protected ProfileLocation profileLocation; protected final long startNanos; + private final long startMillis; /** * Create a new JdbcTransaction. @@ -203,6 +204,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes { this.startNanos = System.nanoTime(); if (manager == null) { + this.startMillis = System.currentTimeMillis(); this.logSql = false; this.logSummary = false; this.skipCacheAfterWrite = true; @@ -210,6 +212,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes { this.batchOnCascadeMode = PersistBatch.NONE; this.onQueryOnly = OnQueryOnly.ROLLBACK; } else { + this.startMillis = manager.clockNowEpoch(); this.logSql = manager.isLogSql(); this.logSummary = manager.isLogSummary(); this.skipCacheAfterWrite = manager.isSkipCacheAfterWrite(); @@ -235,6 +238,11 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes { return label; } + @Override + public long getStartMillis() { + return startMillis; + } + @Override public long profileOffset() { return (profileStream == null) ? 0 : profileStream.offset(); @@ -853,7 +861,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes { public TransactionEvent getEvent() { queryOnly = false; if (event == null) { - event = new TransactionEvent(); + event = new TransactionEvent(startMillis); } return event; } @@ -1050,7 +1058,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes { // the event has been sent to the transaction manager // for postCommit processing (l2 cache updates etc) // start a new transaction event - event = new TransactionEvent(); + event = new TransactionEvent(startMillis); } catch (Exception e) { doRollback(e); diff --git a/src/main/java/io/ebeaninternal/server/transaction/NoTransaction.java b/src/main/java/io/ebeaninternal/server/transaction/NoTransaction.java index 62ace0ede..bd2b20c35 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/NoTransaction.java +++ b/src/main/java/io/ebeaninternal/server/transaction/NoTransaction.java @@ -47,6 +47,12 @@ class NoTransaction implements SpiTransaction { } + @Override + public long getStartMillis() { + // not used + return System.currentTimeMillis(); + } + @Override public boolean isActive() { // always false diff --git a/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java b/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java index ac3a57560..b715d2ee3 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java +++ b/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java @@ -14,6 +14,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.util.List; +import java.util.Set; /** * Performs post commit processing using a background thread. @@ -88,10 +89,10 @@ public final class PostCommitProcessing { processTableEvents(event.getEventTables()); if (manager.notifyL2CacheInForeground) { // process l2 cache changes in foreground - processCacheChanges(event.buildCacheChanges(manager.viewInvalidation)); + processCacheChanges(event.buildCacheChanges()); } else { // collect l2 cache changes for delayed background processing - cacheChanges = event.buildCacheChanges(manager.viewInvalidation); + cacheChanges = event.buildCacheChanges(); } } @@ -165,7 +166,12 @@ public final class PostCommitProcessing { */ private void processCacheChanges(CacheChangeSet cacheChanges) { if (cacheChanges != null) { - manager.processViewInvalidation(cacheChanges.apply()); + Set touched = cacheChanges.touchedTables(); + if (touched != null && !touched.isEmpty()) { + manager.processTouchedTables(touched, cacheChanges.modificationTimestamp()); + // TODO: Propagate touched tables to other cluster members + } + cacheChanges.apply(); } } diff --git a/src/main/java/io/ebeaninternal/server/transaction/TableModState.java b/src/main/java/io/ebeaninternal/server/transaction/TableModState.java new file mode 100644 index 000000000..d2aa4143a --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/transaction/TableModState.java @@ -0,0 +1,51 @@ +package io.ebeaninternal.server.transaction; + +import io.ebean.cache.QueryCacheEntry; +import io.ebean.cache.QueryCacheEntryValidate; + +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +/** + * Holds timestamp of last modification per table. + *

+ * This information is used to validate entries in the L2 query caches. + *

+ */ +public class TableModState implements QueryCacheEntryValidate { + + private Map tableModStamp = new ConcurrentHashMap<>(); + + /** + * Set the modified timestamp on the tables that have been touched. + */ + public void touch(Set touchedTables, long modTimestamp) { + + for (String tableName : touchedTables) { + tableModStamp.put(tableName, modTimestamp); + } + } + + /** + * Return true if all the tables are valid based on timestamp comparison. + */ + public boolean isValid(Set tables, long sinceTimestamp) { + for (String tableName : tables) { + Long modTime = tableModStamp.get(tableName); + if (modTime != null && modTime > sinceTimestamp ) { + return false; + } + } + return true; + } + + @Override + public boolean isValid(QueryCacheEntry entry) { + Set dependentTables = entry.getDependentTables(); + if (dependentTables != null && !dependentTables.isEmpty()) { + return isValid(dependentTables, entry.getTimestamp()); + } + return true; + } +} diff --git a/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java b/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java index 0e9c91d62..0ff2894d6 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java +++ b/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java @@ -46,7 +46,7 @@ import java.util.Set; /** * Manages transactions. *

- * Keeps the Cache and Cluster in synch when transactions are committed. + * Keeps the Cache and Cluster in sync when transactions are committed. *

*/ public class TransactionManager implements SpiTransactionManager { @@ -136,6 +136,8 @@ public class TransactionManager implements SpiTransactionManager { private final TimedMetricMap txnNamed; private final TransactionScopeManager scopeManager; + private final TableModState tableModState; + /** * Create the TransactionManager */ @@ -157,6 +159,7 @@ public class TransactionManager implements SpiTransactionManager { this.clusterManager = options.clusterManager; this.serverName = options.config.getName(); this.scopeManager = options.scopeManager; + this.tableModState = options.tableModState; this.backgroundExecutor = options.backgroundExecutor; this.dataSourceSupplier = options.dataSourceSupplier; this.docStoreActive = options.config.getDocStoreConfig().isActive(); @@ -178,6 +181,13 @@ public class TransactionManager implements SpiTransactionManager { scopeManager.register(this); } + /** + * Return the NOW timestamp in epoch millis. + */ + public long clockNowEpoch() { + return System.currentTimeMillis(); // TODO: Review when we supply a Clock via ServerConfig. + } + /** * Create a new scoped transaction. */ @@ -434,7 +444,7 @@ public class TransactionManager implements SpiTransactionManager { private void externalModificationEvent(TransactionEventTable tableEvents) { - TransactionEvent event = new TransactionEvent(); + TransactionEvent event = new TransactionEvent(clockNowEpoch()); event.add(tableEvents); PostCommitProcessing postCommit = new PostCommitProcessing(clusterManager, this, event); @@ -495,9 +505,10 @@ public class TransactionManager implements SpiTransactionManager { /** * Invalidate the query caches for entities based on views. */ - public void processViewInvalidation(Set viewInvalidation) { - if (!viewInvalidation.isEmpty()) { - beanDescriptorManager.processViewInvalidation(viewInvalidation); + public void processTouchedTables(Set touchedTables, long modTimestamp) { + tableModState.touch(touchedTables, modTimestamp); + if (viewInvalidation) { + beanDescriptorManager.processViewInvalidation(touchedTables); } } diff --git a/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java b/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java index fe8b8d217..16f5dafe9 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java +++ b/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java @@ -24,10 +24,13 @@ public class TransactionManagerOptions { final SpiProfileHandler profileHandler; final TransactionScopeManager scopeManager; final SpiLogManager logManager; + final TableModState tableModState; + public TransactionManagerOptions(boolean notifyL2CacheInForeground, ServerConfig config, TransactionScopeManager scopeManager, ClusterManager clusterManager, BackgroundExecutor backgroundExecutor, DocStoreUpdateProcessor docStoreUpdateProcessor, - BeanDescriptorManager descMgr, DataSourceSupplier dataSourceSupplier, SpiProfileHandler profileHandler, SpiLogManager logManager) { + BeanDescriptorManager descMgr, DataSourceSupplier dataSourceSupplier, SpiProfileHandler profileHandler, + SpiLogManager logManager, TableModState tableModState) { this.notifyL2CacheInForeground = notifyL2CacheInForeground; this.config = config; @@ -39,6 +42,7 @@ public class TransactionManagerOptions { this.dataSourceSupplier = dataSourceSupplier; this.profileHandler = profileHandler; this.logManager = logManager; + this.tableModState = tableModState; } } diff --git a/src/test/java/io/ebeaninternal/server/cache/DefaultCacheHolderTest.java b/src/test/java/io/ebeaninternal/server/cache/DefaultCacheHolderTest.java index 3094c9ad5..046ae15d6 100644 --- a/src/test/java/io/ebeaninternal/server/cache/DefaultCacheHolderTest.java +++ b/src/test/java/io/ebeaninternal/server/cache/DefaultCacheHolderTest.java @@ -3,6 +3,8 @@ package io.ebeaninternal.server.cache; import io.ebean.cache.ServerCacheFactory; import io.ebean.cache.ServerCacheOptions; import io.ebean.cache.ServerCacheType; +import io.ebean.config.ServerConfig; +import io.ebeaninternal.server.transaction.TableModState; import org.tests.model.basic.Contact; import org.tests.model.basic.Customer; import org.junit.Test; @@ -17,10 +19,17 @@ public class DefaultCacheHolderTest { private final ServerCacheFactory cacheFactory = new DefaultServerCacheFactory(); private final ServerCacheOptions defaultOptions = new ServerCacheOptions(); - @Test - public void getCache_normal() throws Exception { + private CacheManagerOptions options() { + return new CacheManagerOptions(null, new ServerConfig(), true) + .with(defaultOptions, defaultOptions) + .with(cacheFactory, new TableModState()); + } - DefaultCacheHolder holder = new DefaultCacheHolder(cacheFactory, defaultOptions, defaultOptions, null); + + @Test + public void getCache_normal() { + + DefaultCacheHolder holder = new DefaultCacheHolder(options()); DefaultServerCache cache = cache(holder, Customer.class, "customer"); assertThat(cache.getName()).isEqualTo("customer_B"); @@ -40,7 +49,9 @@ public class DefaultCacheHolderTest { @Test public void getCache_multiTenant() throws Exception { - DefaultCacheHolder holder = new DefaultCacheHolder(cacheFactory, defaultOptions, defaultOptions, tenantId::get); + CacheManagerOptions builder = options().with(tenantId::get); + + DefaultCacheHolder holder = new DefaultCacheHolder(builder); tenantId.set("ten_1"); DefaultServerCache cache = cache(holder, Customer.class, "customer"); @@ -48,26 +59,26 @@ public class DefaultCacheHolderTest { cache.put("1", "value-for-tenant1"); cache.put("2", "an other value-for-tenant1"); - + assertThat(cache.size()).isEqualTo(2); - + tenantId.set("ten_2"); cache.put("1", "value-for-tenant2"); cache.put("2", "an other value-for-tenant2"); - + assertThat(cache.size()).isEqualTo(4); - + assertThat(cache.get("1")).isEqualTo("value-for-tenant2"); assertThat(cache.get("2")).isEqualTo("an other value-for-tenant2"); - - + + tenantId.set("ten_1"); - + assertThat(cache.get("1")).isEqualTo("value-for-tenant1"); assertThat(cache.get("2")).isEqualTo("an other value-for-tenant1"); - - Exception exInThread[] = new Exception[1]; + + Exception exInThread[] = new Exception[1]; Thread t = new Thread() { @Override public void run() { @@ -77,29 +88,29 @@ public class DefaultCacheHolderTest { cache.put("1", "value-for-tenant2"); cache.put("2", "an other value-for-tenant2"); - + tenantId.set(null); - + cache.clear(); } catch (Exception e) { exInThread[0] = e; } }; }; - + // do some async work t.start(); t.join(); - if (exInThread[0] != null) { + if (exInThread[0] != null) { throw exInThread[0]; } assertThat(cache.size()).isEqualTo(0); } - @Test - public void clearAll() throws Exception { - DefaultCacheHolder holder = new DefaultCacheHolder(cacheFactory, defaultOptions, defaultOptions, null); + public void clearAll() { + + DefaultCacheHolder holder = new DefaultCacheHolder(options()); DefaultServerCache cache = cache(holder, Customer.class, "customer"); cache.put("foo", "foo"); assertThat(cache.size()).isEqualTo(1); @@ -109,8 +120,11 @@ public class DefaultCacheHolderTest { } @Test - public void clearAll_multiTenant() throws Exception { - DefaultCacheHolder holder = new DefaultCacheHolder(cacheFactory, defaultOptions, defaultOptions, tenantId::get); + public void clearAll_multiTenant() { + + CacheManagerOptions options = options().with(tenantId::get); + + DefaultCacheHolder holder = new DefaultCacheHolder(options); DefaultServerCache cache = cache(holder, Customer.class, "customer"); cache.put("foo", "foo"); assertThat(cache.size()).isEqualTo(1); @@ -118,5 +132,5 @@ public class DefaultCacheHolderTest { holder.clearAll(); assertThat(cache.size()).isEqualTo(0); } - + } diff --git a/src/test/java/io/ebeaninternal/server/cache/DefaultCacheHolder_getCacheOptions_Test.java b/src/test/java/io/ebeaninternal/server/cache/DefaultCacheHolder_getCacheOptions_Test.java index beefe1f47..6cbc82c8f 100644 --- a/src/test/java/io/ebeaninternal/server/cache/DefaultCacheHolder_getCacheOptions_Test.java +++ b/src/test/java/io/ebeaninternal/server/cache/DefaultCacheHolder_getCacheOptions_Test.java @@ -2,6 +2,7 @@ package io.ebeaninternal.server.cache; import io.ebean.cache.ServerCacheOptions; import io.ebean.cache.ServerCacheType; +import io.ebean.config.ServerConfig; import org.tests.model.basic.Article; import org.tests.model.basic.Order; import org.tests.model.basic.Product; @@ -19,7 +20,10 @@ public class DefaultCacheHolder_getCacheOptions_Test { defaultOptions.setMaxSize(10000); defaultOptions.setMaxSecsToLive(120); - this.cacheHolder = new DefaultCacheHolder(null, defaultOptions, defaultOptions, null); + CacheManagerOptions builder = new CacheManagerOptions(null, new ServerConfig(), true) + .with(defaultOptions, defaultOptions); + + this.cacheHolder = new DefaultCacheHolder(builder); } @Test diff --git a/src/test/java/io/ebeaninternal/server/cache/DefaultServerCacheConfigTest.java b/src/test/java/io/ebeaninternal/server/cache/DefaultServerCacheConfigTest.java new file mode 100644 index 000000000..bb03cce18 --- /dev/null +++ b/src/test/java/io/ebeaninternal/server/cache/DefaultServerCacheConfigTest.java @@ -0,0 +1,52 @@ +package io.ebeaninternal.server.cache; + +import io.ebean.cache.ServerCacheConfig; +import io.ebean.cache.ServerCacheOptions; +import org.junit.Test; + +import static org.junit.Assert.*; + +public class DefaultServerCacheConfigTest { + + + private DefaultServerCacheConfig create(int maxSize, int maxIdleSecs, int maxSecsToLive, int trimFreq) { + ServerCacheOptions options = new ServerCacheOptions(); + options.setMaxSize(maxSize); + options.setMaxIdleSecs(maxIdleSecs); + options.setMaxSecsToLive(maxSecsToLive); + options.setTrimFrequency(trimFreq); + + return new DefaultServerCacheConfig(new ServerCacheConfig(null, null, options, null, null)); + } + + @Test + public void trimFreq_halfIdle() { + + assertEquals(create(10000,10,20, 0).determineTrimFrequency(), 4); + } + + + @Test + public void trimFreq_halfIdle_withRounding() { + + assertEquals(create(10000,11,20, 0).determineTrimFrequency(), 4); + } + + @Test + public void trimFreq_halfTTL() { + + assertEquals(create(10000,0,20, 0).determineTrimFrequency(), 9); + } + + @Test + public void trimFreq_halfTTL_withRounding() { + + assertEquals(create(10000,0,21, 0).determineTrimFrequency(), 9); + } + + @Test + public void trimFreq_explicit() { + + assertEquals(create(10000,10,20, 42).determineTrimFrequency(), 42); + } +} diff --git a/src/test/java/io/ebeaninternal/server/cache/DefaultServerCacheTest.java b/src/test/java/io/ebeaninternal/server/cache/DefaultServerCacheTest.java index 00ff68631..ead5158ff 100644 --- a/src/test/java/io/ebeaninternal/server/cache/DefaultServerCacheTest.java +++ b/src/test/java/io/ebeaninternal/server/cache/DefaultServerCacheTest.java @@ -1,6 +1,8 @@ package io.ebeaninternal.server.cache; +import io.ebean.cache.ServerCacheConfig; import io.ebean.cache.ServerCacheOptions; +import io.ebean.cache.ServerCacheType; import org.junit.Test; import static org.junit.Assert.assertEquals; @@ -15,11 +17,13 @@ public class DefaultServerCacheTest { cacheOptions.setMaxSecsToLive(600); cacheOptions.setTrimFrequency(60); - return new DefaultServerCache("foo", null, cacheOptions); + ServerCacheConfig con = new ServerCacheConfig(ServerCacheType.BEAN, "foo", cacheOptions, null, null); + DefaultServerCacheConfig config = new DefaultServerCacheConfig(con); + return new DefaultServerCache(config); } @Test - public void testGetHitRatio() throws Exception { + public void testGetHitRatio() { DefaultServerCache cache = createCache(); assertEquals(0, cache.getHitRatio()); @@ -34,7 +38,7 @@ public class DefaultServerCacheTest { } @Test - public void testSize() throws Exception { + public void testSize() { DefaultServerCache cache = createCache(); assertEquals(0, cache.size()); @@ -50,39 +54,5 @@ public class DefaultServerCacheTest { assertEquals(0, cache.size()); } - @Test - public void trimFreq_halfIdle() throws Exception { - - DefaultServerCache cache = new DefaultServerCache("", null, null, 10000, 10, 20, 0); - assertEquals(cache.trimFrequency, 4); - } - - @Test - public void trimFreq_halfIdle_withRounding() throws Exception { - - DefaultServerCache cache = new DefaultServerCache("", null, null, 10000, 11, 20, 0); - assertEquals(cache.trimFrequency, 4); - } - - @Test - public void trimFreq_halfTTL() throws Exception { - - DefaultServerCache cache = new DefaultServerCache("", null, null, 10000, 0, 20, 0); - assertEquals(cache.trimFrequency, 9); - } - - @Test - public void trimFreq_halfTTL_withRounding() throws Exception { - - DefaultServerCache cache = new DefaultServerCache("", null, null, 10000, 0, 21, 0); - assertEquals(cache.trimFrequency, 9); - } - - @Test - public void trimFreq_explicit() throws Exception { - - DefaultServerCache cache = new DefaultServerCache("", null, null, 10000, 10, 20, 42); - assertEquals(cache.trimFrequency, 42); - } } diff --git a/src/test/java/io/ebeaninternal/server/cache/DefaultServerCache_RunEvictionTest.java b/src/test/java/io/ebeaninternal/server/cache/DefaultServerCache_RunEvictionTest.java index 4ba1d82a4..a2d630c98 100644 --- a/src/test/java/io/ebeaninternal/server/cache/DefaultServerCache_RunEvictionTest.java +++ b/src/test/java/io/ebeaninternal/server/cache/DefaultServerCache_RunEvictionTest.java @@ -1,7 +1,9 @@ package io.ebeaninternal.server.cache; +import io.ebean.cache.ServerCacheConfig; import io.ebean.cache.ServerCacheOptions; import io.ebean.cache.ServerCacheStatistics; +import io.ebean.cache.ServerCacheType; import org.junit.Ignore; import org.junit.Test; @@ -19,7 +21,8 @@ public class DefaultServerCache_RunEvictionTest { cacheOptions.setMaxSecsToLive(2); cacheOptions.setTrimFrequency(1); - return new DefaultServerCache("foo", null, cacheOptions); + ServerCacheConfig con = new ServerCacheConfig(ServerCacheType.BEAN, "foo", cacheOptions, null, null); + return new DefaultServerCache(new DefaultServerCacheConfig(con)); } private final DefaultServerCache cache; diff --git a/src/test/java/io/ebeaninternal/server/transaction/TableModStateTest.java b/src/test/java/io/ebeaninternal/server/transaction/TableModStateTest.java new file mode 100644 index 000000000..5abcb1820 --- /dev/null +++ b/src/test/java/io/ebeaninternal/server/transaction/TableModStateTest.java @@ -0,0 +1,44 @@ +package io.ebeaninternal.server.transaction; + +import org.junit.Test; + +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; + +import static org.junit.Assert.*; + +public class TableModStateTest { + + private TableModState tableModState = new TableModState(); + + @Test + public void isValid() { + + long now = System.currentTimeMillis(); + + tableModState.touch(setOf("one", "two", "three"), now); + + // empty + assertTrue(tableModState.isValid(Collections.emptySet(), 12L)); + + // no entry + assertTrue(tableModState.isValid(setOf("noEntry"), 12L)); + + // later timestamp + assertTrue(tableModState.isValid(setOf("one"), now + 1)); + assertTrue(tableModState.isValid(setOf("one", "two", "noEntry"), now + 1)); + + // invalid + assertFalse(tableModState.isValid(setOf("one"), now - 1)); + assertFalse(tableModState.isValid(setOf("one", "two"), now - 1)); + assertFalse(tableModState.isValid(setOf("three", "two"), now - 1)); + + } + + private Set setOf(String... tables) { + Set touched = new HashSet<>(); + Collections.addAll(touched, tables); + return touched; + } +} diff --git a/src/test/java/org/tests/cache/TestQueryCacheTableDependency.java b/src/test/java/org/tests/cache/TestQueryCacheTableDependency.java new file mode 100644 index 000000000..f578fdfdf --- /dev/null +++ b/src/test/java/org/tests/cache/TestQueryCacheTableDependency.java @@ -0,0 +1,59 @@ +package org.tests.cache; + +import io.ebean.BaseTestCase; +import io.ebean.Ebean; +import io.ebean.cache.ServerCache; +import org.junit.Test; +import org.tests.model.basic.Address; +import org.tests.model.basic.Customer; +import org.tests.model.basic.ResetBasicData; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +public class TestQueryCacheTableDependency extends BaseTestCase { + + @Test + public void testFindCountOnDependent() { + + ResetBasicData.reset(); + + ServerCache customerCache = Ebean.getServerCacheManager().getQueryCache(Customer.class); + customerCache.clear(); + + List
addrs = Ebean.find(Address.class) + .where().eq("line2", "St Lukes") + .findList(); + + int custs = Ebean.find(Customer.class).setUseQueryCache(true).setReadOnly(true) + .where().eq("billingAddress.line2", "St Lukes") + .findCount(); + + assertThat(custs).isEqualTo(3); + + custs = Ebean.find(Customer.class).setUseQueryCache(true).setReadOnly(true) + .where().eq("billingAddress.line2", "St Lukes") + .findCount(); + + assertThat(custs).isEqualTo(3); + + Address a1 = addrs.get(0); + a1.setLine2("St Lucky"); + Ebean.save(a1); + + custs = Ebean.find(Customer.class).setUseQueryCache(true).setReadOnly(true) + .where().eq("billingAddress.line2", "St Lukes") + .findCount(); + + assertThat(custs).isEqualTo(2); // cache says 3 + + + custs = Ebean.find(Customer.class).setUseQueryCache(true).setReadOnly(true) + .where().eq("billingAddress.line2", "St Lucky") + .findCount(); + + assertThat(custs).isEqualTo(1); + + } +} diff --git a/src/test/java/org/tests/model/basic/Address.java b/src/test/java/org/tests/model/basic/Address.java index 83d1b5c18..a11d38a0d 100644 --- a/src/test/java/org/tests/model/basic/Address.java +++ b/src/test/java/org/tests/model/basic/Address.java @@ -1,5 +1,6 @@ package org.tests.model.basic; +import io.ebean.annotation.InvalidateQueryCache; import org.tests.model.basic.metaannotation.SizeMedium; import javax.persistence.Column; @@ -14,6 +15,11 @@ import java.sql.Timestamp; /** * Address entity bean. */ +// Address is not L2 cached directly but it is joined to queries that are cached +// What InvalidateQueryCache means is that we propagate a table modification event +// when address is changed ... and cached queries that join to address will be +// invalidated accordingly. +@InvalidateQueryCache @Entity @Table(name = "o_address") public class Address { From 6a9ab49ef046915c710d9e887702999265c5af86 Mon Sep 17 00:00:00 2001 From: rob bygrave Date: Sat, 16 Jun 2018 00:09:03 +1200 Subject: [PATCH 2/4] #1427 - QueryCache should be cleared, if one of a dependent bean is updated Propagation across cluster. --- .../io/ebean/cache/ServerCacheFactory.java | 9 ++++++ .../ebean/cache/ServerCacheNotification.java | 29 +++++++++++++++++++ .../io/ebean/cache/ServerCacheNotify.java | 12 ++++++++ .../ebean/cache/ServerCacheNotifyPlugin.java | 17 +++++++++++ .../cache/DefaultServerCacheFactory.java | 18 ++++++++++-- .../server/core/InternalConfiguration.java | 17 +++++++++-- .../transaction/PostCommitProcessing.java | 6 ++-- .../server/transaction/TableModState.java | 27 ++++++++++++++--- .../transaction/TransactionManager.java | 5 ++++ .../TransactionManagerOptions.java | 5 +++- 10 files changed, 133 insertions(+), 12 deletions(-) create mode 100644 src/main/java/io/ebean/cache/ServerCacheNotification.java create mode 100644 src/main/java/io/ebean/cache/ServerCacheNotify.java create mode 100644 src/main/java/io/ebean/cache/ServerCacheNotifyPlugin.java diff --git a/src/main/java/io/ebean/cache/ServerCacheFactory.java b/src/main/java/io/ebean/cache/ServerCacheFactory.java index d59bfd3c4..54c71e7d1 100644 --- a/src/main/java/io/ebean/cache/ServerCacheFactory.java +++ b/src/main/java/io/ebean/cache/ServerCacheFactory.java @@ -10,4 +10,13 @@ public interface ServerCacheFactory { */ ServerCache createCache(ServerCacheConfig config); + /** + * Return a ServerCacheNotify that we will send ServerCacheNotification events to. + *

+ * This is used if a ServerCacheNotifyPlugin is not supplied. + *

+ * + * @param listener The listener that should be used to process the notification events. + */ + ServerCacheNotify createCacheNotify(ServerCacheNotify listener); } diff --git a/src/main/java/io/ebean/cache/ServerCacheNotification.java b/src/main/java/io/ebean/cache/ServerCacheNotification.java new file mode 100644 index 000000000..4b5b08a6f --- /dev/null +++ b/src/main/java/io/ebean/cache/ServerCacheNotification.java @@ -0,0 +1,29 @@ +package io.ebean.cache; + +import java.util.Set; + +/** + * Notification event that dependent tables have been modified. + *

+ * This is sent to other interested servers (in the cluster). + *

+ */ +public class ServerCacheNotification { + + private final long modifyTimestamp; + + private final Set dependentTables; + + public ServerCacheNotification(long modifyTimestamp, Set dependentTables) { + this.modifyTimestamp = modifyTimestamp; + this.dependentTables = dependentTables; + } + + public long getModifyTimestamp() { + return modifyTimestamp; + } + + public Set getDependentTables() { + return dependentTables; + } +} diff --git a/src/main/java/io/ebean/cache/ServerCacheNotify.java b/src/main/java/io/ebean/cache/ServerCacheNotify.java new file mode 100644 index 000000000..3e9f30a56 --- /dev/null +++ b/src/main/java/io/ebean/cache/ServerCacheNotify.java @@ -0,0 +1,12 @@ +package io.ebean.cache; + +/** + * Interface for both listening to notification changes and sending them to other members of the cluster. + */ +public interface ServerCacheNotify { + + /** + * Notify other server cache members of the table modifications or process the notifications. + */ + void notify(ServerCacheNotification notification); +} diff --git a/src/main/java/io/ebean/cache/ServerCacheNotifyPlugin.java b/src/main/java/io/ebean/cache/ServerCacheNotifyPlugin.java new file mode 100644 index 000000000..2e87a067c --- /dev/null +++ b/src/main/java/io/ebean/cache/ServerCacheNotifyPlugin.java @@ -0,0 +1,17 @@ +package io.ebean.cache; + +import io.ebean.config.ServerConfig; + +/** + * Plugin that provides a ServerCacheNotify implementation. + *

+ * Is supplied this will be used to send the ServerCacheNotification event to other cluster members. + *

+ */ +public interface ServerCacheNotifyPlugin { + + /** + * Create a ServerCacheNotify implementation given the server configuration. + */ + ServerCacheNotify create(ServerConfig serverConfig); +} diff --git a/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheFactory.java b/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheFactory.java index a519ea469..1b0476f3c 100644 --- a/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheFactory.java +++ b/src/main/java/io/ebeaninternal/server/cache/DefaultServerCacheFactory.java @@ -4,6 +4,8 @@ import io.ebean.BackgroundExecutor; import io.ebean.cache.ServerCache; import io.ebean.cache.ServerCacheConfig; import io.ebean.cache.ServerCacheFactory; +import io.ebean.cache.ServerCacheNotification; +import io.ebean.cache.ServerCacheNotify; /** @@ -16,14 +18,14 @@ class DefaultServerCacheFactory implements ServerCacheFactory { /** * Construct when l2 cache is disabled. */ - public DefaultServerCacheFactory() { + DefaultServerCacheFactory() { this.executor = null; } /** * Construct with executor service. */ - public DefaultServerCacheFactory(BackgroundExecutor executor) { + DefaultServerCacheFactory(BackgroundExecutor executor) { this.executor = executor; } @@ -43,4 +45,16 @@ class DefaultServerCacheFactory implements ServerCacheFactory { return cache; } + @Override + public ServerCacheNotify createCacheNotify(ServerCacheNotify listener) { + return new NoopServerCacheNotify(); + } + + private static class NoopServerCacheNotify implements ServerCacheNotify { + + @Override + public void notify(ServerCacheNotification notification) { + // do nothing + } + } } diff --git a/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java b/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java index 8d201d9c9..4beb65c58 100644 --- a/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java +++ b/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java @@ -5,6 +5,8 @@ import io.ebean.ExpressionFactory; import io.ebean.annotation.Platform; import io.ebean.cache.ServerCacheFactory; import io.ebean.cache.ServerCacheManager; +import io.ebean.cache.ServerCacheNotify; +import io.ebean.cache.ServerCacheNotifyPlugin; import io.ebean.cache.ServerCacheOptions; import io.ebean.cache.ServerCachePlugin; import io.ebean.config.ExternalTransactionManager; @@ -138,6 +140,8 @@ public class InternalConfiguration { private final ServerCachePlugin serverCachePlugin; + private ServerCacheNotify cacheNotify; + private boolean localL2Caching; private final ExpressionFactory expressionFactory; @@ -418,7 +422,8 @@ public class InternalConfiguration { TransactionManagerOptions options = new TransactionManagerOptions(notifyL2CacheInForeground, serverConfig, scopeManager, clusterManager, backgroundExecutor, - indexUpdateProcessor, beanDescriptorManager, dataSource(), profileHandler(), logManager, tableModState); + indexUpdateProcessor, beanDescriptorManager, dataSource(), profileHandler(), logManager, + tableModState, cacheNotify); if (serverConfig.isExplicitTransactionBeginMode()) { return new ExplicitTransactionManager(options); @@ -565,7 +570,7 @@ public class InternalConfiguration { if (iterator.hasNext()) { // use the cacheFactory (via classpath service loader) plugin = iterator.next(); - logger.debug("using ServerCacheFactory {}", serverCachePlugin.getClass()); + logger.debug("using ServerCacheFactory {}", plugin.getClass()); } else { // use the built in default l2 caching which is local cache based localL2Caching = true; @@ -587,6 +592,14 @@ public class InternalConfiguration { ServerCacheFactory factory = serverCachePlugin.create(serverConfig, backgroundExecutor); + ServerCacheNotifyPlugin notifyPlugin = serverConfig.service(ServerCacheNotifyPlugin.class); + if (notifyPlugin != null) { + // plugin supplied so use that to send notifications + cacheNotify = notifyPlugin.create(serverConfig); + } else { + cacheNotify = factory.createCacheNotify(tableModState); + } + // reasonable default settings are for a cache per bean type ServerCacheOptions beanOptions = new ServerCacheOptions(); beanOptions.setMaxSize(serverConfig.getCacheMaxSize()); diff --git a/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java b/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java index b715d2ee3..08a85fd7d 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java +++ b/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java @@ -22,7 +22,7 @@ import java.util.Set; * This includes Cluster notification, and BeanPersistListeners. *

*/ -public final class PostCommitProcessing { +final class PostCommitProcessing { private static final Logger logger = LoggerFactory.getLogger(PostCommitProcessing.class); @@ -51,7 +51,7 @@ public final class PostCommitProcessing { /** * Create for an external modification. */ - public PostCommitProcessing(ClusterManager clusterManager, TransactionManager manager, TransactionEvent event) { + PostCommitProcessing(ClusterManager clusterManager, TransactionManager manager, TransactionEvent event) { this.clusterManager = clusterManager; this.manager = manager; @@ -68,7 +68,7 @@ public final class PostCommitProcessing { /** * Create for a transaction. */ - public PostCommitProcessing(ClusterManager clusterManager, TransactionManager manager, SpiTransaction transaction) { + PostCommitProcessing(ClusterManager clusterManager, TransactionManager manager, SpiTransaction transaction) { this.clusterManager = clusterManager; this.manager = manager; diff --git a/src/main/java/io/ebeaninternal/server/transaction/TableModState.java b/src/main/java/io/ebeaninternal/server/transaction/TableModState.java index d2aa4143a..b876a4ed0 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/TableModState.java +++ b/src/main/java/io/ebeaninternal/server/transaction/TableModState.java @@ -2,6 +2,10 @@ package io.ebeaninternal.server.transaction; import io.ebean.cache.QueryCacheEntry; import io.ebean.cache.QueryCacheEntryValidate; +import io.ebean.cache.ServerCacheNotification; +import io.ebean.cache.ServerCacheNotify; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import java.util.Map; import java.util.Set; @@ -13,24 +17,28 @@ import java.util.concurrent.ConcurrentHashMap; * This information is used to validate entries in the L2 query caches. *

*/ -public class TableModState implements QueryCacheEntryValidate { +public class TableModState implements QueryCacheEntryValidate, ServerCacheNotify { + + private static final Logger log = LoggerFactory.getLogger("io.ebean.cache.TABLEMOD"); private Map tableModStamp = new ConcurrentHashMap<>(); /** * Set the modified timestamp on the tables that have been touched. */ - public void touch(Set touchedTables, long modTimestamp) { - + void touch(Set touchedTables, long modTimestamp) { for (String tableName : touchedTables) { tableModStamp.put(tableName, modTimestamp); } + if (log.isDebugEnabled()) { + log.debug("TableModState updated - " + tableModStamp); + } } /** * Return true if all the tables are valid based on timestamp comparison. */ - public boolean isValid(Set tables, long sinceTimestamp) { + boolean isValid(Set tables, long sinceTimestamp) { for (String tableName : tables) { Long modTime = tableModStamp.get(tableName); if (modTime != null && modTime > sinceTimestamp ) { @@ -48,4 +56,15 @@ public class TableModState implements QueryCacheEntryValidate { } return true; } + + /** + * Update the table modification timestamps based on remote table modification events. + */ + @Override + public void notify(ServerCacheNotification notification) { + + // TODO: Change to use ClockService ... + long modifyTimestamp = notification.getModifyTimestamp(); + touch(notification.getDependentTables(), modifyTimestamp); + } } diff --git a/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java b/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java index 0ff2894d6..b462b970b 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java +++ b/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java @@ -5,6 +5,8 @@ import io.ebean.ProfileLocation; import io.ebean.TxScope; import io.ebean.annotation.PersistBatch; import io.ebean.annotation.TxType; +import io.ebean.cache.ServerCacheNotification; +import io.ebean.cache.ServerCacheNotify; import io.ebean.config.CurrentTenantProvider; import io.ebean.config.dbplatform.DatabasePlatform; import io.ebean.config.dbplatform.DatabasePlatform.OnQueryOnly; @@ -137,6 +139,7 @@ public class TransactionManager implements SpiTransactionManager { private final TransactionScopeManager scopeManager; private final TableModState tableModState; + private final ServerCacheNotify cacheNotify; /** * Create the TransactionManager @@ -160,6 +163,7 @@ public class TransactionManager implements SpiTransactionManager { this.serverName = options.config.getName(); this.scopeManager = options.scopeManager; this.tableModState = options.tableModState; + this.cacheNotify = options.cacheNotify; this.backgroundExecutor = options.backgroundExecutor; this.dataSourceSupplier = options.dataSourceSupplier; this.docStoreActive = options.config.getDocStoreConfig().isActive(); @@ -510,6 +514,7 @@ public class TransactionManager implements SpiTransactionManager { if (viewInvalidation) { beanDescriptorManager.processViewInvalidation(touchedTables); } + cacheNotify.notify(new ServerCacheNotification(modTimestamp, touchedTables)); } /** diff --git a/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java b/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java index 16f5dafe9..1b8ce39a8 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java +++ b/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java @@ -1,6 +1,7 @@ package io.ebeaninternal.server.transaction; import io.ebean.BackgroundExecutor; +import io.ebean.cache.ServerCacheNotify; import io.ebean.config.ServerConfig; import io.ebeaninternal.api.SpiLogManager; import io.ebeaninternal.api.SpiProfileHandler; @@ -25,12 +26,13 @@ public class TransactionManagerOptions { final TransactionScopeManager scopeManager; final SpiLogManager logManager; final TableModState tableModState; + final ServerCacheNotify cacheNotify; public TransactionManagerOptions(boolean notifyL2CacheInForeground, ServerConfig config, TransactionScopeManager scopeManager, ClusterManager clusterManager, BackgroundExecutor backgroundExecutor, DocStoreUpdateProcessor docStoreUpdateProcessor, BeanDescriptorManager descMgr, DataSourceSupplier dataSourceSupplier, SpiProfileHandler profileHandler, - SpiLogManager logManager, TableModState tableModState) { + SpiLogManager logManager, TableModState tableModState, ServerCacheNotify cacheNotify) { this.notifyL2CacheInForeground = notifyL2CacheInForeground; this.config = config; @@ -43,6 +45,7 @@ public class TransactionManagerOptions { this.profileHandler = profileHandler; this.logManager = logManager; this.tableModState = tableModState; + this.cacheNotify = cacheNotify; } } From bd3c99e67f79ef588fa0b8e58f7790aa97143046 Mon Sep 17 00:00:00 2001 From: rob bygrave Date: Sat, 16 Jun 2018 00:22:44 +1200 Subject: [PATCH 3/4] #1427 - QueryCache should be cleared, if one of a dependent bean is updated Wrap Clock as ClockService to use in TransactionManager and DefaultServer (and allow for changing the Clock implementation for testing). --- .../server/core/ClockService.java | 29 +++++++++++++++++++ .../server/core/DefaultServer.java | 8 ++--- .../server/core/InternalConfiguration.java | 9 +++++- .../ImplicitReadOnlyTransaction.java | 2 +- .../server/transaction/JdbcTransaction.java | 2 +- .../transaction/TransactionManager.java | 9 ++++-- .../TransactionManagerOptions.java | 5 +++- 7 files changed, 53 insertions(+), 11 deletions(-) create mode 100644 src/main/java/io/ebeaninternal/server/core/ClockService.java diff --git a/src/main/java/io/ebeaninternal/server/core/ClockService.java b/src/main/java/io/ebeaninternal/server/core/ClockService.java new file mode 100644 index 000000000..227a7b1cf --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/core/ClockService.java @@ -0,0 +1,29 @@ +package io.ebeaninternal.server.core; + +import java.time.Clock; + +/** + * Wraps the Clock such that we can change the Clock for testing purposes. + */ +public class ClockService { + + private Clock clock; + + public ClockService(Clock clock) { + this.clock = clock; + } + + /** + * Change the clock for testing purposes. + */ + public void setClock(Clock clock) { + this.clock = clock; + } + + /** + * Return the Clock current timestamp in millis. + */ + public long nowMillis() { + return clock.millis(); + } +} diff --git a/src/main/java/io/ebeaninternal/server/core/DefaultServer.java b/src/main/java/io/ebeaninternal/server/core/DefaultServer.java index 89fe28b6b..a540ecb1b 100644 --- a/src/main/java/io/ebeaninternal/server/core/DefaultServer.java +++ b/src/main/java/io/ebeaninternal/server/core/DefaultServer.java @@ -154,7 +154,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer { /** * Clock to use for WhenModified and WhenCreated. */ - private Clock clock; + private ClockService clockService; private final CallStackFactory callStackFactory; @@ -283,7 +283,7 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer { this.beanLoader = new DefaultBeanLoader(this); this.jsonContext = config.createJsonContext(this); this.dataTimeZone = config.getDataTimeZone(); - this.clock = config.getServerConfig().getClock(); + this.clockService = config.getClockService(); DocStoreIntegration docStoreComponents = config.createDocStoreIntegration(this); this.transactionManager = config.createTransactionManager(docStoreComponents.updateProcessor()); @@ -515,12 +515,12 @@ public final class DefaultServer implements SpiServer, SpiEbeanServer { @Override public long clockNow() { - return clock.millis(); + return clockService.nowMillis(); } @Override public void setClock(Clock clock) { - this.clock = clock; + this.clockService.setClock(clock); } @Override diff --git a/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java b/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java index 4beb65c58..f3b537217 100644 --- a/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java +++ b/src/main/java/io/ebeaninternal/server/core/InternalConfiguration.java @@ -122,6 +122,8 @@ public class InternalConfiguration { private final DtoBeanManager dtoBeanManager; + private final ClockService clockService; + private final DataTimeZone dataTimeZone; private final Binder binder; @@ -166,6 +168,7 @@ public class InternalConfiguration { this.online = online; this.serverConfig = serverConfig; + this.clockService = new ClockService(serverConfig.getClock()); this.logManager = initLogManager(); this.docStoreFactory = initDocStoreFactory(serverConfig.service(DocStoreFactory.class)); this.jsonFactory = serverConfig.getJsonFactory(); @@ -233,6 +236,10 @@ public class InternalConfiguration { return docStoreFactory; } + public ClockService getClockService() { + return clockService; + } + /** * Check if this is a SpiServerPlugin and if so 'collect' it to give the complete list * later on the DefaultServer for late call to configure(). @@ -423,7 +430,7 @@ public class InternalConfiguration { TransactionManagerOptions options = new TransactionManagerOptions(notifyL2CacheInForeground, serverConfig, scopeManager, clusterManager, backgroundExecutor, indexUpdateProcessor, beanDescriptorManager, dataSource(), profileHandler(), logManager, - tableModState, cacheNotify); + tableModState, cacheNotify, clockService); if (serverConfig.isExplicitTransactionBeginMode()) { return new ExplicitTransactionManager(options); diff --git a/src/main/java/io/ebeaninternal/server/transaction/ImplicitReadOnlyTransaction.java b/src/main/java/io/ebeaninternal/server/transaction/ImplicitReadOnlyTransaction.java index 1ccc44dbd..44e5b4bd1 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/ImplicitReadOnlyTransaction.java +++ b/src/main/java/io/ebeaninternal/server/transaction/ImplicitReadOnlyTransaction.java @@ -80,7 +80,7 @@ class ImplicitReadOnlyTransaction implements SpiTransaction, TxnProfileEventCode this.connection = connection; this.persistenceContext = new DefaultPersistenceContext(); this.startNanos = System.nanoTime(); - this.startMillis = manager.clockNowEpoch(); + this.startMillis = manager.clockNowMillis(); } /** diff --git a/src/main/java/io/ebeaninternal/server/transaction/JdbcTransaction.java b/src/main/java/io/ebeaninternal/server/transaction/JdbcTransaction.java index da8e53781..2fbaab79a 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/JdbcTransaction.java +++ b/src/main/java/io/ebeaninternal/server/transaction/JdbcTransaction.java @@ -212,7 +212,7 @@ public class JdbcTransaction implements SpiTransaction, TxnProfileEventCodes { this.batchOnCascadeMode = PersistBatch.NONE; this.onQueryOnly = OnQueryOnly.ROLLBACK; } else { - this.startMillis = manager.clockNowEpoch(); + this.startMillis = manager.clockNowMillis(); this.logSql = manager.isLogSql(); this.logSummary = manager.isLogSummary(); this.skipCacheAfterWrite = manager.isSkipCacheAfterWrite(); diff --git a/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java b/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java index b462b970b..4a007a9b3 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java +++ b/src/main/java/io/ebeaninternal/server/transaction/TransactionManager.java @@ -29,6 +29,7 @@ import io.ebeaninternal.metric.MetricFactory; import io.ebeaninternal.metric.TimedMetric; import io.ebeaninternal.metric.TimedMetricMap; import io.ebeaninternal.server.cluster.ClusterManager; +import io.ebeaninternal.server.core.ClockService; import io.ebeaninternal.server.deploy.BeanDescriptorManager; import io.ebeaninternal.server.profile.TimedProfileLocation; import io.ebeaninternal.server.profile.TimedProfileLocationRegistry; @@ -140,6 +141,7 @@ public class TransactionManager implements SpiTransactionManager { private final TableModState tableModState; private final ServerCacheNotify cacheNotify; + private final ClockService clockService; /** * Create the TransactionManager @@ -164,6 +166,7 @@ public class TransactionManager implements SpiTransactionManager { this.scopeManager = options.scopeManager; this.tableModState = options.tableModState; this.cacheNotify = options.cacheNotify; + this.clockService = options.clockService; this.backgroundExecutor = options.backgroundExecutor; this.dataSourceSupplier = options.dataSourceSupplier; this.docStoreActive = options.config.getDocStoreConfig().isActive(); @@ -188,8 +191,8 @@ public class TransactionManager implements SpiTransactionManager { /** * Return the NOW timestamp in epoch millis. */ - public long clockNowEpoch() { - return System.currentTimeMillis(); // TODO: Review when we supply a Clock via ServerConfig. + public long clockNowMillis() { + return clockService.nowMillis(); } /** @@ -448,7 +451,7 @@ public class TransactionManager implements SpiTransactionManager { private void externalModificationEvent(TransactionEventTable tableEvents) { - TransactionEvent event = new TransactionEvent(clockNowEpoch()); + TransactionEvent event = new TransactionEvent(clockNowMillis()); event.add(tableEvents); PostCommitProcessing postCommit = new PostCommitProcessing(clusterManager, this, event); diff --git a/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java b/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java index 1b8ce39a8..f55389a19 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java +++ b/src/main/java/io/ebeaninternal/server/transaction/TransactionManagerOptions.java @@ -6,6 +6,7 @@ import io.ebean.config.ServerConfig; import io.ebeaninternal.api.SpiLogManager; import io.ebeaninternal.api.SpiProfileHandler; import io.ebeaninternal.server.cluster.ClusterManager; +import io.ebeaninternal.server.core.ClockService; import io.ebeaninternal.server.deploy.BeanDescriptorManager; import io.ebeanservice.docstore.api.DocStoreUpdateProcessor; @@ -27,12 +28,13 @@ public class TransactionManagerOptions { final SpiLogManager logManager; final TableModState tableModState; final ServerCacheNotify cacheNotify; + final ClockService clockService; public TransactionManagerOptions(boolean notifyL2CacheInForeground, ServerConfig config, TransactionScopeManager scopeManager, ClusterManager clusterManager, BackgroundExecutor backgroundExecutor, DocStoreUpdateProcessor docStoreUpdateProcessor, BeanDescriptorManager descMgr, DataSourceSupplier dataSourceSupplier, SpiProfileHandler profileHandler, - SpiLogManager logManager, TableModState tableModState, ServerCacheNotify cacheNotify) { + SpiLogManager logManager, TableModState tableModState, ServerCacheNotify cacheNotify, ClockService clockService) { this.notifyL2CacheInForeground = notifyL2CacheInForeground; this.config = config; @@ -46,6 +48,7 @@ public class TransactionManagerOptions { this.logManager = logManager; this.tableModState = tableModState; this.cacheNotify = cacheNotify; + this.clockService = clockService; } } From 9c73309dfd0bfe935202183e228a9b0282262fb9 Mon Sep 17 00:00:00 2001 From: rob bygrave Date: Sat, 16 Jun 2018 15:15:54 +1200 Subject: [PATCH 4/4] #1427 - QueryCache should be cleared, if one of a dependent bean is updated Update RemoteTransactionEvent, add (back) common binary message handling (Common to TCP, K8s and upcoming binary pubsub) --- .../api/TransactionEventTable.java | 6 +- .../server/cache/RemoteCacheEvent.java | 4 +- .../server/cluster/ClusterManager.java | 2 +- .../server/cluster/MessageServerProvider.java | 14 ++ .../binarymessage/BinaryDataReader.java | 75 +++++++++ .../binarymessage/BinaryDataWriter.java | 49 ++++++ .../{ => binarymessage}/BinaryMessage.java | 3 +- .../BinaryMessageList.java | 2 +- .../cluster/binarymessage/ClusterMessage.java | 147 ++++++++++++++++++ .../InvalidMessageException.java | 8 + .../binarymessage/MessageReadWrite.java | 37 +++++ .../server/cluster/binarymessage/MsgKeys.java | 19 +++ .../server/transaction/BeanPersistIds.java | 18 ++- .../transaction/PostCommitProcessing.java | 4 + .../server/transaction/RemoteTableMod.java | 55 +++++++ .../transaction/RemoteTransactionEvent.java | 16 +- .../binarymessage/MessageReadWriteTest.java | 118 ++++++++++++++ 17 files changed, 563 insertions(+), 14 deletions(-) create mode 100644 src/main/java/io/ebeaninternal/server/cluster/MessageServerProvider.java create mode 100644 src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryDataReader.java create mode 100644 src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryDataWriter.java rename src/main/java/io/ebeaninternal/server/cluster/{ => binarymessage}/BinaryMessage.java (94%) rename src/main/java/io/ebeaninternal/server/cluster/{ => binarymessage}/BinaryMessageList.java (85%) create mode 100644 src/main/java/io/ebeaninternal/server/cluster/binarymessage/ClusterMessage.java create mode 100644 src/main/java/io/ebeaninternal/server/cluster/binarymessage/InvalidMessageException.java create mode 100644 src/main/java/io/ebeaninternal/server/cluster/binarymessage/MessageReadWrite.java create mode 100644 src/main/java/io/ebeaninternal/server/cluster/binarymessage/MsgKeys.java create mode 100644 src/main/java/io/ebeaninternal/server/transaction/RemoteTableMod.java create mode 100644 src/test/java/io/ebeaninternal/server/cluster/binarymessage/MessageReadWriteTest.java diff --git a/src/main/java/io/ebeaninternal/api/TransactionEventTable.java b/src/main/java/io/ebeaninternal/api/TransactionEventTable.java index cde507c1a..fe5338cf4 100644 --- a/src/main/java/io/ebeaninternal/api/TransactionEventTable.java +++ b/src/main/java/io/ebeaninternal/api/TransactionEventTable.java @@ -1,8 +1,8 @@ package io.ebeaninternal.api; import io.ebean.event.BulkTableEvent; -import io.ebeaninternal.server.cluster.BinaryMessage; -import io.ebeaninternal.server.cluster.BinaryMessageList; +import io.ebeaninternal.server.cluster.binarymessage.BinaryMessage; +import io.ebeaninternal.server.cluster.binarymessage.BinaryMessageList; import java.io.DataInput; import java.io.DataOutputStream; @@ -102,7 +102,7 @@ public final class TransactionEventTable implements Serializable { os.writeBoolean(insert); os.writeBoolean(update); os.writeBoolean(delete); - + os.close(); msgList.add(msg); } diff --git a/src/main/java/io/ebeaninternal/server/cache/RemoteCacheEvent.java b/src/main/java/io/ebeaninternal/server/cache/RemoteCacheEvent.java index e0def309f..2d04df0f5 100644 --- a/src/main/java/io/ebeaninternal/server/cache/RemoteCacheEvent.java +++ b/src/main/java/io/ebeaninternal/server/cache/RemoteCacheEvent.java @@ -1,7 +1,7 @@ package io.ebeaninternal.server.cache; -import io.ebeaninternal.server.cluster.BinaryMessage; -import io.ebeaninternal.server.cluster.BinaryMessageList; +import io.ebeaninternal.server.cluster.binarymessage.BinaryMessage; +import io.ebeaninternal.server.cluster.binarymessage.BinaryMessageList; import java.io.DataInput; import java.io.DataOutputStream; diff --git a/src/main/java/io/ebeaninternal/server/cluster/ClusterManager.java b/src/main/java/io/ebeaninternal/server/cluster/ClusterManager.java index 6fd56c713..6f02f5b50 100644 --- a/src/main/java/io/ebeaninternal/server/cluster/ClusterManager.java +++ b/src/main/java/io/ebeaninternal/server/cluster/ClusterManager.java @@ -13,7 +13,7 @@ import java.util.concurrent.ConcurrentHashMap; /** * Manages the cluster service. */ -public class ClusterManager { +public class ClusterManager implements MessageServerProvider { private static final Logger clusterLogger = LoggerFactory.getLogger("io.ebean.Cluster"); diff --git a/src/main/java/io/ebeaninternal/server/cluster/MessageServerProvider.java b/src/main/java/io/ebeaninternal/server/cluster/MessageServerProvider.java new file mode 100644 index 000000000..f3c388d2b --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/cluster/MessageServerProvider.java @@ -0,0 +1,14 @@ +package io.ebeaninternal.server.cluster; + +import io.ebean.EbeanServer; + +/** + * Returns EbeanServer instances for remote message reading. + */ +public interface MessageServerProvider { + + /** + * Return the EbeanServer instance by name. + */ + EbeanServer getServer(String name); +} diff --git a/src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryDataReader.java b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryDataReader.java new file mode 100644 index 000000000..f097172f4 --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryDataReader.java @@ -0,0 +1,75 @@ +package io.ebeaninternal.server.cluster.binarymessage; + +import io.ebeaninternal.api.SpiEbeanServer; +import io.ebeaninternal.api.TransactionEventTable; +import io.ebeaninternal.server.cache.RemoteCacheEvent; +import io.ebeaninternal.server.cluster.MessageServerProvider; +import io.ebeaninternal.server.transaction.BeanPersistIds; +import io.ebeaninternal.server.transaction.RemoteTableMod; +import io.ebeaninternal.server.transaction.RemoteTransactionEvent; + +import java.io.ByteArrayInputStream; +import java.io.DataInputStream; +import java.io.IOException; + +/** + * Reads the binary message returning RemoteTransactionEvent. + */ +class BinaryDataReader { + + private final MessageServerProvider serverProvider; + private final DataInputStream dataInput; + + private SpiEbeanServer server; + private RemoteTransactionEvent event; + + BinaryDataReader(MessageServerProvider serverProvider, byte[] data) { + this.serverProvider = serverProvider; + this.dataInput = new DataInputStream(new ByteArrayInputStream(data)); + } + + /** + * Read the binary message returning a RemoteTransactionEvent. + */ + RemoteTransactionEvent read() throws IOException { + + String serverName = dataInput.readUTF(); + + this.server = (SpiEbeanServer) serverProvider.getServer(serverName); + if (server == null) { + throw new IllegalStateException("EbeanServer not found for name [" + serverName + "]"); + } + this.event = new RemoteTransactionEvent(server); + boolean more = dataInput.readBoolean(); + while (more) { + readMessage(); + more = dataInput.readBoolean(); + } + return event; + } + + private void readMessage() throws IOException { + + int msgType = dataInput.readInt(); + switch (msgType) { + case BinaryMessage.TYPE_BEANIUD: + event.addBeanPersistIds(BeanPersistIds.readBinaryMessage(server, dataInput)); + break; + + case BinaryMessage.TYPE_TABLEIUD: + event.addTableIUD(TransactionEventTable.TableIUD.readBinaryMessage(dataInput)); + break; + + case BinaryMessage.TYPE_CACHE: + event.addRemoteCacheEvent(RemoteCacheEvent.readBinaryMessage(dataInput)); + break; + + case BinaryMessage.TYPE_TABLEMOD: + event.addRemoteTableMod(RemoteTableMod.readBinaryMessage(dataInput)); + break; + + default: + throw new RuntimeException("Invalid Transaction msgType " + msgType); + } + } +} diff --git a/src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryDataWriter.java b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryDataWriter.java new file mode 100644 index 000000000..8f08bba56 --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryDataWriter.java @@ -0,0 +1,49 @@ +package io.ebeaninternal.server.cluster.binarymessage; + +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; +import java.io.IOException; + +/** + * Writes BinaryMessageList to DataHolder raw byte[]. + */ +class BinaryDataWriter { + + private final BinaryMessageList messageList; + + private final ByteArrayOutputStream buffer; + + private final DataOutputStream dataOut; + + private final String serverName; + + BinaryDataWriter(String serverName, BinaryMessageList messageList) { + this.serverName = serverName; + this.messageList = messageList; + this.buffer = new ByteArrayOutputStream(256); + this.dataOut = new DataOutputStream(buffer); + } + + /** + * Write the message as raw byte[]. + */ + byte[] write() throws IOException { + + //TODO: Add Protocol version and "podId" to support pubsub cluster + //dataOut.writeInt(MsgKeys.MESSAGE_KEY); + dataOut.writeUTF(serverName); + for (BinaryMessage msg : messageList.getList()) { + write(msg); + } + dataOut.writeBoolean(false); + dataOut.flush(); + return buffer.toByteArray(); + } + + private void write(BinaryMessage msg) throws IOException { + + dataOut.writeBoolean(true); + dataOut.write(msg.getByteArray()); + } + +} diff --git a/src/main/java/io/ebeaninternal/server/cluster/BinaryMessage.java b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryMessage.java similarity index 94% rename from src/main/java/io/ebeaninternal/server/cluster/BinaryMessage.java rename to src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryMessage.java index b78b3f7eb..158df57da 100644 --- a/src/main/java/io/ebeaninternal/server/cluster/BinaryMessage.java +++ b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryMessage.java @@ -1,4 +1,4 @@ -package io.ebeaninternal.server.cluster; +package io.ebeaninternal.server.cluster.binarymessage; import java.io.ByteArrayOutputStream; import java.io.DataOutputStream; @@ -24,6 +24,7 @@ public class BinaryMessage { public static final int TYPE_BEANIUD = 1; public static final int TYPE_TABLEIUD = 2; public static final int TYPE_CACHE = 3; + public static final int TYPE_TABLEMOD = 4; public static final int TYPE_MSGACK = 8; public static final int TYPE_MSGRESEND = 9; diff --git a/src/main/java/io/ebeaninternal/server/cluster/BinaryMessageList.java b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryMessageList.java similarity index 85% rename from src/main/java/io/ebeaninternal/server/cluster/BinaryMessageList.java rename to src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryMessageList.java index f84b7995c..fd057fd03 100644 --- a/src/main/java/io/ebeaninternal/server/cluster/BinaryMessageList.java +++ b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/BinaryMessageList.java @@ -1,4 +1,4 @@ -package io.ebeaninternal.server.cluster; +package io.ebeaninternal.server.cluster.binarymessage; import java.util.ArrayList; import java.util.List; diff --git a/src/main/java/io/ebeaninternal/server/cluster/binarymessage/ClusterMessage.java b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/ClusterMessage.java new file mode 100644 index 000000000..2e6b189ea --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/ClusterMessage.java @@ -0,0 +1,147 @@ +package io.ebeaninternal.server.cluster.binarymessage; + +import java.io.DataInputStream; +import java.io.DataOutputStream; +import java.io.IOException; + +/** + * The message broadcast around the cluster. + */ +public class ClusterMessage { + + private static final int MAX_LENGTH = 10 * 1024 * 1024; + + private final String registerIp; + + private final String podName; + + private final boolean register; + + private final byte[] data; + + /** + * Create a register message. + */ + public static ClusterMessage register(String registerIp, boolean register, String podName) { + return new ClusterMessage(registerIp, register, podName); + } + + /** + * Create a transaction message. + */ + public static ClusterMessage transEvent(byte[] data) { + return new ClusterMessage(data); + } + + /** + * Create for register online/offline message. + */ + private ClusterMessage(String registerIp, boolean register, String podName) { + this.registerIp = registerIp; + this.register = register; + this.podName = podName; + this.data = null; + } + + /** + * Create for a transaction message. + */ + private ClusterMessage(byte[] data) { + this.data = data; + this.registerIp = null; + this.podName = null; + this.register = false; + } + + public String toString() { + StringBuilder sb = new StringBuilder(); + if (registerIp != null) { + sb.append("register "); + sb.append(register); + sb.append(" "); + sb.append(registerIp); + } else { + sb.append("[data]"); + } + return sb.toString(); + } + + /** + * Return true if this is a register event as opposed to a transaction message. + */ + public boolean isRegisterEvent() { + return registerIp != null; + } + + /** + * Return the register host for online/offline message. + */ + public String getRegisterIp() { + return registerIp; + } + + public String getPodName() { + return podName; + } + + /** + * Return true if register is true for a online/offline message. + */ + public boolean isRegister() { + return register; + } + + /** + * Return the raw message data. + */ + public byte[] getData() { + return data; + } + + /** + * Write the message in binary form. + */ + public void write(DataOutputStream dataOutput) throws IOException { + + if (data != null) { + // write data message + dataOutput.writeInt(MsgKeys.DATA); + dataOutput.writeInt(data.length); + dataOutput.write(data); + } else { + // write header message + dataOutput.writeInt(MsgKeys.HEADER); + dataOutput.writeUTF(getRegisterIp()); + dataOutput.writeBoolean(register); + dataOutput.writeUTF(getPodName()); + } + dataOutput.flush(); + } + + /** + * Read the message from binary form. + */ + public static ClusterMessage read(DataInputStream dataInput) throws IOException, InvalidMessageException { + + int key = dataInput.readInt(); + if (key == MsgKeys.DATA) { + int length = dataInput.readInt(); + if (length > MAX_LENGTH) { + throw new IOException("Message data too large length:"+length); + } + byte[] data = new byte[length]; + dataInput.readFully(data); + return new ClusterMessage(data); + + } else if (key == MsgKeys.HEADER) { + String host = dataInput.readUTF(); + boolean registered = dataInput.readBoolean(); + String podName = dataInput.readUTF(); + return new ClusterMessage(host, registered, podName); + + } else { + throw new InvalidMessageException("Invalid message key:" + key); + } + } + +} diff --git a/src/main/java/io/ebeaninternal/server/cluster/binarymessage/InvalidMessageException.java b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/InvalidMessageException.java new file mode 100644 index 000000000..d05d0b3c2 --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/InvalidMessageException.java @@ -0,0 +1,8 @@ +package io.ebeaninternal.server.cluster.binarymessage; + +public class InvalidMessageException extends Exception { + + public InvalidMessageException(String message) { + super(message); + } +} diff --git a/src/main/java/io/ebeaninternal/server/cluster/binarymessage/MessageReadWrite.java b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/MessageReadWrite.java new file mode 100644 index 000000000..fe7cab1c2 --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/MessageReadWrite.java @@ -0,0 +1,37 @@ +package io.ebeaninternal.server.cluster.binarymessage; + +import io.ebeaninternal.server.cluster.MessageServerProvider; +import io.ebeaninternal.server.transaction.RemoteTransactionEvent; + +import java.io.IOException; + +/** + * Mechanism to convert RemoteTransactionEvent to/from byte[] content. + */ +public class MessageReadWrite { + + private final MessageServerProvider serverProvider; + + public MessageReadWrite(MessageServerProvider serverProvider) { + this.serverProvider = serverProvider; + } + + /** + * Convert the RemoteTransactionEvent to raw byte[] content. + */ + public byte[] write(RemoteTransactionEvent transEvent) throws IOException { + + BinaryMessageList messageList = new BinaryMessageList(); + transEvent.writeBinaryMessage(messageList); + + return new BinaryDataWriter(transEvent.getServerName(), messageList).write(); + } + + /** + * Convert the byte[] content to RemoteTransactionEvent. + */ + public RemoteTransactionEvent read(byte[] data) throws IOException { + + return new BinaryDataReader(serverProvider, data).read(); + } +} diff --git a/src/main/java/io/ebeaninternal/server/cluster/binarymessage/MsgKeys.java b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/MsgKeys.java new file mode 100644 index 000000000..9daeaa9bd --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/cluster/binarymessage/MsgKeys.java @@ -0,0 +1,19 @@ +package io.ebeaninternal.server.cluster.binarymessage; + +public interface MsgKeys { + + /** + * Used to identify client on connection initiation. + */ + int HELLO = 182; + + /** + * Used to confirm protocol on reading messages. + */ + int HEADER = 11; + + /** + * Used to confirm protocol on reading messages. + */ + int DATA = 12; +} diff --git a/src/main/java/io/ebeaninternal/server/transaction/BeanPersistIds.java b/src/main/java/io/ebeaninternal/server/transaction/BeanPersistIds.java index cf77829ee..934877759 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/BeanPersistIds.java +++ b/src/main/java/io/ebeaninternal/server/transaction/BeanPersistIds.java @@ -1,8 +1,8 @@ package io.ebeaninternal.server.transaction; import io.ebeaninternal.api.SpiEbeanServer; -import io.ebeaninternal.server.cluster.BinaryMessage; -import io.ebeaninternal.server.cluster.BinaryMessageList; +import io.ebeaninternal.server.cluster.binarymessage.BinaryMessage; +import io.ebeaninternal.server.cluster.binarymessage.BinaryMessageList; import io.ebeaninternal.server.core.PersistRequest; import io.ebeaninternal.server.deploy.BeanDescriptor; import io.ebeaninternal.server.deploy.id.IdBinder; @@ -140,7 +140,7 @@ public class BeanPersistIds { idBinder.writeData(os, idList.get(i)); } - os.flush(); + os.close(); msgList.add(m); } while (i < eof); @@ -167,7 +167,7 @@ public class BeanPersistIds { return sb.toString(); } - void addId(PersistRequest.Type type, Serializable id) { + public void addId(PersistRequest.Type type, Serializable id) { switch (type) { case INSERT: addInsertId(id); @@ -210,10 +210,18 @@ public class BeanPersistIds { return beanDescriptor; } - List getDeleteIds() { + public List getDeleteIds() { return deleteIds; } + public List getInsertIds() { + return insertIds; + } + + public List getUpdateIds() { + return updateIds; + } + /** * Notify the cache of this event that came from another server in the cluster. */ diff --git a/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java b/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java index 08a85fd7d..c84593646 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java +++ b/src/main/java/io/ebeaninternal/server/transaction/PostCommitProcessing.java @@ -211,6 +211,10 @@ final class PostCommitProcessing { RemoteTransactionEvent remoteTransactionEvent = new RemoteTransactionEvent(serverName); + Set touched = cacheChanges.touchedTables(); + if (touched != null && !touched.isEmpty()) { + remoteTransactionEvent.addRemoteTableMod(new RemoteTableMod(cacheChanges.modificationTimestamp(), touched)); + } if (beanPersistIdMap != null) { for (BeanPersistIds beanPersist : beanPersistIdMap.values()) { remoteTransactionEvent.addBeanPersistIds(beanPersist); diff --git a/src/main/java/io/ebeaninternal/server/transaction/RemoteTableMod.java b/src/main/java/io/ebeaninternal/server/transaction/RemoteTableMod.java new file mode 100644 index 000000000..1c6eda418 --- /dev/null +++ b/src/main/java/io/ebeaninternal/server/transaction/RemoteTableMod.java @@ -0,0 +1,55 @@ +package io.ebeaninternal.server.transaction; + +import io.ebeaninternal.server.cluster.binarymessage.BinaryMessage; +import io.ebeaninternal.server.cluster.binarymessage.BinaryMessageList; + +import java.io.DataInput; +import java.io.DataOutputStream; +import java.io.IOException; +import java.util.LinkedHashSet; +import java.util.Set; + +public class RemoteTableMod { + + private final long timestamp; + + private final Set tables; + + public RemoteTableMod(long timestamp, Set tables) { + this.timestamp = timestamp; + this.tables = tables; + } + + public long getTimestamp() { + return timestamp; + } + + public Set getTables() { + return tables; + } + + public static RemoteTableMod readBinaryMessage(DataInput dataInput) throws IOException { + + long timestamp = dataInput.readLong(); + int count = dataInput.readInt(); + + Set tables = new LinkedHashSet<>(); + for (int i = 0; i < count; i++) { + tables.add(dataInput.readUTF()); + } + return new RemoteTableMod(timestamp, tables); + } + + public void writeBinary(BinaryMessageList msgList) throws IOException { + BinaryMessage msg = new BinaryMessage(tables.size() * 20 + 30); + DataOutputStream os = msg.getOs(); + os.writeInt(BinaryMessage.TYPE_TABLEMOD); + os.writeLong(timestamp); + os.writeInt(tables.size()); + for (String table : tables) { + os.writeUTF(table); + } + os.close(); + msgList.add(msg); + } +} diff --git a/src/main/java/io/ebeaninternal/server/transaction/RemoteTransactionEvent.java b/src/main/java/io/ebeaninternal/server/transaction/RemoteTransactionEvent.java index f6047d209..54112ca66 100644 --- a/src/main/java/io/ebeaninternal/server/transaction/RemoteTransactionEvent.java +++ b/src/main/java/io/ebeaninternal/server/transaction/RemoteTransactionEvent.java @@ -3,7 +3,7 @@ package io.ebeaninternal.server.transaction; import io.ebeaninternal.api.SpiEbeanServer; import io.ebeaninternal.api.TransactionEventTable.TableIUD; import io.ebeaninternal.server.cache.RemoteCacheEvent; -import io.ebeaninternal.server.cluster.BinaryMessageList; +import io.ebeaninternal.server.cluster.binarymessage.BinaryMessageList; import java.io.IOException; import java.util.ArrayList; @@ -19,6 +19,8 @@ public class RemoteTransactionEvent implements Runnable { private RemoteCacheEvent remoteCacheEvent; + private RemoteTableMod remoteTableMod; + private String serverName; private transient SpiEbeanServer server; @@ -53,6 +55,10 @@ public class RemoteTransactionEvent implements Runnable { public void writeBinaryMessage(BinaryMessageList msgList) throws IOException { + if (remoteTableMod != null) { + remoteTableMod.writeBinary(msgList); + } + if (tableList != null) { for (TableIUD aTableList : tableList) { aTableList.writeBinaryMessage(msgList); @@ -113,6 +119,10 @@ public class RemoteTransactionEvent implements Runnable { tableList.add(tableIud); } + public void addRemoteTableMod(RemoteTableMod remoteTableMod) { + this.remoteTableMod = remoteTableMod; + } + public String getServerName() { return serverName; } @@ -141,4 +151,8 @@ public class RemoteTransactionEvent implements Runnable { return remoteCacheEvent; } + public RemoteTableMod getRemoteTableMod() { + return remoteTableMod; + } + } diff --git a/src/test/java/io/ebeaninternal/server/cluster/binarymessage/MessageReadWriteTest.java b/src/test/java/io/ebeaninternal/server/cluster/binarymessage/MessageReadWriteTest.java new file mode 100644 index 000000000..cf5eeee87 --- /dev/null +++ b/src/test/java/io/ebeaninternal/server/cluster/binarymessage/MessageReadWriteTest.java @@ -0,0 +1,118 @@ +package io.ebeaninternal.server.cluster.binarymessage; + +import io.ebean.BaseTestCase; +import io.ebean.EbeanServer; +import io.ebeaninternal.api.TDSpiEbeanServer; +import io.ebeaninternal.api.TransactionEventTable; +import io.ebeaninternal.server.cache.RemoteCacheEvent; +import io.ebeaninternal.server.cluster.MessageServerProvider; +import io.ebeaninternal.server.core.PersistRequest; +import io.ebeaninternal.server.deploy.BeanDescriptor; +import io.ebeaninternal.server.transaction.BeanPersistIds; +import io.ebeaninternal.server.transaction.RemoteTableMod; +import io.ebeaninternal.server.transaction.RemoteTransactionEvent; +import org.junit.Test; +import org.tests.model.basic.Customer; + +import java.io.IOException; +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; + +import static org.assertj.core.api.Assertions.assertThat; + +public class MessageReadWriteTest extends BaseTestCase { + + private BeanDescriptor customerBeanDescriptor = getBeanDescriptor(Customer.class); + + private TDEbeanServer mockEbeanServer = new TDEbeanServer(); + + private MessageReadWrite readWrite = new MessageReadWrite(new TDMessageServerProvider()); + + @Override + protected BeanDescriptor getBeanDescriptor(Class cls) { + return super.getBeanDescriptor(cls); + } + + @Test + public void readWrite() throws IOException { + + RemoteTransactionEvent event = new RemoteTransactionEvent("db"); + + long timestamp = System.currentTimeMillis(); + Set tables = new HashSet<>(); + Collections.addAll(tables, "one", "two", "three"); + + event.addRemoteTableMod(new RemoteTableMod(timestamp, tables)); + event.addRemoteCacheEvent(new RemoteCacheEvent(Customer.class)); + + event.addTableIUD(new TransactionEventTable.TableIUD("foo", true, false, true)); + event.addTableIUD(new TransactionEventTable.TableIUD("bar", false, true, false)); + + + BeanPersistIds beanPersistIds = new BeanPersistIds(customerBeanDescriptor); + beanPersistIds.addId(PersistRequest.Type.INSERT, 42); + beanPersistIds.addId(PersistRequest.Type.INSERT, 43); + beanPersistIds.addId(PersistRequest.Type.UPDATE, 55); + beanPersistIds.addId(PersistRequest.Type.DELETE, 66); + + event.addBeanPersistIds(beanPersistIds); + + byte[] binaryMessage = readWrite.write(event); + + RemoteTransactionEvent read = readWrite.read(binaryMessage); + + // cache event + RemoteCacheEvent remoteCacheEvent = read.getRemoteCacheEvent(); + assertThat(remoteCacheEvent.getClearCaches()).containsOnly(Customer.class.getName()); + + // table mod + RemoteTableMod remoteTableMod = read.getRemoteTableMod(); + assertThat(remoteTableMod.getTimestamp()).isEqualTo(timestamp); + assertThat(remoteTableMod.getTables()).isEqualTo(tables); + + // Bean persist ids + List beanPersistList = read.getBeanPersistList(); + assertThat(beanPersistList).hasSize(3); + assertThat(beanPersistList.get(0).getInsertIds()).containsOnly(42, 43); + assertThat(beanPersistList.get(1).getUpdateIds()).containsOnly(55); + assertThat(beanPersistList.get(2).getDeleteIds()).containsOnly(66); + } + + @Test + public void readWrite_RemoteTableMod() throws IOException { + + RemoteTransactionEvent event = new RemoteTransactionEvent("db"); + + long timestamp = System.currentTimeMillis(); + Set tables = new HashSet<>(); + Collections.addAll(tables, "one", "two", "three"); + + event.addRemoteTableMod(new RemoteTableMod(timestamp, tables)); + + byte[] binaryMessage = readWrite.write(event); + + RemoteTransactionEvent read = readWrite.read(binaryMessage); + + RemoteTableMod remoteTableMod = read.getRemoteTableMod(); + + assertThat(remoteTableMod.getTimestamp()).isEqualTo(timestamp); + assertThat(remoteTableMod.getTables()).isEqualTo(tables); + } + + class TDMessageServerProvider implements MessageServerProvider { + + @Override + public EbeanServer getServer(String name) { + return mockEbeanServer; + } + } + + class TDEbeanServer extends TDSpiEbeanServer { + @Override + public BeanDescriptor getBeanDescriptorById(String descriptorId) { + return customerBeanDescriptor; + } + } +}