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; } }