[jbosscache-commits] JBoss Cache SVN: r7513 - in core/branches/flat/src: main/java/org/horizon/cluster and 28 other directories.

Author: manik.surtani(a)jboss.com Date: 2009-01-19 14:02:51 -0500 (Mon, 19 Jan 2009) New Revision: 7513 Added: core/branches/flat/src/main/java/org/horizon/factories/KnownComponentNames.java core/branches/flat/src/main/java/org/horizon/factories/annotations/ComponentName.java core/branches/flat/src/main/java/org/horizon/manager/CacheManager.java core/branches/flat/src/main/java/org/horizon/manager/DefaultCacheManager.java core/branches/flat/src/main/java/org/horizon/notifications/CacheManagerNotifier.java core/branches/flat/src/main/java/org/horizon/notifications/CacheManagerNotifierImpl.java core/branches/flat/src/main/java/org/horizon/notifications/CacheNotifier.java core/branches/flat/src/main/java/org/horizon/notifications/CacheNotifierImpl.java core/branches/flat/src/main/java/org/horizon/notifications/Listenable.java core/branches/flat/src/main/java/org/horizon/notifications/Notifier.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/Listener.java core/branches/flat/src/main/java/org/horizon/remoting/InboundInvocationHandler.java core/branches/flat/src/main/java/org/horizon/remoting/InboundInvocationHandlerImpl.java core/branches/flat/src/main/java/org/horizon/remoting/RPCManagerImpl.java core/branches/flat/src/main/java/org/horizon/remoting/ResponseFilter.java core/branches/flat/src/main/java/org/horizon/remoting/ResponseMode.java core/branches/flat/src/main/java/org/horizon/remoting/transport/Transport.java core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/ core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/CommandAwareRpcDispatcher.java core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsAddress.java core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsResponseFilterAdapter.java core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsTransport.java Removed: core/branches/flat/src/main/java/org/horizon/config/NonOverridable.java core/branches/flat/src/main/java/org/horizon/config/OldFileFormatException.java core/branches/flat/src/main/java/org/horizon/manager/CacheManager.java core/branches/flat/src/main/java/org/horizon/marshall/CommandAwareRpcDispatcher.java core/branches/flat/src/main/java/org/horizon/notifications/Notifier.java core/branches/flat/src/main/java/org/horizon/notifications/NotifierImpl.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheListener.java core/branches/flat/src/main/java/org/horizon/remoting/ChannelMessageListener.java core/branches/flat/src/main/java/org/horizon/remoting/RPCManagerImpl.java core/branches/flat/src/main/java/org/horizon/remoting/transport/Transport.java Modified: core/branches/flat/src/main/java/org/horizon/Cache.java core/branches/flat/src/main/java/org/horizon/CacheDelegate.java core/branches/flat/src/main/java/org/horizon/CacheSPI.java core/branches/flat/src/main/java/org/horizon/cluster/ReplicationQueue.java core/branches/flat/src/main/java/org/horizon/commands/CommandsFactory.java core/branches/flat/src/main/java/org/horizon/commands/CommandsFactoryImpl.java core/branches/flat/src/main/java/org/horizon/commands/tx/PrepareCommand.java core/branches/flat/src/main/java/org/horizon/config/Configuration.java core/branches/flat/src/main/java/org/horizon/config/ConfigurationComponent.java core/branches/flat/src/main/java/org/horizon/config/GlobalConfiguration.java core/branches/flat/src/main/java/org/horizon/config/parsing/XmlConfigurationParserJBC3.java core/branches/flat/src/main/java/org/horizon/eviction/EvictionWatcher.java core/branches/flat/src/main/java/org/horizon/factories/AbstractComponentRegistry.java core/branches/flat/src/main/java/org/horizon/factories/ComponentRegistry.java core/branches/flat/src/main/java/org/horizon/factories/DefaultCacheFactory.java core/branches/flat/src/main/java/org/horizon/factories/EmptyConstructorFactory.java core/branches/flat/src/main/java/org/horizon/factories/GlobalComponentRegistry.java core/branches/flat/src/main/java/org/horizon/interceptors/InvalidationInterceptor.java core/branches/flat/src/main/java/org/horizon/interceptors/InvocationContextInterceptor.java core/branches/flat/src/main/java/org/horizon/interceptors/ReplicationInterceptor.java core/branches/flat/src/main/java/org/horizon/interceptors/TxInterceptor.java core/branches/flat/src/main/java/org/horizon/interceptors/base/BaseRpcInterceptor.java core/branches/flat/src/main/java/org/horizon/jmx/JmxRegistrationManager.java core/branches/flat/src/main/java/org/horizon/loader/ClusteredCacheLoader.java core/branches/flat/src/main/java/org/horizon/loader/SingletonStoreCacheLoader.java core/branches/flat/src/main/java/org/horizon/marshall/CacheMarshallerStarobrno.java core/branches/flat/src/main/java/org/horizon/notifications/EventLog.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryActivated.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryCreated.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryEvicted.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryInvalidated.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryLoaded.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryModified.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryPassivated.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryRemoved.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryVisited.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheStarted.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheStopped.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/TransactionCompleted.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/TransactionRegistered.java core/branches/flat/src/main/java/org/horizon/notifications/annotation/ViewChanged.java core/branches/flat/src/main/java/org/horizon/notifications/event/EventImpl.java core/branches/flat/src/main/java/org/horizon/notifications/event/ViewChangedEvent.java core/branches/flat/src/main/java/org/horizon/remoting/RPCManager.java core/branches/flat/src/main/java/org/horizon/transaction/GlobalTransaction.java core/branches/flat/src/main/java/org/horizon/transaction/TransactionTable.java core/branches/flat/src/main/java/org/horizon/tree/TreeCache.java core/branches/flat/src/main/java/org/horizon/tree/TreeCacheImpl.java core/branches/flat/src/main/java/org/horizon/util/TestingUtil.java core/branches/flat/src/main/java/org/horizon/util/TypedProperties.java core/branches/flat/src/main/java/org/horizon/util/internals/ViewChangeListener.java core/branches/flat/src/test/java/org/horizon/BasicTest.java core/branches/flat/src/test/java/org/horizon/UnitTestCacheFactory.java core/branches/flat/src/test/java/org/horizon/api/CacheAPITest.java core/branches/flat/src/test/java/org/horizon/api/mvcc/PutForExternalReadTest.java core/branches/flat/src/test/java/org/horizon/api/tree/SyncReplTxTest.java core/branches/flat/src/test/java/org/horizon/api/tree/TreeCacheAPITest.java core/branches/flat/src/test/java/org/horizon/replication/AsyncReplTest.java core/branches/flat/src/test/java/org/horizon/replication/ExceptionTest.java core/branches/flat/src/test/java/org/horizon/replication/SyncCacheListenerTest.java core/branches/flat/src/test/java/org/horizon/replication/SyncReplTest.java Log: Well, it compiles ... Modified: core/branches/flat/src/main/java/org/horizon/Cache.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/Cache.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/Cache.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -25,17 +25,16 @@ import org.horizon.context.InvocationContext; import org.horizon.interceptors.base.CommandInterceptor; import org.horizon.lifecycle.Lifecycle; -import org.jgroups.Address; +import org.horizon.manager.CacheManager; +import org.horizon.notifications.Listenable; -import java.util.List; -import java.util.Set; import java.util.concurrent.ConcurrentMap; /** * @author Mircea.Markus(a)jboss.com * @since 1.0 */ -public interface Cache<K, V> extends ConcurrentMap<K, V>, Lifecycle { +public interface Cache<K, V> extends ConcurrentMap<K, V>, Lifecycle, Listenable { /** * Under special operating behavior, associates the value with the specified key. <ul> <li> Only goes through if the * key specified does not exist; no-op otherwise (similar to {@link java.util.concurrent.ConcurrentMap#putIfAbsent(Object, @@ -65,12 +64,6 @@ Configuration getConfiguration(); - void addCacheListener(Object listener); - - void removeCacheListener(Object listener); - - Set<Object> getCacheListeners(); - InvocationContext getInvocationContext(); void setInvocationContext(InvocationContext ctx); @@ -84,12 +77,8 @@ public void endBatch(boolean successful); - List<Address> getMembers(); - String getName(); - Address getLocalAddress(); - String getVersion(); void addInterceptor(CommandInterceptor i, int position); @@ -99,4 +88,6 @@ void removeInterceptor(int position); void removeInterceptor(Class<? extends CommandInterceptor> interceptorType); + + CacheManager getCacheManager(); } Modified: core/branches/flat/src/main/java/org/horizon/CacheDelegate.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/CacheDelegate.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/CacheDelegate.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -48,13 +48,13 @@ import org.horizon.lock.LockManager; import org.horizon.logging.Log; import org.horizon.logging.LogFactory; +import org.horizon.manager.CacheManager; import org.horizon.marshall.Marshaller; -import org.horizon.notifications.Notifier; +import org.horizon.notifications.CacheNotifier; import org.horizon.remoting.RPCManager; import org.horizon.statetransfer.StateTransferManager; import org.horizon.transaction.GlobalTransaction; import org.horizon.transaction.TransactionTable; -import org.jgroups.Address; import javax.transaction.Transaction; import javax.transaction.TransactionManager; @@ -73,7 +73,7 @@ protected CommandsFactory commandsFactory; protected InterceptorChain invoker; protected Configuration config; - protected Notifier notifier; + protected CacheNotifier notifier; protected BatchContainer batchContainer; protected ComponentRegistry componentRegistry; protected TransactionManager transactionManager; @@ -84,6 +84,7 @@ private DataContainer dataContainer; private LockManager lockManager; private static final Log log = LogFactory.getLog(CacheDelegate.class); + private CacheManager cacheManager; public CacheDelegate(String name) { this.name = name; @@ -95,13 +96,14 @@ CommandsFactory commandsFactory, InterceptorChain interceptorChain, Configuration configuration, - Notifier notifier, + CacheNotifier notifier, ComponentRegistry componentRegistry, TransactionManager transactionManager, BatchContainer batchContainer, RPCManager rpcManager, DataContainer dataContainer, Marshaller marshaller, - LockManager lockManager) { + LockManager lockManager, + CacheManager cacheManager) { this.invocationContextContainer = invocationContextContainer; this.commandsFactory = commandsFactory; this.invoker = interceptorChain; @@ -115,6 +117,7 @@ this.dataContainer = dataContainer; this.marshaller = marshaller; this.lockManager = lockManager; + this.cacheManager = cacheManager; } public V putIfAbsent(K key, V value) { @@ -233,16 +236,16 @@ return config; } - public void addCacheListener(Object listener) { - notifier.addCacheListener(listener); + public void addListener(Object listener) { + notifier.addListener(listener); } - public void removeCacheListener(Object listener) { - notifier.removeCacheListener(listener); + public void removeListener(Object listener) { + notifier.removeListener(listener); } - public Set getCacheListeners() { - return notifier.getCacheListeners(); + public Set<Object> getListeners() { + return notifier.getListeners(); } public InvocationContext getInvocationContext() { @@ -317,14 +320,10 @@ throw new IllegalStateException();//todo Implement me properly } - public Notifier getNotifier() { + public CacheNotifier getNotifier() { return notifier; } - public String getClusterName() { - return config.getClusterName(); - } - public GlobalTransaction getCurrentTransaction(Transaction tx, boolean createIfNotExists) { throw new IllegalStateException();//todo Implement me properly } @@ -353,10 +352,6 @@ batchContainer.endBatch(successful); } - public List<Address> getMembers() { - return rpcManager.getMembers(); - } - public Object getDirect(Object key) { throw new UnsupportedOperationException("Not implemented");//todo please implement! } @@ -373,10 +368,6 @@ return Version.decodeVersion(Version.getVersionShort()); } - public Address getLocalAddress() { - return rpcManager.getLocalAddress(); - } - public void setName(String name) { this.name = name; } @@ -399,4 +390,8 @@ public BatchContainer getBatchContainer() { return batchContainer; } + + public CacheManager getCacheManager() { + return cacheManager; + } } Modified: core/branches/flat/src/main/java/org/horizon/CacheSPI.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/CacheSPI.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/CacheSPI.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -29,7 +29,7 @@ import org.horizon.loader.CacheLoaderManager; import org.horizon.lock.LockManager; import org.horizon.marshall.Marshaller; -import org.horizon.notifications.Notifier; +import org.horizon.notifications.CacheNotifier; import org.horizon.remoting.RPCManager; import org.horizon.statetransfer.StateTransferManager; import org.horizon.transaction.GlobalTransaction; @@ -141,17 +141,12 @@ * From 2.1.0, Interceptor authors should obtain this by injection rather than this method. See the {@link * org.horizon.factories.annotations.Inject} annotation. * - * @return the notifier attached with this instance of the cache. See {@link org.horizon.notifications.Notifier}, a - * class that is responsible for emitting notifications to registered CacheListeners. + * @return the notifier attached with this instance of the cache. See {@link org.horizon.notifications.CacheNotifier}, + * a class that is responsible for emitting notifications to registered CacheListeners. */ - Notifier getNotifier(); + CacheNotifier getNotifier(); /** - * @return the name of the cluster. Null if running in local mode. - */ - String getClusterName(); - - /** * Returns the global transaction for this local transaction. Optionally creates a new global transaction if it does * not exist. * Modified: core/branches/flat/src/main/java/org/horizon/cluster/ReplicationQueue.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/cluster/ReplicationQueue.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/cluster/ReplicationQueue.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -31,6 +31,7 @@ import org.horizon.logging.Log; import org.horizon.logging.LogFactory; import org.horizon.remoting.RPCManager; +import org.horizon.remoting.ResponseMode; import java.util.ArrayList; import java.util.LinkedList; @@ -154,7 +155,7 @@ ReplicateCommand replicateCommand = commandsFactory.buildReplicateCommand(toReplicate); // send to all live nodes in the cluster - rpcManager.callRemoteMethods(null, replicateCommand, false, configuration.getSyncReplTimeout(), false); + rpcManager.invokeRemotely(null, replicateCommand, ResponseMode.ASYNCHRONOUS, configuration.getSyncReplTimeout()); } catch (Throwable t) { log.error("failed replicating " + toReplicate.size() + " elements in replication queue", t); Modified: core/branches/flat/src/main/java/org/horizon/commands/CommandsFactory.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/commands/CommandsFactory.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/commands/CommandsFactory.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -36,8 +36,8 @@ import org.horizon.commands.write.ReplaceCommand; import org.horizon.factories.scopes.Scope; import org.horizon.factories.scopes.Scopes; +import org.horizon.remoting.transport.Address; import org.horizon.transaction.GlobalTransaction; -import org.jgroups.Address; import java.util.List; import java.util.Map; Modified: core/branches/flat/src/main/java/org/horizon/commands/CommandsFactoryImpl.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/commands/CommandsFactoryImpl.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/commands/CommandsFactoryImpl.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -38,9 +38,9 @@ import org.horizon.container.DataContainer; import org.horizon.factories.annotations.Inject; import org.horizon.interceptors.InterceptorChain; -import org.horizon.notifications.Notifier; +import org.horizon.notifications.CacheNotifier; +import org.horizon.remoting.transport.Address; import org.horizon.transaction.GlobalTransaction; -import org.jgroups.Address; import java.util.List; import java.util.Map; @@ -51,11 +51,11 @@ */ public class CommandsFactoryImpl implements CommandsFactory { private DataContainer dataContainer; - private Notifier notifier; + private CacheNotifier notifier; private InterceptorChain interceptorChain; @Inject - public void setupDependencies(DataContainer container, Notifier notifier, InterceptorChain interceptorChain) { + public void setupDependencies(DataContainer container, CacheNotifier notifier, InterceptorChain interceptorChain) { this.dataContainer = container; this.notifier = notifier; this.interceptorChain = interceptorChain; Modified: core/branches/flat/src/main/java/org/horizon/commands/tx/PrepareCommand.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/commands/tx/PrepareCommand.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/commands/tx/PrepareCommand.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -26,8 +26,8 @@ import org.horizon.commands.VisitableCommand; import org.horizon.commands.Visitor; import org.horizon.context.InvocationContext; +import org.horizon.remoting.transport.Address; import org.horizon.transaction.GlobalTransaction; -import org.jgroups.Address; import java.util.ArrayList; import java.util.Collection; Modified: core/branches/flat/src/main/java/org/horizon/config/Configuration.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/config/Configuration.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/config/Configuration.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -21,21 +21,14 @@ */ package org.horizon.config; -import org.horizon.CacheException; -import org.horizon.Version; -import org.horizon.config.parsing.JGroupsStackParser; import org.horizon.factories.annotations.NonVolatile; import org.horizon.factories.annotations.Start; import org.horizon.lock.IsolationLevel; -import org.horizon.marshall.Marshaller; import org.horizon.util.ReflectionUtil; -import org.w3c.dom.Element; -import java.net.URL; import java.util.Collections; import java.util.HashMap; import java.util.List; -import java.util.Locale; import java.util.Map; /** @@ -47,10 +40,6 @@ @NonVolatile public class Configuration extends ConfigurationComponent { private static final long serialVersionUID = 5553791890144997466L; - - private Marshaller marshaller; - - private transient JGroupsStackParser jGroupsStackParser = new JGroupsStackParser(); private boolean invocationBatchingEnabled; private Map<String, EvictionCacheConfig> evictionCacheConfigs = new HashMap<String, EvictionCacheConfig>(4); @@ -64,24 +53,6 @@ } /** - * Behavior of the JVM shutdown hook registered by the cache - */ - public static enum ShutdownHookBehavior { - /** - * By default a shutdown hook is registered if no MBean server (apart from the JDK default) is detected. - */ - DEFAULT, - /** - * Forces the cache to register a shutdown hook even if an MBean server is detected. - */ - REGISTER, - /** - * Forces the cache NOT to register a shutdown hook, even if no MBean server is detected. - */ - DONT_REGISTER - } - - /** * Cache replication mode. */ public static enum CacheMode { @@ -123,37 +94,11 @@ } - public static CacheMode legacyModeToCacheMode(int legacyMode) { - switch (legacyMode) { - case 1: - return CacheMode.LOCAL; - case 2: - return CacheMode.REPL_ASYNC; - case 3: - return CacheMode.REPL_SYNC; - case 4: - return CacheMode.INVALIDATION_ASYNC; - case 5: - return CacheMode.INVALIDATION_SYNC; - default: - throw new IllegalArgumentException("Unknown legacy cache mode " + - legacyMode); - } - } - - /** - * Default replication version, from {@link Version#getVersionShort}. - */ - public static final short DEFAULT_REPLICATION_VERSION = Version.getVersionShort(); - // ------------------------------------------------------------------------------------------------------------ // CONFIGURATION OPTIONS // ------------------------------------------------------------------------------------------------------------ - @NonOverridable - private String clusterName = "JBossCache-Cluster"; - @NonOverridable - private String clusterConfig = null; + private boolean useReplQueue = false; @Dynamic private int replQueueMaxElements = 1000; @@ -162,7 +107,6 @@ private boolean exposeManagementStatistics = true; @Dynamic private boolean fetchInMemoryState = true; - private short replicationVersion = DEFAULT_REPLICATION_VERSION; @Dynamic private long lockAcquisitionTimeout = 10000; @Dynamic @@ -176,7 +120,6 @@ private boolean lockParentForChildInsertRemove = false; @Dynamic private EvictionConfig evictionConfig = null; - private boolean useRegionBasedMarshalling = false; private String transactionManagerLookupClass = null; private CacheLoaderConfig cacheLoaderConfig = null; @Dynamic @@ -184,23 +127,11 @@ @Dynamic private boolean syncRollbackPhase = false; - private String muxStackName = null; - private boolean usingMultiplexer = false; private transient RuntimeConfig runtimeConfig; - private String marshallerClass; - private ShutdownHookBehavior shutdownHookBehavior = ShutdownHookBehavior.DEFAULT; private boolean useLazyDeserialization = false; - private int objectInputStreamPoolSize = 50; - private int objectOutputStreamPoolSize = 50; private List<CustomInterceptorConfig> customInterceptors = Collections.emptyList(); private boolean writeSkewCheck = false; private int concurrencyLevel = 500; - private int listenerAsyncPoolSize = 1; - private int listenerAsyncQueueSize = 50000; - private int serializationExecutorPoolSize = 0; - private int serializationExecutorQueueSize = 50000; - @NonOverridable - private URL jgroupsConfigFile; @Start(priority = 1) private void correctIsolationLevels() { @@ -220,15 +151,6 @@ // SETTERS - MAKE SURE ALL SETTERS PERFORM testImmutability()!!! // ------------------------------------------------------------------------------------------------------------ - public void setCacheMarshaller(Marshaller instance) { - testImmutability("marshaller"); - marshaller = instance; - } - - public Marshaller getMarshaller() { - return marshaller; - } - public boolean isWriteSkewCheck() { return writeSkewCheck; } @@ -247,23 +169,6 @@ this.concurrencyLevel = concurrencyLevel; } - /** - * Converts a list of elements to a Java Groups property string. - */ - public void setClusterConfig(Element config) { - setClusterConfig(jGroupsStackParser.parseClusterConfigXml(config)); - } - - public void setClusterName(String clusterName) { - testImmutability("clusterName"); - this.clusterName = clusterName; - } - - public void setClusterConfig(String clusterConfig) { - testImmutability("clusterConfig"); - this.clusterConfig = clusterConfig; - } - public void setReplQueueMaxElements(int replQueueMaxElements) { testImmutability("replQueueMaxElements"); this.replQueueMaxElements = replQueueMaxElements; @@ -296,15 +201,6 @@ this.fetchInMemoryState = fetchInMemoryState; } - public void setReplicationVersion(short replicationVersion) { - testImmutability("replicationVersion"); - this.replicationVersion = replicationVersion; - } - - public void setReplVersionString(String replVersionString) { - setReplicationVersion(replVersionString == null ? 0 : Version.getVersionShort(replVersionString)); - } - public void setLockAcquisitionTimeout(long lockAcquisitionTimeout) { testImmutability("lockAcquisitionTimeout"); this.lockAcquisitionTimeout = lockAcquisitionTimeout; @@ -352,19 +248,6 @@ this.evictionConfig = config; } - /** - * This is a deprecated configuration option. While it will be supported for the 2.x series for backward - * compatibility, expect to see it disappear in 3.x. - * <p/> - * With {@link #isUseLazyDeserialization()}, which is enabled by default, custom class loaders are handled - * implicitly. See the user guide for details on how this is handled. - * <p/> - */ - public void setUseRegionBasedMarshalling(boolean useRegionBasedMarshalling) { - testImmutability("useRegionBasedMarshalling"); - this.useRegionBasedMarshalling = useRegionBasedMarshalling; - } - public void setTransactionManagerLookupClass(String transactionManagerLookupClass) { testImmutability("transactionManagerLookupClass"); this.transactionManagerLookupClass = transactionManagerLookupClass; @@ -386,19 +269,6 @@ this.syncRollbackPhase = syncRollbackPhase; } - /** - * Sets the size of the asynchronous listener notification thread pool size. Defaults to 1, and if set to below 1, - * all async listeners (specified with {@link org.horizon.notifications.annotation.CacheListener#sync()} are notified - * synchronously. - * - * @param listenerAsyncPoolSize number of threads in pool - * @since 1.0 - */ - public void setListenerAsyncPoolSize(int listenerAsyncPoolSize) { - testImmutability("listenerAsyncPoolSize"); - this.listenerAsyncPoolSize = listenerAsyncPoolSize; - } - public void setUseReplQueue(boolean useReplQueue) { testImmutability("useReplQueue"); this.useReplQueue = useReplQueue; @@ -414,10 +284,6 @@ this.stateRetrievalTimeout = stateRetrievalTimeout; } - private static String uc(String s) { - return s.toUpperCase(Locale.ENGLISH); - } - public void setIsolationLevel(String isolationLevel) { testImmutability("isolationLevel"); if (isolationLevel == null) throw new ConfigurationException("Isolation level cannot be null", "IsolationLevel"); @@ -428,14 +294,6 @@ } } - public String getIsolationLevelString() { - return isolationLevel == null ? null : isolationLevel.toString(); - } - - public void setIsolationLevelString(String isolationLevel) { - setIsolationLevel(isolationLevel); - } - /** * Sets whether inserting or removing a node requires a write lock on the node's parent (when pessimistic locking is * used.) @@ -447,110 +305,20 @@ this.lockParentForChildInsertRemove = lockParentForChildInsertRemove; } - public void setMultiplexerStack(String stackName) { - testImmutability("muxStackName"); - this.muxStackName = stackName; - } - - public boolean isUsingMultiplexer() { - return usingMultiplexer; - } - - public void setUsingMultiplexer(boolean usingMultiplexer) { - testImmutability("usingMultiplexer"); - this.usingMultiplexer = usingMultiplexer; - } - - public void setShutdownHookBehavior(ShutdownHookBehavior shutdownHookBehavior) { - testImmutability("shutdownHookBehavior"); - this.shutdownHookBehavior = shutdownHookBehavior; - } - - public void setShutdownHookBehavior(String shutdownHookBehavior) { - testImmutability("shutdownHookBehavior"); - if (shutdownHookBehavior == null) - throw new ConfigurationException("Shutdown hook behavior cannot be null", "ShutdownHookBehavior"); - this.shutdownHookBehavior = ShutdownHookBehavior.valueOf(uc(shutdownHookBehavior)); - if (this.shutdownHookBehavior == null) { - log.warn("Unknown shutdown hook behavior '" + shutdownHookBehavior + "', using defaults."); - this.shutdownHookBehavior = ShutdownHookBehavior.DEFAULT; - } - } - public void setUseLazyDeserialization(boolean useLazyDeserialization) { testImmutability("useLazyDeserialization"); this.useLazyDeserialization = useLazyDeserialization; } - /** - * Initialises the size of the object input stream pool size, which defaults to 50. - * - * @param objectInputStreamPoolSize - * @since 1.0 - */ - public void setObjectInputStreamPoolSize(int objectInputStreamPoolSize) { - testImmutability("objectInputStreamPoolSize"); - this.objectInputStreamPoolSize = objectInputStreamPoolSize; - } - - /** - * Initialises the size of the object output stream pool size, which defaults to 50. - * - * @param objectOutputStreamPoolSize - * @since 1.0 - */ - public void setObjectOutputStreamPoolSize(int objectOutputStreamPoolSize) { - testImmutability("objectOutputStreamPoolSize"); - this.objectOutputStreamPoolSize = objectOutputStreamPoolSize; - } - - /** - * Sets the async replication serialization executor pool size for async replication. Has no effect if the - * replication queue is used. - * - * @param serializationExecutorPoolSize number of threads to use - */ - public void setSerializationExecutorPoolSize(int serializationExecutorPoolSize) { - testImmutability("serializationExecutorPoolSize"); - this.serializationExecutorPoolSize = serializationExecutorPoolSize; - } - - public void setListenerAsyncQueueSize(int listenerAsyncQueueSize) { - testImmutability("listenerAsyncQueueSize"); - this.listenerAsyncQueueSize = listenerAsyncQueueSize; - } - - public void setSerializationExecutorQueueSize(int serializationExecutorQueueSize) { - testImmutability("serializationExecutorQueueSize"); - this.serializationExecutorQueueSize = serializationExecutorQueueSize; - } - - public void setJgroupsConfigFile(URL jgroupsConfigFile) { - testImmutability("jgroupsConfigFile"); - this.jgroupsConfigFile = jgroupsConfigFile; - } - // ------------------------------------------------------------------------------------------------------------ // GETTERS // ------------------------------------------------------------------------------------------------------------ - public ShutdownHookBehavior getShutdownHookBehavior() { - return this.shutdownHookBehavior; - } - public boolean isUseReplQueue() { return useReplQueue; } - public String getClusterName() { - return clusterName; - } - - public String getClusterConfig() { - return clusterConfig; - } - public int getReplQueueMaxElements() { return replQueueMaxElements; } @@ -575,14 +343,6 @@ return fetchInMemoryState; } - public short getReplicationVersion() { - return replicationVersion; - } - - public String getReplVersionString() { - return Version.decodeVersion(replicationVersion); - } - public long getLockAcquisitionTimeout() { return lockAcquisitionTimeout; } @@ -613,10 +373,6 @@ return lockParentForChildInsertRemove; } - public boolean isUseRegionBasedMarshalling() { - return useRegionBasedMarshalling; - } - public String getTransactionManagerLookupClass() { return transactionManagerLookupClass; } @@ -633,26 +389,10 @@ return syncRollbackPhase; } - /** - * Gets the size of the asynchronous listener notification thread pool size. Defaults to 1, and if set to below 1, - * all async listeners (specified with {@link org.horizon.notifications.annotation.CacheListener#sync()} are notified - * synchronously. - * - * @return thread pool size - * @since 1.0 - */ - public int getListenerAsyncPoolSize() { - return listenerAsyncPoolSize; - } - public long getStateRetrievalTimeout() { return stateRetrievalTimeout; } - public String getMultiplexerStack() { - return muxStackName; - } - public boolean isUseLazyDeserialization() { return useLazyDeserialization; } @@ -675,61 +415,6 @@ this.runtimeConfig = runtimeConfig; } - public String getMarshallerClass() { - return marshallerClass; - } - - public void setMarshallerClass(String marshallerClass) { - this.marshallerClass = marshallerClass; - } - - /** - * @return the size of he object input stream pool - * @since 1.0 - */ - public int getObjectInputStreamPoolSize() { - return objectInputStreamPoolSize; - } - - /** - * @return the size of he object output stream pool - * @since 1.0 - */ - public int getObjectOutputStreamPoolSize() { - return objectOutputStreamPoolSize; - } - - - /** - * Returns a {@link java.net.URL} to a default JGroups configuration file. - * - * @return a default JGroups config file - */ - public URL getDefaultClusterConfig() { - URL url = getClass().getClassLoader().getResource("flush-udp.xml"); - if (log.isTraceEnabled()) log.trace("Using default JGroups configuration file " + url); - return url; - } - - /** - * @return the serialization executor pool size. - */ - public int getSerializationExecutorPoolSize() { - return serializationExecutorPoolSize; - } - - public int getListenerAsyncQueueSize() { - return listenerAsyncQueueSize; - } - - public int getSerializationExecutorQueueSize() { - return serializationExecutorQueueSize; - } - - public URL getJgroupsConfigFile() { - return jgroupsConfigFile; - } - // ------------------------------------------------------------------------------------------------------------ // HELPERS // ------------------------------------------------------------------------------------------------------------ @@ -750,56 +435,35 @@ if (inactiveOnStartup != that.inactiveOnStartup) return false; if (lockAcquisitionTimeout != that.lockAcquisitionTimeout) return false; if (lockParentForChildInsertRemove != that.lockParentForChildInsertRemove) return false; - if (objectInputStreamPoolSize != that.objectInputStreamPoolSize) return false; - if (objectOutputStreamPoolSize != that.objectOutputStreamPoolSize) return false; if (replQueueInterval != that.replQueueInterval) return false; if (replQueueMaxElements != that.replQueueMaxElements) return false; - if (replicationVersion != that.replicationVersion) return false; if (stateRetrievalTimeout != that.stateRetrievalTimeout) return false; if (syncCommitPhase != that.syncCommitPhase) return false; if (syncReplTimeout != that.syncReplTimeout) return false; if (syncRollbackPhase != that.syncRollbackPhase) return false; if (useLazyDeserialization != that.useLazyDeserialization) return false; - if (useRegionBasedMarshalling != that.useRegionBasedMarshalling) return false; if (useReplQueue != that.useReplQueue) return false; - if (usingMultiplexer != that.usingMultiplexer) return false; if (cacheLoaderConfig != null ? !cacheLoaderConfig.equals(that.cacheLoaderConfig) : that.cacheLoaderConfig != null) return false; if (cacheMode != that.cacheMode) return false; - if (clusterConfig != null ? !clusterConfig.equals(that.clusterConfig) : that.clusterConfig != null) return false; - if (clusterName != null ? !clusterName.equals(that.clusterName) : that.clusterName != null) return false; if (evictionConfig != null ? !evictionConfig.equals(that.evictionConfig) : that.evictionConfig != null) return false; if (isolationLevel != that.isolationLevel) return false; - if (marshaller != null ? !marshaller.equals(that.marshaller) : that.marshaller != null) return false; - if (marshallerClass != null ? !marshallerClass.equals(that.marshallerClass) : that.marshallerClass != null) - return false; - if (muxStackName != null ? !muxStackName.equals(that.muxStackName) : that.muxStackName != null) return false; if (runtimeConfig != null ? !runtimeConfig.equals(that.runtimeConfig) : that.runtimeConfig != null) return false; - if (shutdownHookBehavior != that.shutdownHookBehavior) return false; if (transactionManagerLookupClass != null ? !transactionManagerLookupClass.equals(that.transactionManagerLookupClass) : that.transactionManagerLookupClass != null) return false; - if (listenerAsyncPoolSize != that.listenerAsyncPoolSize) return false; - if (serializationExecutorPoolSize != that.serializationExecutorPoolSize) return false; - if (jgroupsConfigFile != that.jgroupsConfigFile) return false; - if (listenerAsyncQueueSize != that.listenerAsyncQueueSize) return false; - if (serializationExecutorQueueSize != that.serializationExecutorQueueSize) return false; return true; } @Override public int hashCode() { - int result; - result = (marshaller != null ? marshaller.hashCode() : 0); - result = 31 * result + (clusterName != null ? clusterName.hashCode() : 0); - result = 31 * result + (clusterConfig != null ? clusterConfig.hashCode() : 0); + int result = 0; result = 31 * result + (useReplQueue ? 1 : 0); result = 31 * result + replQueueMaxElements; result = 31 * result + (int) (replQueueInterval ^ (replQueueInterval >>> 32)); result = 31 * result + (exposeManagementStatistics ? 1 : 0); result = 31 * result + (fetchInMemoryState ? 1 : 0); - result = 31 * result + (int) replicationVersion; result = 31 * result + (int) (lockAcquisitionTimeout ^ (lockAcquisitionTimeout >>> 32)); result = 31 * result + (int) (syncReplTimeout ^ (syncReplTimeout >>> 32)); result = 31 * result + (cacheMode != null ? cacheMode.hashCode() : 0); @@ -808,24 +472,12 @@ result = 31 * result + (isolationLevel != null ? isolationLevel.hashCode() : 0); result = 31 * result + (lockParentForChildInsertRemove ? 1 : 0); result = 31 * result + (evictionConfig != null ? evictionConfig.hashCode() : 0); - result = 31 * result + (useRegionBasedMarshalling ? 1 : 0); result = 31 * result + (transactionManagerLookupClass != null ? transactionManagerLookupClass.hashCode() : 0); result = 31 * result + (cacheLoaderConfig != null ? cacheLoaderConfig.hashCode() : 0); result = 31 * result + (syncCommitPhase ? 1 : 0); result = 31 * result + (syncRollbackPhase ? 1 : 0); - result = 31 * result + (muxStackName != null ? muxStackName.hashCode() : 0); - result = 31 * result + (usingMultiplexer ? 1 : 0); result = 31 * result + (runtimeConfig != null ? runtimeConfig.hashCode() : 0); - result = 31 * result + (marshallerClass != null ? marshallerClass.hashCode() : 0); - result = 31 * result + (shutdownHookBehavior != null ? shutdownHookBehavior.hashCode() : 0); result = 31 * result + (useLazyDeserialization ? 1 : 0); - result = 31 * result + objectInputStreamPoolSize; - result = 31 * result + objectOutputStreamPoolSize; - result = 31 * result + serializationExecutorPoolSize; - result = 31 * result + listenerAsyncPoolSize; - result = 31 * result + serializationExecutorQueueSize; - result = 31 * result + listenerAsyncQueueSize; - result = 31 * result + (jgroupsConfigFile != null ? jgroupsConfigFile.hashCode() : 0); return result; } @@ -856,14 +508,6 @@ return getCacheLoaderConfig() != null && !getCacheLoaderConfig().getIndividualCacheLoaderConfigs().isEmpty(); } - public String getMuxStackName() { - return muxStackName; - } - - public void setMuxStackName(String muxStackName) { - this.muxStackName = muxStackName; - } - /** * Returns the {@link org.horizon.config.CustomInterceptorConfig}, if any, associated with this configuration object. * The custom interceptors will be added to the cache at startup in the sequence defined by this list. @@ -890,18 +534,7 @@ public void applyOverrides(Configuration overrides) { // loop through all overridden elements in the incoming configuration and apply for (String overriddenField : overrides.overriddenConfigurationElements) { - assertAllowedToOverride(overriddenField); ReflectionUtil.setValue(this, overriddenField, ReflectionUtil.getValue(overrides, overriddenField)); } } - - private void assertAllowedToOverride(String fieldname) { - try { - if (getClass().getDeclaredField(fieldname).isAnnotationPresent(NonOverridable.class)) - throw new ConfigurationException("Configuration field " + fieldname + " cannot be overridden!"); - } - catch (NoSuchFieldException e) { - throw new CacheException("Field " + fieldname + " does not exist on Configuration!"); - } - } } Modified: core/branches/flat/src/main/java/org/horizon/config/ConfigurationComponent.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/config/ConfigurationComponent.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/config/ConfigurationComponent.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -30,12 +30,18 @@ import org.horizon.factories.scopes.Scopes; import org.horizon.logging.Log; import org.horizon.logging.LogFactory; +import org.horizon.util.TypedProperties; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; import java.util.Collection; import java.util.Collections; import java.util.HashSet; import java.util.LinkedList; import java.util.List; +import java.util.Locale; +import java.util.Properties; import java.util.Set; /** @@ -104,6 +110,39 @@ } /** + * Safely converts a String to upper case. + * + * @param s string to convert + * @return the string in upper case, or null if s is null. + */ + protected String uc(String s) { + return s == null ? null : s.toUpperCase(Locale.ENGLISH); + } + + /** + * Converts a given {@link Properties} instance to an instance of {@link TypedProperties} + * + * @param p properties to convert + * @return TypedProperties instance + */ + protected TypedProperties toTypedProperties(Properties p) { + return TypedProperties.toTypedProperties(p); + } + + protected TypedProperties toTypedProperties(String s) { + TypedProperties tp = new TypedProperties(); + if (s != null && s.trim().length() > 0) { + InputStream stream = new ByteArrayInputStream(s.getBytes()); + try { + tp.load(stream); + } catch (IOException e) { + throw new ConfigurationException("Unable to parse properties string " + s, e); + } + } + return tp; + } + + /** * Checks field modifications via setters * * @param fieldName Modified: core/branches/flat/src/main/java/org/horizon/config/GlobalConfiguration.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/config/GlobalConfiguration.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/config/GlobalConfiguration.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,6 +1,8 @@ package org.horizon.config; import org.horizon.CacheException; +import org.horizon.Version; +import org.horizon.util.TypedProperties; import java.util.Properties; @@ -11,18 +13,51 @@ * @since 1.0 */ public class GlobalConfiguration extends ConfigurationComponent { + + /** + * Default replication version, from {@link org.horizon.Version#getVersionShort}. + */ + public static final short DEFAULT_REPLICATION_VERSION = Version.getVersionShort(); + String asyncListenerExecutorFactoryClass; + TypedProperties asyncListenerExecutorProperties; String asyncSerializationExecutorFactoryClass; + TypedProperties asyncSerializationExecutorProperties; String evictionScheduledExecutorFactoryClass; + TypedProperties evictionScheduledExecutorProperties; String replicationQueueScheduledExecutorFactoryClass; + TypedProperties replicationQueueScheduledExecutorProperties; String marshallerClass; String marshallVersion; int objectInputStreamPoolSize; int objectOutputStreamPoolSize; String transportClass; - Properties transportProperties; + TypedProperties transportProperties; Configuration defaultConfiguration; + String clusterName = "Horizon-Cluster"; + ShutdownHookBehavior shutdownHookBehavior = ShutdownHookBehavior.DEFAULT; + short replicationVersion = DEFAULT_REPLICATION_VERSION; + + /** + * Behavior of the JVM shutdown hook registered by the cache + */ + public static enum ShutdownHookBehavior { + /** + * By default a shutdown hook is registered if no MBean server (apart from the JDK default) is detected. + */ + DEFAULT, + /** + * Forces the cache to register a shutdown hook even if an MBean server is detected. + */ + REGISTER, + /** + * Forces the cache NOT to register a shutdown hook, even if no MBean server is detected. + */ + DONT_REGISTER + } + + public String getAsyncListenerExecutorFactoryClass() { return asyncListenerExecutorFactoryClass; } @@ -110,9 +145,14 @@ public void setTransportProperties(Properties transportProperties) { testImmutability("transportProperties"); - this.transportProperties = transportProperties; + this.transportProperties = toTypedProperties(transportProperties); } + public void setTransportProperties(String transportPropertiesString) { + testImmutability("transportProperties"); + this.transportProperties = toTypedProperties(transportPropertiesString); + } + public Configuration getDefaultConfiguration() { return defaultConfiguration; } @@ -122,6 +162,109 @@ this.defaultConfiguration = defaultConfiguration; } + public String getClusterName() { + return clusterName; + } + + public void setClusterName(String clusterName) { + testImmutability("clusterName"); + this.clusterName = clusterName; + } + + public ShutdownHookBehavior getShutdownHookBehavior() { + return shutdownHookBehavior; + } + + public void setShutdownHookBehavior(ShutdownHookBehavior shutdownHookBehavior) { + testImmutability("shutdownHookBehavior"); + this.shutdownHookBehavior = shutdownHookBehavior; + } + + public void setShutdownHookBehavior(String shutdownHookBehavior) { + if (shutdownHookBehavior == null) + throw new ConfigurationException("Shutdown hook behavior cannot be null", "ShutdownHookBehavior"); + ShutdownHookBehavior temp = ShutdownHookBehavior.valueOf(uc(shutdownHookBehavior)); + if (temp == null) { + log.warn("Unknown shutdown hook behavior '" + shutdownHookBehavior + "', using defaults."); + temp = ShutdownHookBehavior.DEFAULT; + } + setShutdownHookBehavior(temp); + } + + public Properties getAsyncListenerExecutorProperties() { + return asyncListenerExecutorProperties; + } + + public void setAsyncListenerExecutorProperties(Properties asyncListenerExecutorProperties) { + testImmutability("asyncListenerExecutorProperties"); + this.asyncListenerExecutorProperties = toTypedProperties(asyncListenerExecutorProperties); + } + + public void setAsyncListenerExecutorProperties(String asyncListenerExecutorPropertiesString) { + testImmutability("asyncListenerExecutorProperties"); + this.asyncListenerExecutorProperties = toTypedProperties(asyncListenerExecutorPropertiesString); + } + + public Properties getAsyncSerializationExecutorProperties() { + return asyncSerializationExecutorProperties; + } + + public void setAsyncSerializationExecutorProperties(Properties asyncSerializationExecutorProperties) { + testImmutability("asyncSerializationExecutorProperties"); + this.asyncSerializationExecutorProperties = toTypedProperties(asyncSerializationExecutorProperties); + } + + public void setAsyncSerializationExecutorProperties(String asyncSerializationExecutorPropertiesString) { + testImmutability("asyncSerializationExecutorProperties"); + this.asyncSerializationExecutorProperties = toTypedProperties(asyncSerializationExecutorPropertiesString); + } + + public Properties getEvictionScheduledExecutorProperties() { + return evictionScheduledExecutorProperties; + } + + public void setEvictionScheduledExecutorProperties(Properties evictionScheduledExecutorProperties) { + testImmutability("evictionScheduledExecutorProperties"); + this.evictionScheduledExecutorProperties = toTypedProperties(evictionScheduledExecutorProperties); + } + + public void setEvictionScheduledExecutorProperties(String evictionScheduledExecutorPropertiesString) { + testImmutability("evictionScheduledExecutorProperties"); + this.evictionScheduledExecutorProperties = toTypedProperties(evictionScheduledExecutorPropertiesString); + } + + public Properties getReplicationQueueScheduledExecutorProperties() { + return replicationQueueScheduledExecutorProperties; + } + + public void setReplicationQueueScheduledExecutorProperties(Properties replicationQueueScheduledExecutorProperties) { + testImmutability("replicationQueueScheduledExecutorProperties"); + this.replicationQueueScheduledExecutorProperties = toTypedProperties(replicationQueueScheduledExecutorProperties); + } + + public void setReplicationQueueScheduledExecutorProperties(String replicationQueueScheduledExecutorPropertiesString) { + testImmutability("replicationQueueScheduledExecutorProperties"); + this.replicationQueueScheduledExecutorProperties = toTypedProperties(replicationQueueScheduledExecutorPropertiesString); + } + + public short getReplicationVersion() { + return replicationVersion; + } + + public String getReplicationVersionString() { + return Version.decodeVersion(replicationVersion); + } + + public void setReplicationVersion(short replicationVersion) { + testImmutability("replicationVersion"); + this.replicationVersion = replicationVersion; + } + + public void setReplicationVersionString(String replicationVersionString) { + testImmutability("replicationVersion"); + this.replicationVersion = Version.getVersionShort(replicationVersionString); + } + @Override public boolean equals(Object o) { if (this == o) return true; @@ -131,20 +274,31 @@ if (objectInputStreamPoolSize != that.objectInputStreamPoolSize) return false; if (objectOutputStreamPoolSize != that.objectOutputStreamPoolSize) return false; + if (replicationVersion != that.replicationVersion) return false; if (asyncListenerExecutorFactoryClass != null ? !asyncListenerExecutorFactoryClass.equals(that.asyncListenerExecutorFactoryClass) : that.asyncListenerExecutorFactoryClass != null) return false; + if (asyncListenerExecutorProperties != null ? !asyncListenerExecutorProperties.equals(that.asyncListenerExecutorProperties) : that.asyncListenerExecutorProperties != null) + return false; if (asyncSerializationExecutorFactoryClass != null ? !asyncSerializationExecutorFactoryClass.equals(that.asyncSerializationExecutorFactoryClass) : that.asyncSerializationExecutorFactoryClass != null) return false; + if (asyncSerializationExecutorProperties != null ? !asyncSerializationExecutorProperties.equals(that.asyncSerializationExecutorProperties) : that.asyncSerializationExecutorProperties != null) + return false; + if (clusterName != null ? !clusterName.equals(that.clusterName) : that.clusterName != null) return false; if (defaultConfiguration != null ? !defaultConfiguration.equals(that.defaultConfiguration) : that.defaultConfiguration != null) return false; if (evictionScheduledExecutorFactoryClass != null ? !evictionScheduledExecutorFactoryClass.equals(that.evictionScheduledExecutorFactoryClass) : that.evictionScheduledExecutorFactoryClass != null) return false; + if (evictionScheduledExecutorProperties != null ? !evictionScheduledExecutorProperties.equals(that.evictionScheduledExecutorProperties) : that.evictionScheduledExecutorProperties != null) + return false; if (marshallVersion != null ? !marshallVersion.equals(that.marshallVersion) : that.marshallVersion != null) return false; if (marshallerClass != null ? !marshallerClass.equals(that.marshallerClass) : that.marshallerClass != null) return false; if (replicationQueueScheduledExecutorFactoryClass != null ? !replicationQueueScheduledExecutorFactoryClass.equals(that.replicationQueueScheduledExecutorFactoryClass) : that.replicationQueueScheduledExecutorFactoryClass != null) return false; + if (replicationQueueScheduledExecutorProperties != null ? !replicationQueueScheduledExecutorProperties.equals(that.replicationQueueScheduledExecutorProperties) : that.replicationQueueScheduledExecutorProperties != null) + return false; + if (shutdownHookBehavior != that.shutdownHookBehavior) return false; if (transportClass != null ? !transportClass.equals(that.transportClass) : that.transportClass != null) return false; if (transportProperties != null ? !transportProperties.equals(that.transportProperties) : that.transportProperties != null) @@ -156,9 +310,13 @@ @Override public int hashCode() { int result = asyncListenerExecutorFactoryClass != null ? asyncListenerExecutorFactoryClass.hashCode() : 0; + result = 31 * result + (asyncListenerExecutorProperties != null ? asyncListenerExecutorProperties.hashCode() : 0); result = 31 * result + (asyncSerializationExecutorFactoryClass != null ? asyncSerializationExecutorFactoryClass.hashCode() : 0); + result = 31 * result + (asyncSerializationExecutorProperties != null ? asyncSerializationExecutorProperties.hashCode() : 0); result = 31 * result + (evictionScheduledExecutorFactoryClass != null ? evictionScheduledExecutorFactoryClass.hashCode() : 0); + result = 31 * result + (evictionScheduledExecutorProperties != null ? evictionScheduledExecutorProperties.hashCode() : 0); result = 31 * result + (replicationQueueScheduledExecutorFactoryClass != null ? replicationQueueScheduledExecutorFactoryClass.hashCode() : 0); + result = 31 * result + (replicationQueueScheduledExecutorProperties != null ? replicationQueueScheduledExecutorProperties.hashCode() : 0); result = 31 * result + (marshallerClass != null ? marshallerClass.hashCode() : 0); result = 31 * result + (marshallVersion != null ? marshallVersion.hashCode() : 0); result = 31 * result + objectInputStreamPoolSize; @@ -166,6 +324,9 @@ result = 31 * result + (transportClass != null ? transportClass.hashCode() : 0); result = 31 * result + (transportProperties != null ? transportProperties.hashCode() : 0); result = 31 * result + (defaultConfiguration != null ? defaultConfiguration.hashCode() : 0); + result = 31 * result + (clusterName != null ? clusterName.hashCode() : 0); + result = 31 * result + (shutdownHookBehavior != null ? shutdownHookBehavior.hashCode() : 0); + result = 31 * result + (int) replicationVersion; return result; } Deleted: core/branches/flat/src/main/java/org/horizon/config/NonOverridable.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/config/NonOverridable.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/config/NonOverridable.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,26 +0,0 @@ -package org.horizon.config; - -import java.lang.annotation.Documented; -import java.lang.annotation.ElementType; -import java.lang.annotation.Retention; -import java.lang.annotation.RetentionPolicy; -import java.lang.annotation.Target; - -/** - * This annotation marks configuration attributes as non-overridable by named caches. - * - * @author Manik Surtani - * @since 1.0 - */ -// ensure this annotation is available at runtime. -(a)Retention(RetentionPolicy.RUNTIME) - -// ensure that this annotation is documented on fields in Configuration -@Documented - -// only applies to fields. -(a)Target(ElementType.FIELD) - -public @interface NonOverridable { - -} Deleted: core/branches/flat/src/main/java/org/horizon/config/OldFileFormatException.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/config/OldFileFormatException.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/config/OldFileFormatException.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,39 +0,0 @@ -/* - * JBoss, Home of Professional Open Source. - * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors - * as indicated by the @author tags. See the copyright.txt file in the - * distribution for a full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.horizon.config; - -/** - * This exception indicates that an old configuration file is passed to a parser that only knows how to handle newer - * configuration file. - * - * @author Mircea.Markus(a)jboss.com - * @since 1.0 - */ -public class OldFileFormatException extends ConfigurationException { - public OldFileFormatException(String string) { - super(string); - } - - public OldFileFormatException() { - this("The configuration file has an old format."); - } -} Modified: core/branches/flat/src/main/java/org/horizon/config/parsing/XmlConfigurationParserJBC3.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/config/parsing/XmlConfigurationParserJBC3.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/config/parsing/XmlConfigurationParserJBC3.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -26,6 +26,7 @@ import org.horizon.config.Configuration.CacheMode; import org.horizon.config.ConfigurationException; import org.horizon.config.CustomInterceptorConfig; +import org.horizon.config.GlobalConfiguration; import org.horizon.config.parsing.element.CustomInterceptorsElementParser; import org.horizon.config.parsing.element.EvictionElementParser; import org.horizon.config.parsing.element.LoadersElementParser; @@ -36,7 +37,6 @@ import org.xml.sax.ErrorHandler; import java.io.InputStream; -import java.net.URL; import java.util.List; /** @@ -60,6 +60,7 @@ * the resulting configuration. */ private Configuration config = new Configuration(); + private GlobalConfiguration globalconfig = new GlobalConfiguration(); private Element root; /** @@ -177,7 +178,7 @@ configureAsyncMode(asyncEl); } String cn = getAttributeValue(e, "clusterName"); - if (existsAttribute(cn)) config.setClusterName(cn); +// if (existsAttribute(cn)) config.setClusterName(cn); configureStateRetrieval(getSingleElementInCoreNS("stateRetrieval", e)); configureTransport(getSingleElementInCoreNS("jgroupsConfig", e)); } @@ -205,20 +206,20 @@ private void configureSerialization(Element element) { if (element == null) return; String objectInputStreamPoolSize = getAttributeValue(element, "objectInputStreamPoolSize"); - if (existsAttribute(objectInputStreamPoolSize)) - config.setObjectInputStreamPoolSize(getInt(objectInputStreamPoolSize)); +// if (existsAttribute(objectInputStreamPoolSize)) +// config.setObjectInputStreamPoolSize(getInt(objectInputStreamPoolSize)); String objectOutputStreamPoolSize = getAttributeValue(element, "objectOutputStreamPoolSize"); - if (existsAttribute(objectOutputStreamPoolSize)) - config.setObjectOutputStreamPoolSize(getInt(objectOutputStreamPoolSize)); +// if (existsAttribute(objectOutputStreamPoolSize)) +// config.setObjectOutputStreamPoolSize(getInt(objectOutputStreamPoolSize)); String version = getAttributeValue(element, "version"); - if (existsAttribute(version)) config.setReplVersionString(version); +// if (existsAttribute(version)) config.setReplVersionString(version); String marshallerClass = getAttributeValue(element, "marshallerClass"); - if (existsAttribute(marshallerClass)) config.setMarshallerClass(marshallerClass); +// if (existsAttribute(marshallerClass)) config.setMarshallerClass(marshallerClass); String useLazyDeserialization = getAttributeValue(element, "useLazyDeserialization"); if (existsAttribute(useLazyDeserialization)) config.setUseLazyDeserialization(getBoolean(useLazyDeserialization)); String useRegionBasedMarshalling = getAttributeValue(element, "useRegionBasedMarshalling"); - if (existsAttribute(useRegionBasedMarshalling)) - config.setUseRegionBasedMarshalling(getBoolean(useRegionBasedMarshalling)); +// if (existsAttribute(useRegionBasedMarshalling)) +// config.setUseRegionBasedMarshalling(getBoolean(useRegionBasedMarshalling)); } private void configureCustomInterceptors(Element element) { @@ -231,10 +232,10 @@ private void configureListeners(Element element) { if (element == null) return; //this element is optional String asyncPoolSizeStr = getAttributeValue(element, "asyncPoolSize"); - if (existsAttribute(asyncPoolSizeStr)) config.setListenerAsyncPoolSize(getInt(asyncPoolSizeStr)); +// if (existsAttribute(asyncPoolSizeStr)) config.setListenerAsyncPoolSize(getInt(asyncPoolSizeStr)); String asyncQueueSizeStr = getAttributeValue(element, "asyncQueueSize"); - if (existsAttribute(asyncQueueSizeStr)) config.setListenerAsyncQueueSize(getInt(asyncQueueSizeStr)); +// if (existsAttribute(asyncQueueSizeStr)) config.setListenerAsyncQueueSize(getInt(asyncQueueSizeStr)); } private void configureInvocationBatching(Element element) { @@ -265,29 +266,29 @@ private void configureShutdown(Element element) { if (element == null) return; String hookBehavior = getAttributeValue(element, "hookBehavior"); - if (existsAttribute(hookBehavior)) config.setShutdownHookBehavior(hookBehavior); +// if (existsAttribute(hookBehavior)) config.setShutdownHookBehavior(hookBehavior); } private void configureTransport(Element element) { - if (element == null) return; //transport might be missing - - // first see if a configFile is provided - String cfgFile = getAttributeValue(element, "configFile"); - if (existsAttribute(cfgFile)) { - // try and load this file - URL u = new FileLookup().lookupFileLocation(cfgFile); - config.setJgroupsConfigFile(u); - } else { - String multiplexerStack = getAttributeValue(element, "multiplexerStack"); - if (existsAttribute(multiplexerStack)) { - config.setMultiplexerStack(multiplexerStack); - } else { - JGroupsStackParser stackParser = new JGroupsStackParser(); - String clusterConfigStr = stackParser.parseClusterConfigXml(element); - if (clusterConfigStr != null && clusterConfigStr.trim().length() > 0) - config.setClusterConfig(clusterConfigStr); - } - } +// if (element == null) return; //transport might be missing +// +// // first see if a configFile is provided +// String cfgFile = getAttributeValue(element, "configFile"); +// if (existsAttribute(cfgFile)) { +// // try and load this file +// URL u = new FileLookup().lookupFileLocation(cfgFile); +// config.setJgroupsConfigFile(u); +// } else { +// String multiplexerStack = getAttributeValue(element, "multiplexerStack"); +// if (existsAttribute(multiplexerStack)) { +// config.setMultiplexerStack(multiplexerStack); +// } else { +// JGroupsStackParser stackParser = new JGroupsStackParser(); +// String clusterConfigStr = stackParser.parseClusterConfigXml(element); +// if (clusterConfigStr != null && clusterConfigStr.trim().length() > 0) +// config.setClusterConfig(clusterConfigStr); +// } +// } } private void configureStartup(Element element) { @@ -324,12 +325,12 @@ if (existsAttribute(replQueueMaxElements)) config.setReplQueueMaxElements(getInt(replQueueMaxElements)); String serializationExecutorPoolSize = getAttributeValue(element, "serializationExecutorPoolSize"); - if (existsAttribute(serializationExecutorPoolSize)) - config.setSerializationExecutorPoolSize(getInt(serializationExecutorPoolSize)); +// if (existsAttribute(serializationExecutorPoolSize)) +// config.setSerializationExecutorPoolSize(getInt(serializationExecutorPoolSize)); String serializationExecutorQueueSize = getAttributeValue(element, "serializationExecutorQueueSize"); - if (existsAttribute(serializationExecutorQueueSize)) - config.setSerializationExecutorQueueSize(getInt(serializationExecutorQueueSize)); +// if (existsAttribute(serializationExecutorQueueSize)) +// config.setSerializationExecutorQueueSize(getInt(serializationExecutorQueueSize)); } private void configureLocking(Element element) { Modified: core/branches/flat/src/main/java/org/horizon/eviction/EvictionWatcher.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/eviction/EvictionWatcher.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/eviction/EvictionWatcher.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -2,7 +2,7 @@ import org.horizon.Cache; import org.horizon.notifications.annotation.CacheEntryEvicted; -import org.horizon.notifications.annotation.CacheListener; +import org.horizon.notifications.annotation.Listener; import org.horizon.notifications.event.CacheEntryEvictedEvent; import org.horizon.tree.Fqn; @@ -15,7 +15,7 @@ * @author Manik Surtani (<a href="mailto:manik AT jboss DOT org">manik AT jboss DOT org</a>) * @since 1.0 */ -@CacheListener +@Listener public class EvictionWatcher { Cache<?, ?> cache; List<Object> keysToWaitFor; @@ -27,7 +27,7 @@ public EvictionWatcher(Cache<?, ?> cache, List<Object> keysToWaitFor) { this.cache = cache; this.keysToWaitFor = new ArrayList<Object>(keysToWaitFor); - cache.addCacheListener(this); + cache.addListener(this); } @CacheEntryEvicted Modified: core/branches/flat/src/main/java/org/horizon/factories/AbstractComponentRegistry.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/factories/AbstractComponentRegistry.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/factories/AbstractComponentRegistry.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -37,11 +37,9 @@ import org.horizon.factories.scopes.Scopes; import org.horizon.lifecycle.Lifecycle; import org.horizon.logging.Log; -import org.horizon.logging.LogFactory; import org.horizon.util.BeanUtils; import org.horizon.util.ReflectionUtil; -import javax.management.MBeanServerFactory; import java.lang.reflect.Method; import java.util.ArrayList; import java.util.Collections; @@ -83,8 +81,6 @@ */ private Map<Class, Class<? extends ComponentFactory>> defaultFactories = null; - private static final Log log = LogFactory.getLog(AbstractComponentRegistry.class); - private static final boolean trace = log.isTraceEnabled(); protected static final Object NULL_COMPONENT = new Object(); // component and method containers @@ -93,16 +89,6 @@ CacheStatus state = CacheStatus.INSTANTIATED; /** - * Hook to shut down the cache when the JVM exits. - */ - private Thread shutdownHook; - /** - * A flag that the shutdown hook sets before calling cache.stop(). Allows stop() to identify if it has been called - * from a shutdown hook. - */ - private boolean invokedFromShutdownHook; - - /** * Retrieves the state of the registry * * @return state of the registry @@ -111,6 +97,8 @@ return state; } + protected abstract Log getLog(); + /** * Wires an object instance with dependencies annotated with the {@link Inject} annotation, creating more components * as needed based on the Configuration passed in if these additional components don't exist in the {@link @@ -167,22 +155,23 @@ * @param type type of component */ public void registerComponent(Object component, Class type) { + registerComponent(component, type.getName()); + } - String name = type.getName(); + public void registerComponent(Object component, String name) { Component old = componentLookup.get(name); if (old != null) { // if they are equal don't bother if (old.instance.equals(component)) { - if (trace) - log.trace("Attempting to register a component equal to one that already exists under the same name (" + name + "). Not doing anything."); + getLog().trace("Attempting to register a component equal to one that already exists under the same name ({0}). Not doing anything.", name); return; } } Component c; if (old != null) { - if (trace) log.trace("Replacing old component " + old + " with new instance " + component); + getLog().trace("Replacing old component {0} with new instance {1}", old, component); old.instance = component; old.methodsScanned = false; c = old; @@ -192,10 +181,10 @@ c = new Component(); c.name = name; c.instance = component; - if (trace) log.trace("Registering component " + c + " under name " + name); + getLog().trace("Registering component {0} under name {1}", c, name); componentLookup.put(name, c); } - c.nonVolatile = component.getClass().isAnnotationPresent(NonVolatile.class) || type.isAnnotationPresent(NonVolatile.class); + c.nonVolatile = component.getClass().isAnnotationPresent(NonVolatile.class); addComponentDependencies(c); // inject dependencies for this component c.injectDependencies(); @@ -264,7 +253,8 @@ if (component != null) { registerComponent(component, componentClass); } else if (attemptedFactoryConstruction) { - if (trace) log.trace("Registering a null for component " + componentClass.getSimpleName()); + if (getLog().isTraceEnabled()) + getLog().trace("Registering a null for component {0}", componentClass.getSimpleName()); registerNullComponent(componentClass); } } @@ -356,8 +346,7 @@ */ @SuppressWarnings("unchecked") protected <T> T getFromConfiguration(Class<T> componentClass) { - if (log.isDebugEnabled()) - log.debug("Looking in configuration for an instance of " + componentClass + " that may have been injected from an external source."); + getLog().debug("Looking in configuration for an instance of {0} that may have been injected from an external source.", componentClass); Method getter = BeanUtils.getterMethod(Configuration.class, componentClass); T returnValue = null; @@ -366,7 +355,7 @@ returnValue = (T) getter.invoke(getConfiguration()); } catch (Exception e) { - log.warn("Unable to invoke getter " + getter + " on Configuration.class!", e); + getLog().warn("Unable to invoke getter {0} on Configuration.class!", e, getter); } } @@ -378,7 +367,7 @@ returnValue = (T) getter.invoke(getConfiguration().getRuntimeConfig()); } catch (Exception e) { - log.warn("Unable to invoke getter " + getter + " on RuntimeConfig.class!", e); + getLog().warn("Unable to invoke getter {0} on RuntimeConfig.class!", e, getter); } } } @@ -474,7 +463,8 @@ } } - if (trace) log.trace("Reset volatile components. Registry now contains " + componentLookup.keySet()); + if (getLog().isTraceEnabled()) + getLog().trace("Reset volatile components. Registry now contains {0}", componentLookup.keySet()); } // ------------------------------ START: Publicly available lifecycle methods ----------------------------- @@ -542,7 +532,7 @@ } catch (Throwable t) { if (failed) { - log.warn("Attempted to stop() from FAILED state, but caught exception; try calling destroy()", t); + getLog().warn("Attempted to stop() from FAILED state, but caught exception; try calling destroy()", t); } failed = true; handleLifecycleTransitionFailure(t); @@ -566,7 +556,7 @@ stop(); } catch (CacheException e) { - log.warn("Needed to call stop() before destroying but stop() threw exception. Proceeding to destroy", e); + getLog().warn("Needed to call stop() before destroying but stop() threw exception. Proceeding to destroy", e); } } else return; @@ -639,36 +629,16 @@ addShutdownHook(); - log.info("JBoss Cache version: " + Version.printVersion()); + getLog().info("JBoss Cache version: " + Version.printVersion()); state = CacheStatus.STARTED; } - private void addShutdownHook() { - ArrayList al = MBeanServerFactory.findMBeanServer(null); - boolean registerShutdownHook = (getConfiguration().getShutdownHookBehavior() == Configuration.ShutdownHookBehavior.DEFAULT && al.size() == 0) - || getConfiguration().getShutdownHookBehavior() == Configuration.ShutdownHookBehavior.REGISTER; + protected void addShutdownHook() { + // no op. Override if needed. + } - if (registerShutdownHook) { - if (log.isTraceEnabled()) - log.trace("Registering a shutdown hook. Configured behavior = " + getConfiguration().getShutdownHookBehavior()); - shutdownHook = new Thread() { - @Override - public void run() { - try { - invokedFromShutdownHook = true; - AbstractComponentRegistry.this.stop(); - } - finally { - invokedFromShutdownHook = false; - } - } - }; - - Runtime.getRuntime().addShutdownHook(shutdownHook); - } else { - if (log.isTraceEnabled()) - log.trace("Not registering a shutdown hook. Configured behavior = " + getConfiguration().getShutdownHookBehavior()); - } + protected void removeShutdownHook() { + // no op. Override if needed. } /** @@ -676,8 +646,7 @@ */ private void internalStop() { state = CacheStatus.STOPPING; - // if this is called from a source other than the shutdown hook, deregister the shutdown hook. - if (!invokedFromShutdownHook && shutdownHook != null) Runtime.getRuntime().removeShutdownHook(shutdownHook); + removeShutdownHook(); List<PrioritizedMethod> stopMethods = new ArrayList<PrioritizedMethod>(componentLookup.size()); for (Component c : componentLookup.values()) stopMethods.addAll(c.stopMethods); @@ -723,17 +692,17 @@ * @return true if invocations are allowed, false otherwise. */ public boolean invocationsAllowed(boolean originLocal) { - log.trace("Testing if invocations are allowed."); + getLog().trace("Testing if invocations are allowed."); if (state.allowInvocations()) return true; // if this is a locally originating call and the cache is not in a valid state, return false. if (originLocal) return false; - log.trace("Is remotely originating."); + getLog().trace("Is remotely originating."); // else if this is a remote call and the status is STARTING, wait until the cache starts. if (state == CacheStatus.STARTING) { - log.trace("Cache is starting; block."); + getLog().trace("Cache is starting; block."); try { blockUntilCacheStarts(); return true; @@ -742,7 +711,7 @@ Thread.currentThread().interrupt(); } } else { - log.warn("Received a remote call but the cache is not in STARTED state - ignoring call."); + getLog().warn("Received a remote call but the cache is not in STARTED state - ignoring call."); } return false; } Modified: core/branches/flat/src/main/java/org/horizon/factories/ComponentRegistry.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/factories/ComponentRegistry.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/factories/ComponentRegistry.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -5,6 +5,8 @@ import org.horizon.config.Configuration; import org.horizon.factories.scopes.ScopeDetector; import org.horizon.factories.scopes.Scopes; +import org.horizon.logging.Log; +import org.horizon.logging.LogFactory; import java.util.Map; @@ -15,7 +17,10 @@ * @since 1.0 */ public class ComponentRegistry extends AbstractComponentRegistry { + GlobalComponentRegistry globalComponents; + String cacheName; + Log log; /** * Creates an instance of the component registry. The configuration passed in is automatically registered. @@ -24,9 +29,11 @@ * @param cache cache * @param globalComponents Shared Component Registry to delegate to */ - public ComponentRegistry(Configuration configuration, CacheSPI cache, GlobalComponentRegistry globalComponents) { + public ComponentRegistry(String cacheName, Configuration configuration, CacheSPI cache, GlobalComponentRegistry globalComponents) { try { + this.cacheName = cacheName; + this.log = LogFactory.getLog(ComponentRegistry.class.getName() + ":" + cacheName); registerDefaultClassLoader(null); registerComponent(this, ComponentRegistry.class); registerComponent(configuration, Configuration.class); @@ -39,6 +46,10 @@ } } + protected Log getLog() { + return log; + } + /** * Registers the default class loader. This method *must* be called before any other components are registered, * typically called by bootstrap code. Defensively, it is called in the constructor of ComponentRegistry with a null Modified: core/branches/flat/src/main/java/org/horizon/factories/DefaultCacheFactory.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/factories/DefaultCacheFactory.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/factories/DefaultCacheFactory.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -27,7 +27,7 @@ import org.horizon.config.Configuration; import org.horizon.config.ConfigurationException; import org.horizon.jmx.PlatformMBeanServerRegistration; -import org.horizon.manager.CacheManager; +import org.horizon.manager.DefaultCacheManager; /** * Default implementation of the {@link CacheFactory} interface. @@ -70,23 +70,23 @@ } public Cache<K, V> createDefaultCache(Configuration configuration) throws ConfigurationException { - return createCache(configuration, null, CacheManager.DEFAULT_CACHE_NAME); + return createCache(configuration, null, DefaultCacheManager.DEFAULT_CACHE_NAME); } protected CacheSPI<K, V> createAndWire(Configuration configuration, GlobalComponentRegistry globalComponentRegistry, String cacheName) throws Exception { CacheSPI<K, V> spi = new CacheDelegate<K, V>(cacheName); - bootstrap(spi, configuration, globalComponentRegistry); + bootstrap(cacheName, spi, configuration, globalComponentRegistry); return spi; } /** * Bootstraps this factory with a Configuration and a ComponentRegistry. */ - private void bootstrap(CacheSPI spi, Configuration configuration, GlobalComponentRegistry globalComponentRegistry) { + private void bootstrap(String cacheName, CacheSPI spi, Configuration configuration, GlobalComponentRegistry globalComponentRegistry) { this.configuration = configuration; // injection bootstrap stuff - componentRegistry = new ComponentRegistry(configuration, spi, globalComponentRegistry); + componentRegistry = new ComponentRegistry(cacheName, configuration, spi, globalComponentRegistry); componentRegistry.registerDefaultClassLoader(defaultClassLoader); componentRegistry.registerComponent(spi, CacheSPI.class); componentRegistry.registerComponent(new PlatformMBeanServerRegistration(), PlatformMBeanServerRegistration.class); Modified: core/branches/flat/src/main/java/org/horizon/factories/EmptyConstructorFactory.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/factories/EmptyConstructorFactory.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/factories/EmptyConstructorFactory.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -31,10 +31,7 @@ import org.horizon.loader.CacheLoaderManager; import org.horizon.marshall.Marshaller; import org.horizon.marshall.VersionAwareMarshaller; -import org.horizon.notifications.Notifier; -import org.horizon.remoting.ChannelMessageListener; -import org.horizon.remoting.RPCManager; -import org.horizon.remoting.RPCManagerImpl; +import org.horizon.notifications.CacheNotifier; import org.horizon.transaction.TransactionTable; /** @@ -43,8 +40,8 @@ * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) * @since 1.0 */ -@DefaultFactoryFor(classes = {Notifier.class, - ChannelMessageListener.class, CacheLoaderManager.class, Marshaller.class, InvocationContextContainer.class, +@DefaultFactoryFor(classes = {CacheNotifier.class, + CacheLoaderManager.class, Marshaller.class, InvocationContextContainer.class, TransactionTable.class, BatchContainer.class, ContextFactory.class, EntryFactory.class, CommandsFactory.class}) public class EmptyConstructorFactory extends ComponentFactory { @Override @@ -54,8 +51,6 @@ Class componentImpl; if (componentType.equals(Marshaller.class)) { componentImpl = VersionAwareMarshaller.class; - } else if (componentType.equals(RPCManager.class)) { - componentImpl = RPCManagerImpl.class; } else { // add an "Impl" to the end of the class name and try again componentImpl = getClass().getClassLoader().loadClass(componentType.getName() + "Impl"); Modified: core/branches/flat/src/main/java/org/horizon/factories/GlobalComponentRegistry.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/factories/GlobalComponentRegistry.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/factories/GlobalComponentRegistry.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -2,7 +2,14 @@ import org.horizon.CacheException; import org.horizon.config.GlobalConfiguration; +import static org.horizon.config.GlobalConfiguration.ShutdownHookBehavior.DEFAULT; +import static org.horizon.config.GlobalConfiguration.ShutdownHookBehavior.REGISTER; +import org.horizon.logging.Log; +import org.horizon.logging.LogFactory; +import javax.management.MBeanServerFactory; +import java.util.ArrayList; + /** * // TODO: Manik: Document this! * @@ -10,13 +17,28 @@ * @since 1.0 */ public class GlobalComponentRegistry extends AbstractComponentRegistry { + + private Log log = LogFactory.getLog(GlobalComponentRegistry.class); /** + * Hook to shut down the cache when the JVM exits. + */ + private Thread shutdownHook; + /** + * A flag that the shutdown hook sets before calling cache.stop(). Allows stop() to identify if it has been called + * from a shutdown hook. + */ + private boolean invokedFromShutdownHook; + + private GlobalConfiguration globalConfiguration; + + /** * Creates an instance of the component registry. The configuration passed in is automatically registered. * * @param configuration configuration with which this is created */ public GlobalComponentRegistry(GlobalConfiguration configuration) { try { + globalConfiguration = configuration; registerComponent(configuration, GlobalConfiguration.class); registerComponent(this, GlobalComponentRegistry.class); } @@ -24,4 +46,42 @@ throw new CacheException("Unable to construct a GlobalComponentRegistry!", e); } } + + protected Log getLog() { + return log; + } + + @Override + protected void removeShutdownHook() { + // if this is called from a source other than the shutdown hook, deregister the shutdown hook. + if (!invokedFromShutdownHook && shutdownHook != null) Runtime.getRuntime().removeShutdownHook(shutdownHook); + } + + @Override + protected void addShutdownHook() { + ArrayList al = MBeanServerFactory.findMBeanServer(null); + boolean registerShutdownHook = (globalConfiguration.getShutdownHookBehavior() == DEFAULT && al.size() == 0) + || globalConfiguration.getShutdownHookBehavior() == REGISTER; + + if (registerShutdownHook) { + log.trace("Registering a shutdown hook. Configured behavior = {0}", globalConfiguration.getShutdownHookBehavior()); + shutdownHook = new Thread() { + @Override + public void run() { + try { + invokedFromShutdownHook = true; + GlobalComponentRegistry.this.stop(); + } + finally { + invokedFromShutdownHook = false; + } + } + }; + + Runtime.getRuntime().addShutdownHook(shutdownHook); + } else { + + log.trace("Not registering a shutdown hook. Configured behavior = {0}", globalConfiguration.getShutdownHookBehavior()); + } + } } Added: core/branches/flat/src/main/java/org/horizon/factories/KnownComponentNames.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/factories/KnownComponentNames.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/factories/KnownComponentNames.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,13 @@ +package org.horizon.factories; + +/** + * Holder for known named component names. To be used with {@link org.horizon.factories.annotations.ComponentName} + * annotation. + * + * @author Manik Surtani + * @since 1.0 + */ +public class KnownComponentNames { + public static final String ASYNC_SERIALIZATION_EXECUTOR = "org.horizon.executors.serialization"; + public static final String ASYNC_NOTIFICATION_EXECUTOR = "org.horizon.executors.notification"; +} Added: core/branches/flat/src/main/java/org/horizon/factories/annotations/ComponentName.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/factories/annotations/ComponentName.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/factories/annotations/ComponentName.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,18 @@ +package org.horizon.factories.annotations; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Mechanism for specifying the name of components to retrieve + * + * @author Manik Surtani + * @since 1.0 + */ +(a)Target(ElementType.PARAMETER) +(a)Retention(RetentionPolicy.RUNTIME) +public @interface ComponentName { + String value(); +} Modified: core/branches/flat/src/main/java/org/horizon/interceptors/InvalidationInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/interceptors/InvalidationInterceptor.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/interceptors/InvalidationInterceptor.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -198,7 +198,7 @@ // increment invalidations counter if statistics maintained incrementInvalidations(); InvalidateCommand command = commandsFactory.buildInvalidateCommand(fqn); - if (log.isDebugEnabled()) log.debug("Cache [" + rpcManager.getLocalAddress() + "] replicating " + command); + if (log.isDebugEnabled()) log.debug("Cache [" + rpcManager.getAddress() + "] replicating " + command); // voila, invalidated! replicateCall(ctx, command, synchronous, ctx.getOptionOverrides()); } Modified: core/branches/flat/src/main/java/org/horizon/interceptors/InvocationContextInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/interceptors/InvocationContextInterceptor.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/interceptors/InvocationContextInterceptor.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -199,7 +199,7 @@ * @return true if the gtx is remote, false if it originated locally. */ private boolean isRemoteGlobalTx(GlobalTransaction gtx) { - return gtx != null && (gtx.getAddress() != null) && (!gtx.getAddress().equals(rpcManager.getLocalAddress())); + return gtx != null && (gtx.getAddress() != null) && (!gtx.getAddress().equals(rpcManager.getAddress())); } private void copyInvocationScopeOptionsToTxScope(InvocationContext ctx) { Modified: core/branches/flat/src/main/java/org/horizon/interceptors/ReplicationInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/interceptors/ReplicationInterceptor.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/interceptors/ReplicationInterceptor.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -136,7 +136,7 @@ protected void runPreparePhase(PrepareCommand prepareMethod, GlobalTransaction gtx, InvocationContext ctx) throws Throwable { boolean async = configuration.getCacheMode() == Configuration.CacheMode.REPL_ASYNC; if (trace) { - log.trace("(" + rpcManager.getLocalAddress() + "): running remote prepare for global tx " + gtx + " with async mode=" + async); + log.trace("(" + rpcManager.getAddress() + "): running remote prepare for global tx " + gtx + " with async mode=" + async); } // this method will return immediately if we're the only member (because exclude_self=true) Modified: core/branches/flat/src/main/java/org/horizon/interceptors/TxInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/interceptors/TxInterceptor.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/interceptors/TxInterceptor.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -38,7 +38,7 @@ import org.horizon.factories.context.ContextFactory; import org.horizon.invocation.InvocationContextContainer; import org.horizon.lock.LockManager; -import org.horizon.notifications.Notifier; +import org.horizon.notifications.CacheNotifier; import org.horizon.remoting.RPCManager; import org.horizon.remoting.ReplicationException; import org.horizon.transaction.GlobalTransaction; @@ -69,7 +69,7 @@ public class TxInterceptor extends BaseTransactionalContextInterceptor { protected CommandsFactory commandsFactory; protected RPCManager rpcManager; - private Notifier notifier; + private CacheNotifier notifier; private InvocationContextContainer invocationContextContainer; private ComponentRegistry componentRegistry; private ContextFactory contextFactory; @@ -88,7 +88,7 @@ @Inject public void intialize(RPCManager rpcManager, ContextFactory contextFactory, - Notifier notifier, InvocationContextContainer icc, + CacheNotifier notifier, InvocationContextContainer icc, CommandsFactory factory, ComponentRegistry componentRegistry, LockManager lockManager) { this.contextFactory = contextFactory; this.commandsFactory = factory; @@ -129,7 +129,7 @@ return null; } try { - if (trace) log.trace("(" + rpcManager.getLocalAddress() + ") call on command [" + command + "]"); + if (trace) log.trace("(" + rpcManager.getAddress() + ") call on command [" + command + "]"); GlobalTransaction gtx = ctx.getGlobalTransaction(); Transaction ltx = txTable.getLocalTransaction(gtx, true); // disconnect if we have a current tx associated @@ -172,7 +172,7 @@ return null; } try { - if (trace) log.trace("(" + rpcManager.getLocalAddress() + ") call on command [" + command + "]"); + if (trace) log.trace("(" + rpcManager.getAddress() + ") call on command [" + command + "]"); GlobalTransaction gtx = ctx.getGlobalTransaction(); Transaction ltx = txTable.getLocalTransaction(gtx); if (ltx == null) { @@ -434,7 +434,7 @@ // -------------------------------------------------------------- protected PrepareCommand buildPrepareCommand(GlobalTransaction gtx, List modifications, boolean onePhaseCommit) { - return commandsFactory.buildPrepareCommand(gtx, modifications, rpcManager.getLocalAddress(), onePhaseCommit); + return commandsFactory.buildPrepareCommand(gtx, modifications, rpcManager.getAddress(), onePhaseCommit); } /** Modified: core/branches/flat/src/main/java/org/horizon/interceptors/base/BaseRpcInterceptor.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/interceptors/base/BaseRpcInterceptor.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/interceptors/base/BaseRpcInterceptor.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -30,13 +30,13 @@ import org.horizon.factories.annotations.Inject; import org.horizon.factories.annotations.Start; import org.horizon.remoting.RPCManager; +import org.horizon.remoting.ResponseMode; +import org.horizon.remoting.transport.Address; import org.horizon.transaction.GlobalTransaction; import org.horizon.transaction.TransactionTable; -import org.jgroups.Address; import javax.transaction.Transaction; import java.util.List; -import java.util.Vector; /** * Acts as a base for all RPC calls - subclassed by @@ -90,7 +90,7 @@ replicateCall(ctx, null, call, sync, o, false); } - protected void replicateCall(InvocationContext ctx, Vector<Address> recipients, ReplicableCommand c, boolean sync, Option o, boolean useOutOfBandMessage) throws Throwable { + protected void replicateCall(InvocationContext ctx, List<Address> recipients, ReplicableCommand c, boolean sync, Option o, boolean useOutOfBandMessage) throws Throwable { long syncReplTimeout = configuration.getSyncReplTimeout(); // test for option overrides @@ -114,14 +114,14 @@ replicateCall(recipients, c, sync, true, useOutOfBandMessage, false, syncReplTimeout); } - protected void replicateCall(Vector<Address> recipients, ReplicableCommand call, boolean sync, boolean wrapCacheCommandInReplicateMethod, boolean useOutOfBandMessage, boolean isBroadcast, long timeout) throws Throwable { + protected void replicateCall(List<Address> recipients, ReplicableCommand call, boolean sync, boolean wrapCacheCommandInReplicateMethod, boolean useOutOfBandMessage, boolean isBroadcast, long timeout) throws Throwable { if (trace) log.trace("Broadcasting call " + call + " to recipient list " + recipients); if (!sync && replicationQueue != null) { if (log.isDebugEnabled()) log.debug("Putting call " + call + " on the replication queue."); replicationQueue.add(commandsFactory.buildReplicateCommand(call)); } else { - Vector<Address> callRecipients = recipients; + List<Address> callRecipients = recipients; if (callRecipients == null) { callRecipients = null; if (trace) @@ -130,11 +130,11 @@ ReplicableCommand toCall = wrapCacheCommandInReplicateMethod ? commandsFactory.buildReplicateCommand(call) : call; - List rsps = rpcManager.callRemoteMethods(callRecipients, - toCall, - sync, // is synchronised? - timeout, - useOutOfBandMessage + List rsps = rpcManager.invokeRemotely(callRecipients, + toCall, + sync ? ResponseMode.SYNCHRONOUS : ResponseMode.ASYNCHRONOUS, // is synchronised? + timeout, + useOutOfBandMessage ); if (trace) log.trace("responses=" + rsps); if (sync) checkResponses(rsps); Modified: core/branches/flat/src/main/java/org/horizon/jmx/JmxRegistrationManager.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/jmx/JmxRegistrationManager.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/jmx/JmxRegistrationManager.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -179,7 +179,8 @@ objectNameBase = LOCAL_CACHE_PREFIX + System.currentTimeMillis(); } else //the cache is clustered { - objectNameBase = REPLICATED_CACHE_PREFIX + cacheSpi.getConfiguration().getClusterName(); + // TODO: make sure we append the cache name as well somewhere here, since > 1 cache will be attached to the same channel. + //objectNameBase = REPLICATED_CACHE_PREFIX + cacheSpi.getConfiguration().getGlobalConfiguration().getClusterName(); } } Modified: core/branches/flat/src/main/java/org/horizon/loader/ClusteredCacheLoader.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/loader/ClusteredCacheLoader.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/loader/ClusteredCacheLoader.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -133,7 +133,7 @@ @SuppressWarnings("deprecation") private Object callRemote(DataCommand dataCommand) throws Exception { - if (trace) log.trace("cache=" + cache.getLocalAddress() + "; calling with " + dataCommand); + if (trace) log.trace("cache=" + cache.getCacheManager().getAddress() + "; calling with " + dataCommand); // ClusteredGetCommand clusteredGet = commandsFactory.buildClusteredGetCommand(false, dataCommand); List resps; // JBCACHE-1186 Modified: core/branches/flat/src/main/java/org/horizon/loader/SingletonStoreCacheLoader.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/loader/SingletonStoreCacheLoader.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/loader/SingletonStoreCacheLoader.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -25,19 +25,18 @@ import org.horizon.config.CacheLoaderConfig.IndividualCacheLoaderConfig.SingletonStoreConfig; import org.horizon.logging.Log; import org.horizon.logging.LogFactory; -import org.horizon.notifications.annotation.CacheListener; import org.horizon.notifications.annotation.CacheStarted; import org.horizon.notifications.annotation.CacheStopped; +import org.horizon.notifications.annotation.Listener; import org.horizon.notifications.annotation.ViewChanged; import org.horizon.notifications.event.Event; import org.horizon.notifications.event.ViewChangedEvent; +import org.horizon.remoting.transport.Address; import org.horizon.tree.Fqn; -import org.jgroups.Address; -import org.jgroups.View; import java.io.ObjectInputStream; +import java.util.List; import java.util.Map; -import java.util.Vector; import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; @@ -141,7 +140,7 @@ { super.create(); - cache.addCacheListener(new SingletonStoreListener()); + cache.addListener(new SingletonStoreListener()); } /** @@ -324,17 +323,9 @@ * @param newView View instance containing the new view of the cluster * @return whether the current node is the coordinator or not. */ - private boolean isCoordinator(View newView) { - if (newView != null && localAddress != null) { - Vector mbrs = newView.getMembers(); - if (mbrs != null) { - if (mbrs.size() > 0 && localAddress.equals(mbrs.firstElement())) { - /* This node is the coordinator */ - return true; - } - } - - return false; + private boolean isCoordinator(List<Address> members) { + if (members != null && localAddress != null) { + return members.size() > 0 && localAddress.equals(members.get(0)); } /* Invalid new view, so previous value returned */ @@ -477,7 +468,7 @@ * SingletonStoreCacheLoader reacts to these changes in order to decide which node should interact with the * underlying cache store. */ - @CacheListener + @Listener public class SingletonStoreListener { /** * Cache started, check whether the node is the coordinator and set the singleton store cache loader's active @@ -485,7 +476,7 @@ */ @CacheStarted public void cacheStarted(Event e) { - localAddress = cache.getLocalAddress(); + localAddress = cache.getCacheManager().getAddress(); active = cache.getRPCManager().isCoordinator(); if (log.isDebugEnabled()) log.debug("cache started: " + this); } @@ -502,7 +493,7 @@ */ @ViewChanged public void viewChange(ViewChangedEvent event) { - boolean tmp = isCoordinator(event.getNewView()); + boolean tmp = isCoordinator(event.getNewMemberList()); if (active != tmp) { try { Deleted: core/branches/flat/src/main/java/org/horizon/manager/CacheManager.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/manager/CacheManager.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/manager/CacheManager.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,317 +0,0 @@ -/* - * JBoss, Home of Professional Open Source - * Copyright 2008, Red Hat Middleware LLC, and individual contributors - * by the @authors tag. See the copyright.txt in the distribution for a - * full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.horizon.manager; - -import org.horizon.Cache; -import org.horizon.CacheSPI; -import org.horizon.config.Configuration; -import org.horizon.config.ConfigurationException; -import org.horizon.config.GlobalConfiguration; -import org.horizon.config.parsing.XmlConfigurationParser; -import org.horizon.config.parsing.XmlConfigurationParserImpl; -import org.horizon.factories.DefaultCacheFactory; -import org.horizon.factories.GlobalComponentRegistry; -import org.horizon.lifecycle.Lifecycle; - -import java.io.IOException; -import java.io.InputStream; -import java.util.Map; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; - -/** - * A <tt>CacheManager</tt> is the primary mechanism for retrieving a {@link Cache} instance, and is often used as a - * starting point to using the {@link Cache}. - * <p/> - * <tt>CacheManager</tt>s are heavyweight objects, and we foresee no more than one <tt>CacheManager</tt> being used per - * JVM (unless specific configuration requirements require more than one; but either way, this would be a minimal and - * finite number of instances). - * <p/> - * Constructing a <tt>CacheManager</tt> is done via one of its constructors, which optionally take in a {@link - * org.horizon.config.Configuration} or a path or URL to a configuration XML file. - * <p/> - * Lifecycle - <tt>CacheManager</tt>s have a lifecycle (it implements {@link Lifecycle}) and the default constructors - * also call {@link #start()}. Overloaded versions of the constructors are available, that do not start the - * <tt>CacheManager</tt>, although it must be kept in mind that <tt>CacheManager</tt>s need to be started before they - * can be used to create <tt>Cache</tt> instances. - * <p/> - * Once constructed, <tt>CacheManager</tt>s should be made available to any component that requires a <tt>Cache</tt>, - * via JNDI or via some other mechanism such as an IoC container. - * <p/> - * You obtain <tt>Cache</tt> instances from the <tt>CacheManager</tt> by using one of the overloaded - * <tt>getCache()</tt>, methods. Note that with <tt>getCache()</tt>, there is no guarantee that the instance you get is - * brand-new and empty, since caches are named and shared. Because of this, the <tt>CacheManager</tt> also acts as a - * repository of <tt>Cache</tt>s, and is an effective mechanism of looking up or creating <tt>Cache</tt>s on demand. - * <p/> - * When the system shuts down, it should call {@link #stop()} on the <tt>CacheManager</tt>. This will ensure all caches - * within its scope are properly stopped as well. - * <p/> - * Sample usage: <code> CacheManager manager = CacheManager.getInstance("my-config-file.xml"); Cache entityCache = - * manager.getCache("myEntityCache"); entityCache.put("aPerson", new Person()); - * <p/> - * Configuration myNewConfiguration = new Configuration(); myNewConfiguration.setCacheMode(Configuration.CacheMode.LOCAL); - * manager.defineCache("myLocalCache", myNewConfiguration); Cache localCache = manager.getCache("myLocalCache"); - * </code> - * - * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) - * @since 1.0 - */ -public class CacheManager implements Lifecycle { - public static final String DEFAULT_CACHE_NAME = "org.jboss.starobrno.manager.CacheManager.DEFAULT_CACHE_NAME"; - protected GlobalConfiguration globalConfiguration; - private final ConcurrentMap<String, Cache> caches = new ConcurrentHashMap<String, Cache>(); - private final ConcurrentMap<String, Configuration> configurationOverrides = new ConcurrentHashMap<String, Configuration>(); - private GlobalComponentRegistry globalComponentRegistry; - - - /** - * Constructs and starts a default instance of the CacheManager, using configuration defaults. - */ - public CacheManager() { - this(null, null, true); - } - - /** - * Constructs a default instance of the CacheManager, using configuration defaults. - * - * @param start if true, the cache manager is started - */ - public CacheManager(boolean start) { - this(null, null, start); - } - - /** - * Constructs and starts a new instance of the CacheManager, using the default configuration passed in. Uses - * defaults for a {@link org.horizon.config.GlobalConfiguration}. - * - * @param defaultConfiguration configuration to use as a template for all caches created - */ - public CacheManager(Configuration defaultConfiguration) { - this(null, defaultConfiguration, true); - } - - /** - * Constructs a new instance of the CacheManager, using the default configuration passed in. Uses defaults for a - * {@link org.horizon.config.GlobalConfiguration}. - * - * @param defaultConfiguration configuration file to use as a template for all caches created - * @param start if true, the cache manager is started - */ - public CacheManager(Configuration defaultConfiguration, boolean start) { - this(null, defaultConfiguration, start); - } - - /** - * Constructs and starts a new instance of the CacheManager, using the global configuration passed in, and system - * defaults for the default named cache configuration. - * - * @param globalConfiguration GlobalConfiguration to use for all caches created - */ - public CacheManager(GlobalConfiguration globalConfiguration) { - this(globalConfiguration, null, true); - } - - /** - * Constructs a new instance of the CacheManager, using the global configuration passed in, and system defaults for - * the default named cache configuration. - * - * @param globalConfiguration GlobalConfiguration to use for all caches created - * @param start if true, the cache manager is started. - */ - public CacheManager(GlobalConfiguration globalConfiguration, boolean start) { - this(globalConfiguration, null, start); - } - - /** - * Constructs a new instance of the CacheManager, using the global and default configurations passed in. If either - * of these are null, system defaults are used. - * - * @param globalConfiguration global configuration to use. If null, a default instance is created. - * @param defaultConfiguration default configuration to use. If null, a default instance is created. - * @param start if true, the cache manager is started - */ - public CacheManager(GlobalConfiguration globalConfiguration, Configuration defaultConfiguration, boolean start) { - this.globalConfiguration = globalConfiguration == null ? new GlobalConfiguration() : globalConfiguration.clone(); - this.globalConfiguration.setDefaultConfiguration(defaultConfiguration == null ? new Configuration() : defaultConfiguration.clone()); - globalComponentRegistry = new GlobalComponentRegistry(globalConfiguration); - if (start) start(); - } - - - /** - * Constructs and starts a new instance of the CacheManager, using the configuration file name passed in. This - * constructor first searches for the named file on the classpath, and failing that, treats the file name as an - * absolute path. - * - * @param configurationFile name of configuration file to use as a template for all caches created - * @throws java.io.IOException if there is a problem with the configuration file. - */ - public CacheManager(String configurationFile) throws IOException { - this(configurationFile, true); - } - - /** - * Constructs a new instance of the CacheManager, using the configuration file name passed in. This constructor - * first searches for the named file on the classpath, and failing that, treats the file name as an absolute path. - * - * @param configurationFile name of configuration file to use as a template for all caches created - * @param start if true, the cache manager is started - * @throws java.io.IOException if there is a problem with the configuration file. - */ - public CacheManager(String configurationFile, boolean start) throws IOException { - try { - initialize(new XmlConfigurationParserImpl(configurationFile)); - } - catch (RuntimeException re) { - throw new ConfigurationException(re); - } - if (start) start(); - } - - /** - * Constructs and starts a new instance of the CacheManager, using the input stream passed in to read configuration - * file contents. - * - * @param configurationStream stream containing configuration file contents, to use as a template for all caches - * created - * @throws java.io.IOException if there is a problem with the configuration stream. - */ - public CacheManager(InputStream configurationStream) throws IOException { - this(configurationStream, true); - } - - /** - * Constructs a new instance of the CacheManager, using the input stream passed in to read configuration file - * contents. - * - * @param configurationStream stream containing configuration file contents, to use as a template for all caches - * created - * @param start if true, the cache manager is started - * @throws java.io.IOException if there is a problem reading the configuration stream - */ - public CacheManager(InputStream configurationStream, boolean start) throws IOException { - try { - initialize(new XmlConfigurationParserImpl(configurationStream)); - } - catch (RuntimeException re) { - throw new ConfigurationException(re); - } - if (start) start(); - } - - private void initialize(XmlConfigurationParser initializedParser) { - this.globalConfiguration = initializedParser.parseGlobalConfiguration(); - configurationOverrides.putAll(initializedParser.parseNamedConfigurations()); - } - - /** - * Defines a named cache. Named caches can be defined by using this method, in which case the configuration passed - * in is used to override the default configuration used when this cache manager instance was created. - * <p/> - * The other way to define named caches is declaratively, in the XML file passed in to the cache manager. - * <p/> - * A combination of approaches may also be used, provided the names do not conflict. - * - * @param cacheName name of cache to define - * @param configurationOverride configuration overrides to use - * @throws CacheNameExistsException if the name is already in use. - */ - public void defineCache(String cacheName, Configuration configurationOverride) throws CacheNameExistsException { - if (cacheName == null || configurationOverrides == null) - throw new NullPointerException("Null arguments not allowed"); - if (cacheName.equals(DEFAULT_CACHE_NAME)) - throw new IllegalArgumentException("Cache name cannot be used as it is a reserved, internal name"); - if (configurationOverrides.putIfAbsent(cacheName, configurationOverride) != null) - throw new CacheNameExistsException("Cache name [" + cacheName + "] already in use!"); - } - - - /** - * Retrieves the default cache associated with this cache manager. Note that the default cache does not need to be - * explicitly created with {@link #createCache(String)} since it is automatically created lazily when first used. - * <p/> - * As such, this method is always guaranteed to return the default cache. - * - * @return the default cache. - */ - public Cache getCache() { - return getCache(DEFAULT_CACHE_NAME); - } - - /** - * Retrieves a named cache from the system. If the cache has been previously created with the same name, the running - * cache instance is returned. Otherwise, this method attempts to create the cache first. - * <p/> - * When creating a new cache, this method will use the configuration passed in to the CacheManager on construction, - * as a template, and then optionally apply any overrides previously defined for the named cache using the {@link - * #defineCache(String, org.horizon.config.Configuration)} method, or declared in the configuration file. - * - * @param cacheName name of cache to retrieve - * @return a cache instance identified by cacheName - */ - public Cache getCache(String cacheName) { - if (cacheName == null) - throw new NullPointerException("Null arguments not allowed"); - - if (caches.containsKey(cacheName)) - return caches.get(cacheName); - - return createCache(cacheName); - } - - private Cache createCache(String cacheName) { - Configuration c = globalConfiguration.getDefaultConfiguration().clone(); - if (!cacheName.equals(DEFAULT_CACHE_NAME)) { - Configuration overrides = configurationOverrides.get(cacheName); - if (overrides != null) c.applyOverrides(overrides); - } - - Cache cache = new DefaultCacheFactory().createCache(c, globalComponentRegistry, cacheName); - Cache other = caches.putIfAbsent(cacheName, cache); - if (other == null) { - cache.start(); - return cache; - } else { - return other; - } - } - - public void start() { - // get a hold of the "default" cache to start this? - CacheSPI defaultCache = (CacheSPI) getCache(); - globalComponentRegistry = defaultCache.getComponentRegistry().getSharedComponentRegistry(); - } - - public void stop() { - // make sure we stop the default cache LAST! - Cache defaultCache = null; - for (Map.Entry<String, Cache> entry : caches.entrySet()) { - if (entry.getKey().equals(DEFAULT_CACHE_NAME)) { - defaultCache = entry.getValue(); - } else { - entry.getValue().stop(); - } - } - - if (defaultCache != null) defaultCache.stop(); - } -} Added: core/branches/flat/src/main/java/org/horizon/manager/CacheManager.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/manager/CacheManager.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/manager/CacheManager.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,93 @@ +package org.horizon.manager; + +import org.horizon.Cache; +import org.horizon.config.Configuration; +import org.horizon.lifecycle.Lifecycle; +import org.horizon.notifications.Listenable; +import org.horizon.remoting.transport.Address; + +import java.util.List; + +/** + * A <tt>CacheManager</tt> is the primary mechanism for retrieving a {@link org.horizon.Cache} instance, and is often + * used as a starting point to using the {@link org.horizon.Cache}. + * <p/> + * <tt>CacheManager</tt>s are heavyweight objects, and we foresee no more than one <tt>CacheManager</tt> being used per + * JVM (unless specific configuration requirements require more than one; but either way, this would be a minimal and + * finite number of instances). + * <p/> + * Constructing a <tt>CacheManager</tt> is done via one of its constructors, which optionally take in a {@link + * org.horizon.config.Configuration} or a path or URL to a configuration XML file. + * <p/> + * Lifecycle - <tt>CacheManager</tt>s have a lifecycle (it implements {@link org.horizon.lifecycle.Lifecycle}) and the + * default constructors also call {@link #start()}. Overloaded versions of the constructors are available, that do not + * start the <tt>CacheManager</tt>, although it must be kept in mind that <tt>CacheManager</tt>s need to be started + * before they can be used to create <tt>Cache</tt> instances. + * <p/> + * Once constructed, <tt>CacheManager</tt>s should be made available to any component that requires a <tt>Cache</tt>, + * via JNDI or via some other mechanism such as an IoC container. + * <p/> + * You obtain <tt>Cache</tt> instances from the <tt>CacheManager</tt> by using one of the overloaded + * <tt>getCache()</tt>, methods. Note that with <tt>getCache()</tt>, there is no guarantee that the instance you get is + * brand-new and empty, since caches are named and shared. Because of this, the <tt>CacheManager</tt> also acts as a + * repository of <tt>Cache</tt>s, and is an effective mechanism of looking up or creating <tt>Cache</tt>s on demand. + * <p/> + * When the system shuts down, it should call {@link #stop()} on the <tt>CacheManager</tt>. This will ensure all caches + * within its scope are properly stopped as well. + * <p/> + * Sample usage: <code> CacheManager manager = CacheManager.getInstance("my-config-file.xml"); Cache entityCache = + * manager.getCache("myEntityCache"); entityCache.put("aPerson", new Person()); + * <p/> + * Configuration myNewConfiguration = new Configuration(); myNewConfiguration.setCacheMode(Configuration.CacheMode.LOCAL); + * manager.defineCache("myLocalCache", myNewConfiguration); Cache localCache = manager.getCache("myLocalCache"); + * </code> + * + * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) + * @since 1.0 + */ +public interface CacheManager extends Lifecycle, Listenable { + /** + * Defines a named cache. Named caches can be defined by using this method, in which case the configuration passed + * in is used to override the default configuration used when this cache manager instance was created. + * <p/> + * The other way to define named caches is declaratively, in the XML file passed in to the cache manager. + * <p/> + * A combination of approaches may also be used, provided the names do not conflict. + * + * @param cacheName name of cache to define + * @param configurationOverride configuration overrides to use + * @throws CacheNameExistsException if the name is already in use. + */ + void defineCache(String cacheName, Configuration configurationOverride) throws CacheNameExistsException; + + /** + * Retrieves the default cache associated with this cache manager. + * <p/> + * As such, this method is always guaranteed to return the default cache. + * + * @return the default cache. + */ + Cache getCache(); + + /** + * Retrieves a named cache from the system. If the cache has been previously created with the same name, the running + * cache instance is returned. Otherwise, this method attempts to create the cache first. + * <p/> + * When creating a new cache, this method will use the configuration passed in to the CacheManager on construction, + * as a template, and then optionally apply any overrides previously defined for the named cache using the {@link + * #defineCache(String, org.horizon.config.Configuration)} method, or declared in the configuration file. + * + * @param cacheName name of cache to retrieve + * @return a cache instance identified by cacheName + */ + Cache getCache(String cacheName); + + /** + * @return the name of the cluster. Null if running in local mode. + */ + String getClusterName(); + + List<Address> getMembers(); + + Address getAddress(); +} Copied: core/branches/flat/src/main/java/org/horizon/manager/DefaultCacheManager.java (from rev 7505, core/branches/flat/src/main/java/org/horizon/manager/CacheManager.java) =================================================================== --- core/branches/flat/src/main/java/org/horizon/manager/DefaultCacheManager.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/manager/DefaultCacheManager.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,349 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2008, Red Hat Middleware LLC, and individual contributors + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.horizon.manager; + +import org.horizon.Cache; +import org.horizon.CacheSPI; +import org.horizon.config.Configuration; +import org.horizon.config.ConfigurationException; +import org.horizon.config.GlobalConfiguration; +import org.horizon.config.parsing.XmlConfigurationParser; +import org.horizon.config.parsing.XmlConfigurationParserImpl; +import org.horizon.factories.DefaultCacheFactory; +import org.horizon.factories.GlobalComponentRegistry; +import org.horizon.lifecycle.Lifecycle; +import org.horizon.notifications.CacheManagerNotifier; +import org.horizon.remoting.RPCManager; +import org.horizon.remoting.transport.Address; + +import java.io.IOException; +import java.io.InputStream; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; + +/** + * A <tt>CacheManager</tt> is the primary mechanism for retrieving a {@link Cache} instance, and is often used as a + * starting point to using the {@link Cache}. + * <p/> + * <tt>CacheManager</tt>s are heavyweight objects, and we foresee no more than one <tt>CacheManager</tt> being used per + * JVM (unless specific configuration requirements require more than one; but either way, this would be a minimal and + * finite number of instances). + * <p/> + * Constructing a <tt>CacheManager</tt> is done via one of its constructors, which optionally take in a {@link + * org.horizon.config.Configuration} or a path or URL to a configuration XML file. + * <p/> + * Lifecycle - <tt>CacheManager</tt>s have a lifecycle (it implements {@link Lifecycle}) and the default constructors + * also call {@link #start()}. Overloaded versions of the constructors are available, that do not start the + * <tt>CacheManager</tt>, although it must be kept in mind that <tt>CacheManager</tt>s need to be started before they + * can be used to create <tt>Cache</tt> instances. + * <p/> + * Once constructed, <tt>CacheManager</tt>s should be made available to any component that requires a <tt>Cache</tt>, + * via JNDI or via some other mechanism such as an IoC container. + * <p/> + * You obtain <tt>Cache</tt> instances from the <tt>CacheManager</tt> by using one of the overloaded + * <tt>getCache()</tt>, methods. Note that with <tt>getCache()</tt>, there is no guarantee that the instance you get is + * brand-new and empty, since caches are named and shared. Because of this, the <tt>CacheManager</tt> also acts as a + * repository of <tt>Cache</tt>s, and is an effective mechanism of looking up or creating <tt>Cache</tt>s on demand. + * <p/> + * When the system shuts down, it should call {@link #stop()} on the <tt>CacheManager</tt>. This will ensure all caches + * within its scope are properly stopped as well. + * <p/> + * Sample usage: <code> CacheManager manager = CacheManager.getInstance("my-config-file.xml"); Cache entityCache = + * manager.getCache("myEntityCache"); entityCache.put("aPerson", new Person()); + * <p/> + * Configuration myNewConfiguration = new Configuration(); myNewConfiguration.setCacheMode(Configuration.CacheMode.LOCAL); + * manager.defineCache("myLocalCache", myNewConfiguration); Cache localCache = manager.getCache("myLocalCache"); + * </code> + * + * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) + * @since 1.0 + */ +public class DefaultCacheManager implements CacheManager { + public static final String DEFAULT_CACHE_NAME = "org.jboss.starobrno.manager.DefaultCacheManager.DEFAULT_CACHE_NAME"; + protected GlobalConfiguration globalConfiguration; + private final ConcurrentMap<String, Cache> caches = new ConcurrentHashMap<String, Cache>(); + private final ConcurrentMap<String, Configuration> configurationOverrides = new ConcurrentHashMap<String, Configuration>(); + private GlobalComponentRegistry globalComponentRegistry; + + + /** + * Constructs and starts a default instance of the CacheManager, using configuration defaults. + */ + public DefaultCacheManager() { + this(null, null, true); + } + + /** + * Constructs a default instance of the CacheManager, using configuration defaults. + * + * @param start if true, the cache manager is started + */ + public DefaultCacheManager(boolean start) { + this(null, null, start); + } + + /** + * Constructs and starts a new instance of the CacheManager, using the default configuration passed in. Uses + * defaults for a {@link org.horizon.config.GlobalConfiguration}. + * + * @param defaultConfiguration configuration to use as a template for all caches created + */ + public DefaultCacheManager(Configuration defaultConfiguration) { + this(null, defaultConfiguration, true); + } + + /** + * Constructs a new instance of the CacheManager, using the default configuration passed in. Uses defaults for a + * {@link org.horizon.config.GlobalConfiguration}. + * + * @param defaultConfiguration configuration file to use as a template for all caches created + * @param start if true, the cache manager is started + */ + public DefaultCacheManager(Configuration defaultConfiguration, boolean start) { + this(null, defaultConfiguration, start); + } + + /** + * Constructs and starts a new instance of the CacheManager, using the global configuration passed in, and system + * defaults for the default named cache configuration. + * + * @param globalConfiguration GlobalConfiguration to use for all caches created + */ + public DefaultCacheManager(GlobalConfiguration globalConfiguration) { + this(globalConfiguration, null, true); + } + + /** + * Constructs a new instance of the CacheManager, using the global configuration passed in, and system defaults for + * the default named cache configuration. + * + * @param globalConfiguration GlobalConfiguration to use for all caches created + * @param start if true, the cache manager is started. + */ + public DefaultCacheManager(GlobalConfiguration globalConfiguration, boolean start) { + this(globalConfiguration, null, start); + } + + /** + * Constructs a new instance of the CacheManager, using the global and default configurations passed in. If either + * of these are null, system defaults are used. + * + * @param globalConfiguration global configuration to use. If null, a default instance is created. + * @param defaultConfiguration default configuration to use. If null, a default instance is created. + * @param start if true, the cache manager is started + */ + public DefaultCacheManager(GlobalConfiguration globalConfiguration, Configuration defaultConfiguration, boolean start) { + this.globalConfiguration = globalConfiguration == null ? new GlobalConfiguration() : globalConfiguration.clone(); + this.globalConfiguration.setDefaultConfiguration(defaultConfiguration == null ? new Configuration() : defaultConfiguration.clone()); + globalComponentRegistry = new GlobalComponentRegistry(globalConfiguration); + if (start) start(); + } + + + /** + * Constructs and starts a new instance of the CacheManager, using the configuration file name passed in. This + * constructor first searches for the named file on the classpath, and failing that, treats the file name as an + * absolute path. + * + * @param configurationFile name of configuration file to use as a template for all caches created + * @throws java.io.IOException if there is a problem with the configuration file. + */ + public DefaultCacheManager(String configurationFile) throws IOException { + this(configurationFile, true); + } + + /** + * Constructs a new instance of the CacheManager, using the configuration file name passed in. This constructor + * first searches for the named file on the classpath, and failing that, treats the file name as an absolute path. + * + * @param configurationFile name of configuration file to use as a template for all caches created + * @param start if true, the cache manager is started + * @throws java.io.IOException if there is a problem with the configuration file. + */ + public DefaultCacheManager(String configurationFile, boolean start) throws IOException { + try { + initialize(new XmlConfigurationParserImpl(configurationFile)); + } + catch (RuntimeException re) { + throw new ConfigurationException(re); + } + if (start) start(); + } + + /** + * Constructs and starts a new instance of the CacheManager, using the input stream passed in to read configuration + * file contents. + * + * @param configurationStream stream containing configuration file contents, to use as a template for all caches + * created + * @throws java.io.IOException if there is a problem with the configuration stream. + */ + public DefaultCacheManager(InputStream configurationStream) throws IOException { + this(configurationStream, true); + } + + /** + * Constructs a new instance of the CacheManager, using the input stream passed in to read configuration file + * contents. + * + * @param configurationStream stream containing configuration file contents, to use as a template for all caches + * created + * @param start if true, the cache manager is started + * @throws java.io.IOException if there is a problem reading the configuration stream + */ + public DefaultCacheManager(InputStream configurationStream, boolean start) throws IOException { + try { + initialize(new XmlConfigurationParserImpl(configurationStream)); + } + catch (RuntimeException re) { + throw new ConfigurationException(re); + } + if (start) start(); + } + + private void initialize(XmlConfigurationParser initializedParser) { + this.globalConfiguration = initializedParser.parseGlobalConfiguration(); + configurationOverrides.putAll(initializedParser.parseNamedConfigurations()); + } + + /** + * Defines a named cache. Named caches can be defined by using this method, in which case the configuration passed + * in is used to override the default configuration used when this cache manager instance was created. + * <p/> + * The other way to define named caches is declaratively, in the XML file passed in to the cache manager. + * <p/> + * A combination of approaches may also be used, provided the names do not conflict. + * + * @param cacheName name of cache to define + * @param configurationOverride configuration overrides to use + * @throws CacheNameExistsException if the name is already in use. + */ + public void defineCache(String cacheName, Configuration configurationOverride) throws CacheNameExistsException { + if (cacheName == null || configurationOverrides == null) + throw new NullPointerException("Null arguments not allowed"); + if (cacheName.equals(DEFAULT_CACHE_NAME)) + throw new IllegalArgumentException("Cache name cannot be used as it is a reserved, internal name"); + if (configurationOverrides.putIfAbsent(cacheName, configurationOverride) != null) + throw new CacheNameExistsException("Cache name [" + cacheName + "] already in use!"); + } + + + /** + * Retrieves the default cache associated with this cache manager. Note that the default cache does not need to be + * explicitly created with {@link #createCache(String)} since it is automatically created lazily when first used. + * <p/> + * As such, this method is always guaranteed to return the default cache. + * + * @return the default cache. + */ + public Cache getCache() { + return getCache(DEFAULT_CACHE_NAME); + } + + /** + * Retrieves a named cache from the system. If the cache has been previously created with the same name, the running + * cache instance is returned. Otherwise, this method attempts to create the cache first. + * <p/> + * When creating a new cache, this method will use the configuration passed in to the CacheManager on construction, + * as a template, and then optionally apply any overrides previously defined for the named cache using the {@link + * #defineCache(String, org.horizon.config.Configuration)} method, or declared in the configuration file. + * + * @param cacheName name of cache to retrieve + * @return a cache instance identified by cacheName + */ + public Cache getCache(String cacheName) { + if (cacheName == null) + throw new NullPointerException("Null arguments not allowed"); + + if (caches.containsKey(cacheName)) + return caches.get(cacheName); + + return createCache(cacheName); + } + + public String getClusterName() { + return globalConfiguration.getClusterName(); + } + + public List<Address> getMembers() { + return globalComponentRegistry.getComponent(RPCManager.class).getMembers(); + } + + public Address getAddress() { + return globalComponentRegistry.getComponent(RPCManager.class).getAddress(); + } + + private Cache createCache(String cacheName) { + Configuration c = globalConfiguration.getDefaultConfiguration().clone(); + if (!cacheName.equals(DEFAULT_CACHE_NAME)) { + Configuration overrides = configurationOverrides.get(cacheName); + if (overrides != null) c.applyOverrides(overrides); + } + + Cache cache = new DefaultCacheFactory().createCache(c, globalComponentRegistry, cacheName); + Cache other = caches.putIfAbsent(cacheName, cache); + if (other == null) { + cache.start(); + return cache; + } else { + return other; + } + } + + public void start() { + // get a hold of the "default" cache to start this? + CacheSPI defaultCache = (CacheSPI) getCache(); + globalComponentRegistry = defaultCache.getComponentRegistry().getSharedComponentRegistry(); + } + + public void stop() { + // make sure we stop the default cache LAST! + Cache defaultCache = null; + for (Map.Entry<String, Cache> entry : caches.entrySet()) { + if (entry.getKey().equals(DEFAULT_CACHE_NAME)) { + defaultCache = entry.getValue(); + } else { + entry.getValue().stop(); + } + } + + if (defaultCache != null) defaultCache.stop(); + } + + public void addListener(Object listener) { + CacheManagerNotifier notifier = globalComponentRegistry.getComponent(CacheManagerNotifier.class); + notifier.addListener(listener); + } + + public void removeListener(Object listener) { + CacheManagerNotifier notifier = globalComponentRegistry.getComponent(CacheManagerNotifier.class); + notifier.removeListener(listener); + } + + public Set<Object> getListeners() { + CacheManagerNotifier notifier = globalComponentRegistry.getComponent(CacheManagerNotifier.class); + return notifier.getListeners(); + } +} Property changes on: core/branches/flat/src/main/java/org/horizon/manager/DefaultCacheManager.java ___________________________________________________________________ Name: svn:keywords + Id Revision Name: svn:eol-style + LF Modified: core/branches/flat/src/main/java/org/horizon/marshall/CacheMarshallerStarobrno.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/marshall/CacheMarshallerStarobrno.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/marshall/CacheMarshallerStarobrno.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -30,13 +30,13 @@ import org.horizon.io.ExposedByteArrayOutputStream; import org.horizon.logging.Log; import org.horizon.logging.LogFactory; +import org.horizon.remoting.transport.Address; +import org.horizon.remoting.transport.jgroups.JGroupsAddress; import org.horizon.transaction.GlobalTransaction; import org.horizon.util.FastCopyHashMap; import org.horizon.util.Immutables; import org.jboss.util.NotImplementedException; import org.jboss.util.stream.MarshalledValueInputStream; -import org.jgroups.Address; -import org.jgroups.stack.IpAddress; import org.jgroups.util.Buffer; import java.io.ByteArrayInputStream; @@ -60,7 +60,7 @@ protected static final int MAGICNUMBER_METHODCALL = 1; protected static final int MAGICNUMBER_FQN = 2; protected static final int MAGICNUMBER_GTX = 3; - protected static final int MAGICNUMBER_IPADDRESS = 4; + protected static final int MAGICNUMBER_JG_ADDRESS = 4; protected static final int MAGICNUMBER_ARRAY_LIST = 5; protected static final int MAGICNUMBER_INTEGER = 6; protected static final int MAGICNUMBER_LONG = 7; @@ -158,9 +158,9 @@ out.writeByte(MAGICNUMBER_GTX); if (useRefs) writeReference(out, createReference(o, refMap)); marshallGlobalTransaction((GlobalTransaction) o, out, refMap); - } else if (o instanceof IpAddress) { - out.writeByte(MAGICNUMBER_IPADDRESS); - marshallIpAddress((IpAddress) o, out); + } else if (o instanceof JGroupsAddress) { + out.writeByte(MAGICNUMBER_JG_ADDRESS); + marshallJGroupsAddress((JGroupsAddress) o, out); } else if (o.getClass().equals(ArrayList.class)) { out.writeByte(MAGICNUMBER_ARRAY_LIST); marshallCollection((Collection) o, out, refMap); @@ -251,8 +251,8 @@ marshallObject(globalTransaction.getAddress(), out, refMap); } - private void marshallIpAddress(IpAddress ipAddress, ObjectOutputStream out) throws Exception { - ipAddress.writeExternal(out); + private void marshallJGroupsAddress(JGroupsAddress address, ObjectOutputStream out) throws Exception { + address.writeExternal(out); } @SuppressWarnings("unchecked") @@ -323,8 +323,8 @@ retVal = unmarshallGlobalTransaction(in, refMap); if (useRefs) refMap.putReferencedObject(reference, retVal); return retVal; - case MAGICNUMBER_IPADDRESS: - retVal = unmarshallIpAddress(in); + case MAGICNUMBER_JG_ADDRESS: + retVal = unmarshallJGroupsAddress(in); return retVal; case MAGICNUMBER_ARRAY: return unmarshallArray(in, refMap); @@ -411,10 +411,10 @@ return gtx; } - private IpAddress unmarshallIpAddress(ObjectInputStream in) throws Exception { - IpAddress ipAddress = new IpAddress(); - ipAddress.readExternal(in); - return ipAddress; + private JGroupsAddress unmarshallJGroupsAddress(ObjectInputStream in) throws Exception { + JGroupsAddress address = new JGroupsAddress(); + address.readExternal(in); + return address; } private List unmarshallArrayList(ObjectInputStream in, UnmarshalledReferences refMap) throws Exception { Deleted: core/branches/flat/src/main/java/org/horizon/marshall/CommandAwareRpcDispatcher.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/marshall/CommandAwareRpcDispatcher.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/marshall/CommandAwareRpcDispatcher.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,250 +0,0 @@ -/* - * JBoss, Home of Professional Open Source. - * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors - * as indicated by the @author tags. See the copyright.txt file in the - * distribution for a full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.horizon.marshall; - -import org.horizon.commands.ReplicableCommand; -import org.horizon.commands.VisitableCommand; -import org.horizon.config.Configuration; -import org.horizon.context.InvocationContext; -import org.horizon.factories.ComponentRegistry; -import org.horizon.interceptors.InterceptorChain; -import org.horizon.invocation.InvocationContextContainer; -import org.horizon.util.concurrent.BoundedExecutors; -import org.horizon.util.concurrent.WithinThreadExecutor; -import org.jgroups.Address; -import org.jgroups.Channel; -import org.jgroups.MembershipListener; -import org.jgroups.Message; -import org.jgroups.MessageListener; -import org.jgroups.blocks.RpcDispatcher; -import org.jgroups.blocks.RspFilter; -import org.jgroups.util.Buffer; -import org.jgroups.util.Rsp; -import org.jgroups.util.RspList; - -import java.io.NotSerializableException; -import java.util.Vector; -import java.util.concurrent.Callable; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Future; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; - -/** - * A JGroups RPC dispatcher that knows how to deal with {@link ReplicableCommand}s. - * - * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) - * @since 1.0 - */ -public class CommandAwareRpcDispatcher extends RpcDispatcher { - protected InvocationContextContainer invocationContextContainer; - protected InterceptorChain interceptorChain; - protected ComponentRegistry componentRegistry; - protected boolean trace; - private ExecutorService replicationProcessor; - private AtomicInteger replicationProcessorCount; - private boolean asyncSerial; - - public CommandAwareRpcDispatcher() { - } - - public CommandAwareRpcDispatcher(Channel channel, MessageListener l, MembershipListener l2, Object serverObj, - InvocationContextContainer container, InterceptorChain interceptorChain, - ComponentRegistry componentRegistry) { - super(channel, l, l2, serverObj); - this.invocationContextContainer = container; - this.componentRegistry = componentRegistry; - this.interceptorChain = interceptorChain; - trace = log.isTraceEnabled(); - - // what sort of a repl processor do we need? - Configuration c = componentRegistry.getComponent(Configuration.class); - replicationProcessor = c.getRuntimeConfig().getAsyncSerializationExecutor(); - if (c.getCacheMode().isSynchronous() || - (replicationProcessor == null && c.getSerializationExecutorPoolSize() < 1)) // if an executor has not been injected and the pool size is set - { - // in-process thread. Not async. - replicationProcessor = new WithinThreadExecutor(); - asyncSerial = false; - } else { - asyncSerial = true; - if (replicationProcessor == null) { - replicationProcessorCount = new AtomicInteger(0); - replicationProcessor = BoundedExecutors.newFixedThreadPool(c.isUseReplQueue() ? 1 : c.getSerializationExecutorPoolSize(), - new ThreadFactory() { - public Thread newThread(Runnable r) { - return new Thread(r, "AsyncReplicationProcessor-" + replicationProcessorCount.incrementAndGet()); - } - }, c.getSerializationExecutorQueueSize() - ); - } - } - } - - @Override - public void stop() { - replicationProcessor.shutdownNow(); - try { - replicationProcessor.awaitTermination(60, TimeUnit.SECONDS); - } - catch (InterruptedException e) { - Thread.currentThread().interrupt(); - } - super.stop(); - } - - protected boolean isValid(Message req) { - if (req == null || req.getLength() == 0) { - log.error("message or message buffer is null"); - return false; - } - - return true; - } - - /** - * Similar to {@link #callRemoteMethods(java.util.Vector, org.jgroups.blocks.MethodCall, int, long, boolean, boolean, - * org.jgroups.blocks.RspFilter)} except that this version is aware of {@link ReplicableCommand} objects. - */ - public RspList invokeRemoteCommands(Vector<Address> dests, ReplicableCommand command, int mode, long timeout, - boolean oob, RspFilter filter) throws NotSerializableException, ExecutionException, InterruptedException { - if (dests != null && dests.isEmpty()) { - // don't send if dest list is empty - if (trace) log.trace("Destination list is empty: no need to send message"); - return new RspList(); - } - - if (trace) - log.trace(new StringBuilder("dests=").append(dests).append(", command=").append(command).append(", mode="). - append(mode).append(", timeout=").append(timeout)); - - ReplicationTask replicationTask = new ReplicationTask(command, oob, dests, mode, timeout, false, filter); - Future<RspList> response = replicationProcessor.submit(replicationTask); - if (asyncSerial) { - // don't care about the response. return. - return null; - } else { - RspList retval = response.get(); - if (retval.isEmpty() || containsOnlyNulls(retval)) - return null; - else - return retval; - } - } - - private boolean containsOnlyNulls(RspList l) { - for (Rsp r : l.values()) { - if (r.getValue() != null || !r.wasReceived() || r.wasSuspected()) return false; - } - return true; - } - - /** - * Message contains a Command. Execute it against *this* object and return result. - */ - @Override - public Object handle(Message req) { - if (isValid(req)) { - try { - return executeCommand((ReplicableCommand) req_marshaller.objectFromByteBuffer(req.getBuffer(), req.getOffset(), req.getLength()), req); - } - catch (Throwable x) { - if (trace) log.trace("Problems invoking command.", x); - return x; - } - } else { - return null; - } - } - - protected Object executeCommand(ReplicableCommand cmd, Message req) throws Throwable { - if (cmd == null) throw new NullPointerException("Unable to execute a null command! Message was " + req); - if (trace) log.trace("Executing command: " + cmd + " [sender=" + req.getSrc() + "]"); - - if (cmd instanceof VisitableCommand) { - InvocationContext ctx = invocationContextContainer.get(); - ctx.setOriginLocal(false); - if (!componentRegistry.invocationsAllowed(false)) { - return null; - } - return interceptorChain.invoke(ctx, (VisitableCommand) cmd); - } else { - if (trace) log.trace("This is a non-visitable command - so performing directly and not via the invoker."); - - // need to check cache status for all except buddy replication commands. - if (!componentRegistry.invocationsAllowed(false)) return null; - - return cmd.perform(null); - } - } - - @Override - public String toString() { - return getClass().getSimpleName() + "[Outgoing marshaller: " + req_marshaller + "; incoming marshaller: " + rsp_marshaller + "]"; - } - - private class ReplicationTask implements Callable<RspList> { - private ReplicableCommand command; - private boolean oob; - private Vector<Address> dests; - private int mode; - private long timeout; - private boolean anycasting; - private RspFilter filter; - - private ReplicationTask(ReplicableCommand command, boolean oob, Vector<Address> dests, int mode, long timeout, boolean anycasting, RspFilter filter) { - this.command = command; - this.oob = oob; - this.dests = dests; - this.mode = mode; - this.timeout = timeout; - this.anycasting = anycasting; - this.filter = filter; - } - - public RspList call() throws Exception { - Buffer buf; - try { - buf = req_marshaller.objectToBuffer(command); - } - catch (Exception e) { - throw new RuntimeException("Failure to marshal argument(s)", e); - } - - Message msg = new Message(); - msg.setBuffer(buf); - if (oob) msg.setFlag(Message.OOB); - RspList retval = castMessage(dests, msg, mode, timeout, anycasting, filter); - if (trace) log.trace("responses: " + retval); - - // a null response is 99% likely to be due to a marshalling problem - we throw a NSE, this needs to be changed when - // JGroups supports http://jira.jboss.com/jira/browse/JGRP-193 - // the serialization problem could be on the remote end and this is why we cannot catch this above, when marshalling. - if (retval == null) - throw new NotSerializableException("RpcDispatcher returned a null. This is most often caused by args for " - + command.getClass().getSimpleName() + " not being serializable."); - return retval; - } - } -} \ No newline at end of file Added: core/branches/flat/src/main/java/org/horizon/notifications/CacheManagerNotifier.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/CacheManagerNotifier.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/notifications/CacheManagerNotifier.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,19 @@ +package org.horizon.notifications; + +import org.horizon.remoting.transport.Address; + +import java.util.List; + +/** + * Notifications for the cache manager + * + * @author Manik Surtani + * @since 1.0 + */ +public interface CacheManagerNotifier extends Listenable { + /** + * Notifies all registered listeners of a viewChange event. Note that viewChange notifications are ALWAYS sent + * immediately. + */ + void notifyViewChange(List<Address> members, Address myAddress); +} Added: core/branches/flat/src/main/java/org/horizon/notifications/CacheManagerNotifierImpl.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/CacheManagerNotifierImpl.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/notifications/CacheManagerNotifierImpl.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,36 @@ +package org.horizon.notifications; + +import org.horizon.factories.annotations.NonVolatile; +import org.horizon.factories.scopes.Scope; +import org.horizon.factories.scopes.Scopes; +import org.horizon.remoting.transport.Address; + +import java.util.List; +import java.util.Set; + +/** + * Global, shared notifications. Typically on the CacheManager. See + * + * @author Manik Surtani + * @since 1.0 + */ +@NonVolatile +(a)Scope(Scopes.GLOBAL) +public class CacheManagerNotifierImpl implements CacheManagerNotifier { + + public void notifyViewChange(List<Address> members, Address myAddress) { + // TODO: Manik: Customise this generated block + } + + public void addListener(Object listener) { + // TODO: Manik: Customise this generated block + } + + public void removeListener(Object listener) { + // TODO: Manik: Customise this generated block + } + + public Set<Object> getListeners() { + return null; // TODO: Manik: Customise this generated block + } +} Copied: core/branches/flat/src/main/java/org/horizon/notifications/CacheNotifier.java (from rev 7505, core/branches/flat/src/main/java/org/horizon/notifications/Notifier.java) =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/CacheNotifier.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/notifications/CacheNotifier.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,103 @@ +/* + * JBoss, Home of Professional Open Source. + * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.horizon.notifications; + +import org.horizon.context.InvocationContext; + +import javax.transaction.Transaction; + +/** + * Public interface with all allowed notifications. + * + * @author Mircea.Markus(a)jboss.com + * @since 1.0 + */ +public interface CacheNotifier extends Listenable { + /** + * Notifies all registered listeners of a nodeCreated event. + */ + void notifyCacheEntryCreated(Object key, boolean pre, InvocationContext ctx); + + /** + * Notifies all registered listeners of a nodeModified event. + */ + void notifyCacheEntryModified(Object key, boolean pre, InvocationContext ctx); + + /** + * When notifying about node modifications, in many scenarios there is a need of building a new Map object. If no + * listeners are registered for notification then it is pointless building this object - so guard the notification + * with this call. + */ + boolean shouldNotifyOnNodeModified(); + + /** + * Notifies all registered listeners of a nodeRemoved event. + */ + void notifyCacheEntryRemoved(Object key, boolean pre, InvocationContext ctx); + + /** + * Notifies all registered listeners of a nodeVisited event. + */ + void notifyCacheEntryVisited(Object key, boolean pre, InvocationContext ctx); + + /** + * Notifies all registered listeners of a nodeEvicted event. + */ + void notifyCacheEntryEvicted(Object key, boolean pre, InvocationContext ctx); + + /** + * Notifies all registered listeners of a nodeInvalidated event. + */ + void notifyCacheEntryInvalidated(Object key, boolean pre, InvocationContext ctx); + + /** + * Notifies all registered listeners of a nodeLoaded event. + */ + void notifyCacheEntryLoaded(Object key, boolean pre, InvocationContext ctx); + + /** + * Notifies all registered listeners of a nodeActivated event. + */ + void notifyCacheEntryActivated(Object key, boolean pre, InvocationContext ctx); + + /** + * Notifies all registered listeners of a nodePassivated event. + */ + void notifyCacheEntryPassivated(Object key, boolean pre, InvocationContext ctx); + + /** + * Notifies all registered listeners of a transaction completion event. + * + * @param transaction the transaction that has just completed + * @param successful if true, the transaction committed. If false, this is a rollback event + */ + void notifyTransactionCompleted(Transaction transaction, boolean successful, InvocationContext ctx); + + /** + * Notifies all registered listeners of a transaction registration event. + * + * @param transaction the transaction that has just completed + */ + void notifyTransactionRegistered(Transaction transaction, InvocationContext ctx); + + void notifyNodeInvalidated(Object key, boolean pre, InvocationContext ctx); +} \ No newline at end of file Property changes on: core/branches/flat/src/main/java/org/horizon/notifications/CacheNotifier.java ___________________________________________________________________ Name: svn:keywords + Id Revision Name: svn:eol-style + LF Copied: core/branches/flat/src/main/java/org/horizon/notifications/CacheNotifierImpl.java (from rev 7505, core/branches/flat/src/main/java/org/horizon/notifications/NotifierImpl.java) =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/CacheNotifierImpl.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/notifications/CacheNotifierImpl.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,538 @@ +/* + * JBoss, Home of Professional Open Source. + * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.horizon.notifications; + +import org.horizon.Cache; +import org.horizon.CacheException; +import org.horizon.CacheSPI; +import org.horizon.config.Configuration; +import org.horizon.context.InvocationContext; +import org.horizon.factories.KnownComponentNames; +import org.horizon.factories.annotations.ComponentName; +import org.horizon.factories.annotations.Destroy; +import org.horizon.factories.annotations.Inject; +import org.horizon.factories.annotations.NonVolatile; +import org.horizon.factories.annotations.Start; +import org.horizon.factories.annotations.Stop; +import org.horizon.factories.scopes.Scope; +import org.horizon.factories.scopes.Scopes; +import org.horizon.logging.Log; +import org.horizon.logging.LogFactory; +import org.horizon.notifications.annotation.*; +import org.horizon.notifications.event.*; +import static org.horizon.notifications.event.Event.Type.*; +import org.horizon.util.concurrent.WithinThreadExecutor; + +import javax.transaction.Transaction; +import java.lang.annotation.Annotation; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * Helper class that handles all notifications to registered listeners. + * + * @author <a href="mailto:manik@jboss.org">Manik Surtani (manik(a)jboss.org)&lt;/a&gt; + * @since 1.0 + */ +@NonVolatile +(a)Scope(Scopes.NAMED_CACHE) +public class CacheNotifierImpl implements CacheNotifier { + private static final Log log = LogFactory.getLog(CacheNotifierImpl.class); + + private static final Class[] allowedMethodAnnotations = + { + CacheStarted.class, CacheStopped.class, CacheEntryCreated.class, CacheEntryRemoved.class, CacheEntryVisited.class, CacheEntryModified.class, + CacheEntryActivated.class, CacheEntryPassivated.class, CacheEntryLoaded.class, CacheEntryEvicted.class, TransactionRegistered.class, TransactionCompleted.class, ViewChanged.class, + CacheEntryInvalidated.class + }; + private static final Class[] parameterTypes = + { + CacheStartedEvent.class, CacheStoppedEvent.class, CacheEntryCreatedEvent.class, CacheEntryRemovedEvent.class, CacheEntryVisitedEvent.class, CacheEntryModifiedEvent.class, + CacheEntryActivatedEvent.class, CacheEntryPassivatedEvent.class, CacheEntryLoadedEvent.class, CacheEntryEvictedEvent.class, TransactionRegisteredEvent.class, TransactionCompletedEvent.class, ViewChangedEvent.class, + CacheEntryInvalidatedEvent.class + }; + + final Map<Class<? extends Annotation>, List<ListenerInvocation>> listenersMap = new HashMap<Class<? extends Annotation>, List<ListenerInvocation>>(16, 0.99f); + final List<ListenerInvocation> cacheStartedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> cacheStoppedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> nodeCreatedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> nodeRemovedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> nodeVisitedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> nodeModifiedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> nodeActivatedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> nodePassivatedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> nodeLoadedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> nodeInvalidatedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> nodeEvictedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> transactionRegisteredListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> transactionCompletedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + final List<ListenerInvocation> viewChangedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); + + // final Map<Class, List<ListenerInvocation>> listenerInvocations = new ConcurrentHashMap<Class, List<ListenerInvocation>>(); + private Cache cache; + private boolean useMarshalledValueMaps; + private Configuration config; + // two separate executor services, one for sync and one for async listeners + private ExecutorService syncProcessor; + private ExecutorService asyncProcessor; + private static final AtomicInteger asyncNotifierThreadNumber = new AtomicInteger(0); + + public CacheNotifierImpl() { + listenersMap.put(CacheStarted.class, cacheStartedListeners); + listenersMap.put(CacheStopped.class, cacheStoppedListeners); + listenersMap.put(CacheEntryCreated.class, nodeCreatedListeners); + listenersMap.put(CacheEntryRemoved.class, nodeRemovedListeners); + listenersMap.put(CacheEntryVisited.class, nodeVisitedListeners); + listenersMap.put(CacheEntryModified.class, nodeModifiedListeners); + listenersMap.put(CacheEntryActivated.class, nodeActivatedListeners); + listenersMap.put(CacheEntryPassivated.class, nodePassivatedListeners); + listenersMap.put(CacheEntryLoaded.class, nodeLoadedListeners); + listenersMap.put(CacheEntryEvicted.class, nodeEvictedListeners); + listenersMap.put(TransactionRegistered.class, transactionRegisteredListeners); + listenersMap.put(TransactionCompleted.class, transactionCompletedListeners); + listenersMap.put(ViewChanged.class, viewChangedListeners); + listenersMap.put(CacheEntryInvalidated.class, nodeInvalidatedListeners); + } + + @Inject + void injectDependencies(CacheSPI cache, Configuration config, + @ComponentName(KnownComponentNames.ASYNC_NOTIFICATION_EXECUTOR) ExecutorService executor) { + this.cache = cache; + this.config = config; + this.asyncProcessor = executor; + } + + @Stop + void stop() { + syncProcessor.shutdownNow(); + asyncProcessor.shutdownNow(); + } + + @Destroy + void destroy() { + removeAllCacheListeners(); + } + + @Start + void start() { + useMarshalledValueMaps = config.isUseLazyDeserialization(); + syncProcessor = new WithinThreadExecutor(); + } + + /** + * Loops through all valid methods on the object passed in, and caches the relevant methods as {@link + * CacheNotifierImpl.ListenerInvocation} for invocation by reflection. + * + * @param listener object to be considered as a listener. + */ + @SuppressWarnings("unchecked") + private void validateAndAddListenerInvocation(Object listener) { + boolean sync = testListenerClassValidity(listener.getClass()); + + boolean foundMethods = false; + // now try all methods on the listener for anything that we like. Note that only PUBLIC methods are scanned. + for (Method m : listener.getClass().getMethods()) { + // loop through all valid method annotations + for (int i = 0; i < allowedMethodAnnotations.length; i++) { + if (m.isAnnotationPresent(allowedMethodAnnotations[i])) { + testListenerMethodValidity(m, parameterTypes[i], allowedMethodAnnotations[i].getName()); + addListenerInvocation(allowedMethodAnnotations[i], new ListenerInvocation(listener, m, sync)); + foundMethods = true; + } + } + } + + if (!foundMethods && log.isWarnEnabled()) + log.warn("Attempted to register listener of class " + listener.getClass() + ", but no valid, public methods annotated with method-level event annotations found! Ignoring listener."); + } + + /** + * Tests if a class is properly annotated as a CacheListener and returns whether callbacks on this class should be + * invoked synchronously or asynchronously. + * + * @param listenerClass class to inspect + * @return true if callbacks on this class should use the syncProcessor; false if it should use the asyncProcessor. + */ + private static boolean testListenerClassValidity(Class<?> listenerClass) { + Listener cl = listenerClass.getAnnotation(Listener.class); + if (cl == null) + throw new IncorrectCacheListenerException("Cache listener class MUST be annotated with org.horizon.notifications.annotation.Listener"); + if (!Modifier.isPublic(listenerClass.getModifiers())) + throw new IncorrectCacheListenerException("Cache listener class MUST be public!"); + return cl.sync(); + + } + + private static void testListenerMethodValidity(Method m, Class allowedParameter, String annotationName) { + if (m.getParameterTypes().length != 1 || !m.getParameterTypes()[0].isAssignableFrom(allowedParameter)) + throw new IncorrectCacheListenerException("Methods annotated with " + annotationName + " must accept exactly one parameter, of assignable from type " + allowedParameter.getName()); + if (!m.getReturnType().equals(void.class)) + throw new IncorrectCacheListenerException("Methods annotated with " + annotationName + " should have a return type of void."); + } + + private void addListenerInvocation(Class annotation, ListenerInvocation li) { + List<ListenerInvocation> result = getListenerCollectionForAnnotation(annotation); + result.add(li); + } + + public void addListener(Object listener) { + validateAndAddListenerInvocation(listener); + } + + public void removeListener(Object listener) { + for (Class annotation : allowedMethodAnnotations) removeListenerInvocation(annotation, listener); + } + + private void removeListenerInvocation(Class annotation, Object listener) { + if (listener == null) return; + List<ListenerInvocation> l = getListenerCollectionForAnnotation(annotation); + Set<Object> markedForRemoval = new HashSet<Object>(); + for (ListenerInvocation li : l) { + if (listener.equals(li.target)) markedForRemoval.add(li); + } + l.removeAll(markedForRemoval); + } + + /** + * Removes all listeners from the notifier, including the evictionPolicyListener. + */ + @Stop(priority = 99) + public void removeAllCacheListeners() { + cacheStartedListeners.clear(); + cacheStoppedListeners.clear(); + nodeCreatedListeners.clear(); + nodeRemovedListeners.clear(); + nodeVisitedListeners.clear(); + nodeModifiedListeners.clear(); + nodeActivatedListeners.clear(); + nodePassivatedListeners.clear(); + nodeLoadedListeners.clear(); + nodeEvictedListeners.clear(); + transactionRegisteredListeners.clear(); + transactionCompletedListeners.clear(); + viewChangedListeners.clear(); + } + + public Set<Object> getListeners() { + Set<Object> result = new HashSet<Object>(); + for (List<ListenerInvocation> list : listenersMap.values()) { + for (ListenerInvocation li : list) result.add(li.target); + } + return Collections.unmodifiableSet(result); + } + + public void notifyCacheEntryCreated(Object key, boolean pre, InvocationContext ctx) { + if (!nodeCreatedListeners.isEmpty()) { + boolean originLocal = ctx.isOriginLocal(); + Transaction tx = ctx.getTransaction(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setOriginLocal(originLocal); + e.setPre(pre); + e.setKey(key); + e.setTransaction(tx); + e.setType(CACHE_ENTRY_CREATED); + for (ListenerInvocation listener : nodeCreatedListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + public void notifyCacheEntryModified(Object key, boolean pre, InvocationContext ctx) { + if (!nodeModifiedListeners.isEmpty()) { + boolean originLocal = ctx.isOriginLocal(); + Transaction tx = ctx.getTransaction(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setOriginLocal(originLocal); + e.setPre(pre); + e.setKey(key); + e.setTransaction(tx); + e.setType(CACHE_ENTRY_MODIFIED); + for (ListenerInvocation listener : nodeModifiedListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + public boolean shouldNotifyOnNodeModified() { + return !nodeModifiedListeners.isEmpty(); + } + + public void notifyCacheEntryRemoved(Object key, boolean pre, InvocationContext ctx) { + if (!nodeRemovedListeners.isEmpty()) { + boolean originLocal = ctx.isOriginLocal(); + Transaction tx = ctx.getTransaction(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setOriginLocal(originLocal); + e.setPre(pre); + e.setKey(key); + e.setTransaction(tx); + e.setType(CACHE_ENTRY_REMOVED); + for (ListenerInvocation listener : nodeRemovedListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + public void notifyCacheEntryVisited(Object key, boolean pre, InvocationContext ctx) { + if (!nodeVisitedListeners.isEmpty()) { + Transaction tx = ctx.getTransaction(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setPre(pre); + e.setKey(key); + e.setTransaction(tx); + e.setType(CACHE_ENTRY_VISITED); + for (ListenerInvocation listener : nodeVisitedListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + public void notifyCacheEntryEvicted(final Object key, final boolean pre, InvocationContext ctx) { + if (!nodeEvictedListeners.isEmpty()) { + final boolean originLocal = ctx.isOriginLocal(); + Transaction tx = ctx.getTransaction(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setOriginLocal(originLocal); + e.setPre(pre); + e.setKey(key); + e.setTransaction(tx); + e.setType(CACHE_ENTRY_EVICTED); + for (ListenerInvocation listener : nodeEvictedListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + public void notifyCacheEntryInvalidated(final Object key, final boolean pre, InvocationContext ctx) { + if (!nodeInvalidatedListeners.isEmpty()) { + final boolean originLocal = ctx.isOriginLocal(); + Transaction tx = ctx.getTransaction(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setOriginLocal(originLocal); + e.setPre(pre); + e.setKey(key); + e.setTransaction(tx); + e.setType(CACHE_ENTRY_INVALIDATED); + for (ListenerInvocation listener : nodeInvalidatedListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + public void notifyCacheEntryLoaded(Object key, boolean pre, InvocationContext ctx) { + if (!nodeLoadedListeners.isEmpty()) { + boolean originLocal = ctx.isOriginLocal(); + Transaction tx = ctx.getTransaction(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setOriginLocal(originLocal); + e.setPre(pre); + e.setKey(key); + e.setTransaction(tx); + e.setType(CACHE_ENTRY_LOADED); + for (ListenerInvocation listener : nodeLoadedListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + public void notifyCacheEntryActivated(Object key, boolean pre, InvocationContext ctx) { + if (!nodeActivatedListeners.isEmpty()) { + boolean originLocal = ctx.isOriginLocal(); + Transaction tx = ctx.getTransaction(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setOriginLocal(originLocal); + e.setPre(pre); + e.setKey(key); + e.setTransaction(tx); + e.setType(CACHE_ENTRY_ACTIVATED); + for (ListenerInvocation listener : nodeActivatedListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + public void notifyCacheEntryPassivated(Object key, boolean pre, InvocationContext ctx) { + if (!nodePassivatedListeners.isEmpty()) { + Transaction tx = ctx.getTransaction(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setPre(pre); + e.setKey(key); + e.setTransaction(tx); + e.setType(CACHE_ENTRY_PASSIVATED); + for (ListenerInvocation listener : nodePassivatedListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + /** + * Notifies all registered listeners of a cacheStarted event. + */ + @Start(priority = 99) + public void notifyCacheStarted() { + if (!cacheStartedListeners.isEmpty()) { + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setType(CACHE_STARTED); + for (ListenerInvocation listener : cacheStartedListeners) listener.invoke(e); + } + } + + /** + * Notifies all registered listeners of a cacheStopped event. + */ + @Stop(priority = 98) + public void notifyCacheStopped() { + if (!cacheStoppedListeners.isEmpty()) { + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setType(CACHE_STOPPED); + for (ListenerInvocation listener : cacheStoppedListeners) listener.invoke(e); + } + } + + public void notifyTransactionCompleted(Transaction transaction, boolean successful, InvocationContext ctx) { + if (!transactionCompletedListeners.isEmpty()) { + boolean isOriginLocal = ctx.isOriginLocal(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setOriginLocal(isOriginLocal); + e.setTransaction(transaction); + e.setSuccessful(successful); + e.setType(TRANSACTION_COMPLETED); + for (ListenerInvocation listener : transactionCompletedListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + public void notifyTransactionRegistered(Transaction transaction, InvocationContext ctx) { + if (!transactionRegisteredListeners.isEmpty()) { + boolean isOriginLocal = ctx.isOriginLocal(); + InvocationContext backup = resetInvocationContext(ctx); + EventImpl e = new EventImpl(); + e.setCache(cache); + e.setOriginLocal(isOriginLocal); + e.setTransaction(transaction); + e.setType(TRANSACTION_REGISTERED); + for (ListenerInvocation listener : transactionRegisteredListeners) listener.invoke(e); + restoreInvocationContext(backup); + } + } + + private void restoreInvocationContext(InvocationContext backup) { + InvocationContext currentIC = cache.getInvocationContext(); + backup.clearLookedUpEntries(); + backup.putLookedUpEntries(currentIC.getLookedUpEntries()); + cache.setInvocationContext(backup); + } + + /** + * Resets the current (passed-in) invocation, and returns a temp InvocationContext containing its state so it can be + * restored later using {@link #restoreInvocationContext(InvocationContext)} + * + * @param ctx the current context to be reset + * @return a clone of ctx, before it was reset + */ + private InvocationContext resetInvocationContext(InvocationContext ctx) { + // wipe current context. + cache.setInvocationContext(null); + // get a new Invocation Context + InvocationContext newContext = cache.getInvocationContext(); + newContext.putLookedUpEntries(ctx.getLookedUpEntries()); + return ctx; + } + + /** + * Class that encapsulates a valid invocation for a given registered listener - containing a reference to the method + * to be invoked as well as the target object. + */ + class ListenerInvocation { + private final Object target; + private final Method method; + private final boolean sync; + + public ListenerInvocation(Object target, Method method, boolean sync) { + this.target = target; + this.method = method; + this.sync = sync; + } + + public void invoke(final Event e) { + Runnable r = new Runnable() { + + public void run() { + try { + method.invoke(target, e); + } + catch (InvocationTargetException exception) { + Throwable cause = exception.getCause(); + if (cause != null) + throw new CacheException("Caught exception invoking method " + method + " on listener instance " + target, cause); + else + throw new CacheException("Caught exception invoking method " + method + " on listener instance " + target, exception); + } + catch (IllegalAccessException exception) { + log.warn("Unable to invoke method " + method + " on Object instance " + target + " - removing this target object from list of listeners!", exception); + removeListener(target); + } + } + }; + + if (sync) + syncProcessor.execute(r); + else + asyncProcessor.execute(r); + + } + + } + + private List<ListenerInvocation> getListenerCollectionForAnnotation(Class<? extends Annotation> annotation) { + List<ListenerInvocation> list = listenersMap.get(annotation); + if (list == null) throw new CacheException("Unknown listener annotation: " + annotation); + return list; + } + + public void notifyNodeInvalidated(Object key, boolean pre, InvocationContext ctx) { + throw new UnsupportedOperationException("Not implemented");//todo please implement! + } +} Property changes on: core/branches/flat/src/main/java/org/horizon/notifications/CacheNotifierImpl.java ___________________________________________________________________ Name: svn:keywords + Id Revision Name: svn:eol-style + LF Modified: core/branches/flat/src/main/java/org/horizon/notifications/EventLog.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/EventLog.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/EventLog.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -8,7 +8,7 @@ import java.util.List; -@CacheListener +@Listener public class EventLog { public final List<Event> events = new ArrayList<Event>(); Added: core/branches/flat/src/main/java/org/horizon/notifications/Listenable.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/Listenable.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/notifications/Listenable.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,36 @@ +package org.horizon.notifications; + +import java.util.Set; + +/** + * Interface that denotes that the implementation can have listeners attached to it. + * + * @author Manik Surtani + * @since 1.0 + */ +public interface Listenable { + + /** + * Adds a listener to the component. Typically, listeners would need to be annotated with {@link + * org.horizon.notifications.annotation.Listener} and further to that, contain methods annotated appropriately, + * otherwise the listener will not be registered. + * <p/> + * See the {@link org.horizon.notifications.annotation.Listener} annotation for more information. + * <p/> + * + * @param listener must not be null. + */ + void addListener(Object listener); + + /** + * Removes a listener from the component. + * + * @param listener listener to remove. Must not be null. + */ + void removeListener(Object listener); + + /** + * @return a set of all listeners registered on this component. + */ + Set<Object> getListeners(); +} Deleted: core/branches/flat/src/main/java/org/horizon/notifications/Notifier.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/Notifier.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/Notifier.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,126 +0,0 @@ -/* - * JBoss, Home of Professional Open Source. - * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors - * as indicated by the @author tags. See the copyright.txt file in the - * distribution for a full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.horizon.notifications; - -import org.horizon.context.InvocationContext; -import org.jgroups.View; - -import javax.transaction.Transaction; -import java.util.Set; - -/** - * Public interface with all allowed notifications. - * - * @author Mircea.Markus(a)jboss.com - * @since 1.0 - */ -public interface Notifier { - /** - * Notifies all registered listeners of a nodeCreated event. - */ - void notifyCacheEntryCreated(Object key, boolean pre, InvocationContext ctx); - - /** - * Notifies all registered listeners of a nodeModified event. - */ - void notifyCacheEntryModified(Object key, boolean pre, InvocationContext ctx); - - /** - * When notifying about node modifications, in many scenarios there is a need of building a new Map object. If no - * listeners are registered for notification then it is pointless building this object - so guard the notification - * with this call. - */ - boolean shouldNotifyOnNodeModified(); - - /** - * Notifies all registered listeners of a nodeRemoved event. - */ - void notifyCacheEntryRemoved(Object key, boolean pre, InvocationContext ctx); - - /** - * Notifies all registered listeners of a nodeVisited event. - */ - void notifyCacheEntryVisited(Object key, boolean pre, InvocationContext ctx); - - /** - * Notifies all registered listeners of a nodeEvicted event. - */ - void notifyCacheEntryEvicted(Object key, boolean pre, InvocationContext ctx); - - /** - * Notifies all registered listeners of a nodeInvalidated event. - */ - void notifyCacheEntryInvalidated(Object key, boolean pre, InvocationContext ctx); - - /** - * Notifies all registered listeners of a nodeLoaded event. - */ - void notifyCacheEntryLoaded(Object key, boolean pre, InvocationContext ctx); - - /** - * Notifies all registered listeners of a nodeActivated event. - */ - void notifyCacheEntryActivated(Object key, boolean pre, InvocationContext ctx); - - /** - * Notifies all registered listeners of a nodePassivated event. - */ - void notifyCacheEntryPassivated(Object key, boolean pre, InvocationContext ctx); - - /** - * Notifies all registered listeners of a viewChange event. Note that viewChange notifications are ALWAYS sent - * immediately. - */ - void notifyViewChange(View view, InvocationContext ctx); - - /** - * Notifies all registered listeners of a transaction completion event. - * - * @param transaction the transaction that has just completed - * @param successful if true, the transaction committed. If false, this is a rollback event - */ - void notifyTransactionCompleted(Transaction transaction, boolean successful, InvocationContext ctx); - - /** - * Notifies all registered listeners of a transaction registration event. - * - * @param transaction the transaction that has just completed - */ - void notifyTransactionRegistered(Transaction transaction, InvocationContext ctx); - - /** - * Adds a cache listener to the list of cache listeners registered. - */ - void addCacheListener(Object listener); - - /** - * Removes a cache listener from the list of cache listeners registered. - */ - void removeCacheListener(Object listener); - - /** - * @return Retrieves an (unmodifiable) set of cache listeners registered. - */ - Set<Object> getCacheListeners(); - - void notifyNodeInvalidated(Object key, boolean pre, InvocationContext ctx); -} \ No newline at end of file Added: core/branches/flat/src/main/java/org/horizon/notifications/Notifier.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/Notifier.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/notifications/Notifier.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,11 @@ +package org.horizon.notifications; + +/** + * Common notifier methods + * + * @author Manik Surtani + * @since 1.0 + */ +public interface Notifier { + +} Deleted: core/branches/flat/src/main/java/org/horizon/notifications/NotifierImpl.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/NotifierImpl.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/NotifierImpl.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,567 +0,0 @@ -/* - * JBoss, Home of Professional Open Source. - * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors - * as indicated by the @author tags. See the copyright.txt file in the - * distribution for a full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.horizon.notifications; - -import org.horizon.Cache; -import org.horizon.CacheException; -import org.horizon.CacheSPI; -import org.horizon.config.Configuration; -import org.horizon.context.InvocationContext; -import org.horizon.factories.annotations.Destroy; -import org.horizon.factories.annotations.Inject; -import org.horizon.factories.annotations.NonVolatile; -import org.horizon.factories.annotations.Start; -import org.horizon.factories.annotations.Stop; -import org.horizon.factories.scopes.Scope; -import org.horizon.factories.scopes.Scopes; -import org.horizon.logging.Log; -import org.horizon.logging.LogFactory; -import org.horizon.notifications.annotation.*; -import org.horizon.notifications.event.*; -import static org.horizon.notifications.event.Event.Type.*; -import org.horizon.util.concurrent.BoundedExecutors; -import org.horizon.util.concurrent.WithinThreadExecutor; -import org.jgroups.View; - -import javax.transaction.Transaction; -import java.lang.annotation.Annotation; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; -import java.lang.reflect.Modifier; -import java.util.Collections; -import java.util.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.ThreadFactory; -import java.util.concurrent.atomic.AtomicInteger; - -/** - * Helper class that handles all notifications to registered listeners. - * - * @author <a href="mailto:manik@jboss.org">Manik Surtani (manik(a)jboss.org)&lt;/a&gt; - * @since 1.0 - */ -@NonVolatile -(a)Scope(Scopes.NAMED_CACHE) -public class NotifierImpl implements Notifier { - private static final Log log = LogFactory.getLog(NotifierImpl.class); - - private static final Class emptyMap = Collections.emptyMap().getClass(); - - private static final Class singletonMap = Collections.singletonMap(null, null).getClass(); - private static final Class[] allowedMethodAnnotations = - { - CacheStarted.class, CacheStopped.class, CacheEntryCreated.class, CacheEntryRemoved.class, CacheEntryVisited.class, CacheEntryModified.class, - CacheEntryActivated.class, CacheEntryPassivated.class, CacheEntryLoaded.class, CacheEntryEvicted.class, TransactionRegistered.class, TransactionCompleted.class, ViewChanged.class, - CacheEntryInvalidated.class - }; - private static final Class[] parameterTypes = - { - CacheStartedEvent.class, CacheStoppedEvent.class, CacheEntryCreatedEvent.class, CacheEntryRemovedEvent.class, CacheEntryVisitedEvent.class, CacheEntryModifiedEvent.class, - CacheEntryActivatedEvent.class, CacheEntryPassivatedEvent.class, CacheEntryLoadedEvent.class, CacheEntryEvictedEvent.class, TransactionRegisteredEvent.class, TransactionCompletedEvent.class, ViewChangedEvent.class, - CacheEntryInvalidatedEvent.class - }; - - final Map<Class<? extends Annotation>, List<ListenerInvocation>> listenersMap = new HashMap<Class<? extends Annotation>, List<ListenerInvocation>>(16, 0.99f); - final List<ListenerInvocation> cacheStartedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> cacheStoppedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> nodeCreatedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> nodeRemovedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> nodeVisitedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> nodeModifiedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> nodeActivatedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> nodePassivatedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> nodeLoadedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> nodeInvalidatedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> nodeEvictedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> transactionRegisteredListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> transactionCompletedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - final List<ListenerInvocation> viewChangedListeners = new CopyOnWriteArrayList<ListenerInvocation>(); - - // final Map<Class, List<ListenerInvocation>> listenerInvocations = new ConcurrentHashMap<Class, List<ListenerInvocation>>(); - private Cache cache; - private boolean useMarshalledValueMaps; - private Configuration config; - // two separate executor services, one for sync and one for async listeners - private ExecutorService syncProcessor; - private ExecutorService asyncProcessor; - private static final AtomicInteger asyncNotifierThreadNumber = new AtomicInteger(0); - - public NotifierImpl() { - listenersMap.put(CacheStarted.class, cacheStartedListeners); - listenersMap.put(CacheStopped.class, cacheStoppedListeners); - listenersMap.put(CacheEntryCreated.class, nodeCreatedListeners); - listenersMap.put(CacheEntryRemoved.class, nodeRemovedListeners); - listenersMap.put(CacheEntryVisited.class, nodeVisitedListeners); - listenersMap.put(CacheEntryModified.class, nodeModifiedListeners); - listenersMap.put(CacheEntryActivated.class, nodeActivatedListeners); - listenersMap.put(CacheEntryPassivated.class, nodePassivatedListeners); - listenersMap.put(CacheEntryLoaded.class, nodeLoadedListeners); - listenersMap.put(CacheEntryEvicted.class, nodeEvictedListeners); - listenersMap.put(TransactionRegistered.class, transactionRegisteredListeners); - listenersMap.put(TransactionCompleted.class, transactionCompletedListeners); - listenersMap.put(ViewChanged.class, viewChangedListeners); - listenersMap.put(CacheEntryInvalidated.class, nodeInvalidatedListeners); - } - - @Inject - void injectDependencies(CacheSPI cache, Configuration config) { - this.cache = cache; - this.config = config; - } - - @Stop - void stop() { - syncProcessor.shutdownNow(); - asyncProcessor.shutdownNow(); - } - - @Destroy - void destroy() { - removeAllCacheListeners(); - } - - @Start - void start() { - useMarshalledValueMaps = config.isUseLazyDeserialization(); - syncProcessor = new WithinThreadExecutor(); - - // first try and use an injected executor for async listeners - if ((asyncProcessor = config.getRuntimeConfig().getAsyncCacheListenerExecutor()) == null) { - // create one if needed - if (config.getListenerAsyncPoolSize() > 0) { - asyncProcessor = BoundedExecutors.newFixedThreadPool(config.getListenerAsyncPoolSize(), new ThreadFactory() { - public Thread newThread(Runnable r) { - return new Thread(r, "AsyncNotifier-" + asyncNotifierThreadNumber.getAndIncrement()); - } - }, config.getListenerAsyncQueueSize()); - } else { - // use the same sync executor - asyncProcessor = syncProcessor; - } - } - } - - /** - * Loops through all valid methods on the object passed in, and caches the relevant methods as {@link - * NotifierImpl.ListenerInvocation} for invocation by reflection. - * - * @param listener object to be considered as a listener. - */ - @SuppressWarnings("unchecked") - private void validateAndAddListenerInvocation(Object listener) { - boolean sync = testListenerClassValidity(listener.getClass()); - - boolean foundMethods = false; - // now try all methods on the listener for anything that we like. Note that only PUBLIC methods are scanned. - for (Method m : listener.getClass().getMethods()) { - // loop through all valid method annotations - for (int i = 0; i < allowedMethodAnnotations.length; i++) { - if (m.isAnnotationPresent(allowedMethodAnnotations[i])) { - testListenerMethodValidity(m, parameterTypes[i], allowedMethodAnnotations[i].getName()); - addListenerInvocation(allowedMethodAnnotations[i], new ListenerInvocation(listener, m, sync)); - foundMethods = true; - } - } - } - - if (!foundMethods && log.isWarnEnabled()) - log.warn("Attempted to register listener of class " + listener.getClass() + ", but no valid, public methods annotated with method-level event annotations found! Ignoring listener."); - } - - /** - * Tests if a class is properly annotated as a CacheListener and returns whether callbacks on this class should be - * invoked synchronously or asynchronously. - * - * @param listenerClass class to inspect - * @return true if callbacks on this class should use the syncProcessor; false if it should use the asyncProcessor. - */ - private static boolean testListenerClassValidity(Class<?> listenerClass) { - CacheListener cl = listenerClass.getAnnotation(CacheListener.class); - if (cl == null) - throw new IncorrectCacheListenerException("Cache listener class MUST be annotated with org.horizon.notifications.annotation.CacheListener"); - if (!Modifier.isPublic(listenerClass.getModifiers())) - throw new IncorrectCacheListenerException("Cache listener class MUST be public!"); - return cl.sync(); - - } - - private static void testListenerMethodValidity(Method m, Class allowedParameter, String annotationName) { - if (m.getParameterTypes().length != 1 || !m.getParameterTypes()[0].isAssignableFrom(allowedParameter)) - throw new IncorrectCacheListenerException("Methods annotated with " + annotationName + " must accept exactly one parameter, of assignable from type " + allowedParameter.getName()); - if (!m.getReturnType().equals(void.class)) - throw new IncorrectCacheListenerException("Methods annotated with " + annotationName + " should have a return type of void."); - } - - private void addListenerInvocation(Class annotation, ListenerInvocation li) { - List<ListenerInvocation> result = getListenerCollectionForAnnotation(annotation); - result.add(li); - } - - public void addCacheListener(Object listener) { - validateAndAddListenerInvocation(listener); - } - - public void removeCacheListener(Object listener) { - for (Class annotation : allowedMethodAnnotations) removeListenerInvocation(annotation, listener); - } - - private void removeListenerInvocation(Class annotation, Object listener) { - if (listener == null) return; - List<ListenerInvocation> l = getListenerCollectionForAnnotation(annotation); - Set<Object> markedForRemoval = new HashSet<Object>(); - for (ListenerInvocation li : l) { - if (listener.equals(li.target)) markedForRemoval.add(li); - } - l.removeAll(markedForRemoval); - } - - /** - * Removes all listeners from the notifier, including the evictionPolicyListener. - */ - @Stop(priority = 99) - public void removeAllCacheListeners() { - cacheStartedListeners.clear(); - cacheStoppedListeners.clear(); - nodeCreatedListeners.clear(); - nodeRemovedListeners.clear(); - nodeVisitedListeners.clear(); - nodeModifiedListeners.clear(); - nodeActivatedListeners.clear(); - nodePassivatedListeners.clear(); - nodeLoadedListeners.clear(); - nodeEvictedListeners.clear(); - transactionRegisteredListeners.clear(); - transactionCompletedListeners.clear(); - viewChangedListeners.clear(); - } - - public Set<Object> getCacheListeners() { - Set<Object> result = new HashSet<Object>(); - for (List<ListenerInvocation> list : listenersMap.values()) { - for (ListenerInvocation li : list) result.add(li.target); - } - return Collections.unmodifiableSet(result); - } - - public void notifyCacheEntryCreated(Object key, boolean pre, InvocationContext ctx) { - if (!nodeCreatedListeners.isEmpty()) { - boolean originLocal = ctx.isOriginLocal(); - Transaction tx = ctx.getTransaction(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setOriginLocal(originLocal); - e.setPre(pre); - e.setKey(key); - e.setTransaction(tx); - e.setType(CACHE_ENTRY_CREATED); - for (ListenerInvocation listener : nodeCreatedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - public void notifyCacheEntryModified(Object key, boolean pre, InvocationContext ctx) { - if (!nodeModifiedListeners.isEmpty()) { - boolean originLocal = ctx.isOriginLocal(); - Transaction tx = ctx.getTransaction(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setOriginLocal(originLocal); - e.setPre(pre); - e.setKey(key); - e.setTransaction(tx); - e.setType(CACHE_ENTRY_MODIFIED); - for (ListenerInvocation listener : nodeModifiedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - public boolean shouldNotifyOnNodeModified() { - return !nodeModifiedListeners.isEmpty(); - } - - public void notifyCacheEntryRemoved(Object key, boolean pre, InvocationContext ctx) { - if (!nodeRemovedListeners.isEmpty()) { - boolean originLocal = ctx.isOriginLocal(); - Transaction tx = ctx.getTransaction(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setOriginLocal(originLocal); - e.setPre(pre); - e.setKey(key); - e.setTransaction(tx); - e.setType(CACHE_ENTRY_REMOVED); - for (ListenerInvocation listener : nodeRemovedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - public void notifyCacheEntryVisited(Object key, boolean pre, InvocationContext ctx) { - if (!nodeVisitedListeners.isEmpty()) { - Transaction tx = ctx.getTransaction(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setPre(pre); - e.setKey(key); - e.setTransaction(tx); - e.setType(CACHE_ENTRY_VISITED); - for (ListenerInvocation listener : nodeVisitedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - public void notifyCacheEntryEvicted(final Object key, final boolean pre, InvocationContext ctx) { - if (!nodeEvictedListeners.isEmpty()) { - final boolean originLocal = ctx.isOriginLocal(); - Transaction tx = ctx.getTransaction(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setOriginLocal(originLocal); - e.setPre(pre); - e.setKey(key); - e.setTransaction(tx); - e.setType(CACHE_ENTRY_EVICTED); - for (ListenerInvocation listener : nodeEvictedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - public void notifyCacheEntryInvalidated(final Object key, final boolean pre, InvocationContext ctx) { - if (!nodeInvalidatedListeners.isEmpty()) { - final boolean originLocal = ctx.isOriginLocal(); - Transaction tx = ctx.getTransaction(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setOriginLocal(originLocal); - e.setPre(pre); - e.setKey(key); - e.setTransaction(tx); - e.setType(CACHE_ENTRY_INVALIDATED); - for (ListenerInvocation listener : nodeInvalidatedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - public void notifyCacheEntryLoaded(Object key, boolean pre, InvocationContext ctx) { - if (!nodeLoadedListeners.isEmpty()) { - boolean originLocal = ctx.isOriginLocal(); - Transaction tx = ctx.getTransaction(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setOriginLocal(originLocal); - e.setPre(pre); - e.setKey(key); - e.setTransaction(tx); - e.setType(CACHE_ENTRY_LOADED); - for (ListenerInvocation listener : nodeLoadedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - public void notifyCacheEntryActivated(Object key, boolean pre, InvocationContext ctx) { - if (!nodeActivatedListeners.isEmpty()) { - boolean originLocal = ctx.isOriginLocal(); - Transaction tx = ctx.getTransaction(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setOriginLocal(originLocal); - e.setPre(pre); - e.setKey(key); - e.setTransaction(tx); - e.setType(CACHE_ENTRY_ACTIVATED); - for (ListenerInvocation listener : nodeActivatedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - public void notifyCacheEntryPassivated(Object key, boolean pre, InvocationContext ctx) { - if (!nodePassivatedListeners.isEmpty()) { - Transaction tx = ctx.getTransaction(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setPre(pre); - e.setKey(key); - e.setTransaction(tx); - e.setType(CACHE_ENTRY_PASSIVATED); - for (ListenerInvocation listener : nodePassivatedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - /** - * Notifies all registered listeners of a cacheStarted event. - */ - @Start(priority = 99) - public void notifyCacheStarted() { - if (!cacheStartedListeners.isEmpty()) { - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setType(CACHE_STARTED); - for (ListenerInvocation listener : cacheStartedListeners) listener.invoke(e); - } - } - - /** - * Notifies all registered listeners of a cacheStopped event. - */ - @Stop(priority = 98) - public void notifyCacheStopped() { - if (!cacheStoppedListeners.isEmpty()) { - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setType(CACHE_STOPPED); - for (ListenerInvocation listener : cacheStoppedListeners) listener.invoke(e); - } - } - - public void notifyViewChange(final View newView, InvocationContext ctx) { - if (!viewChangedListeners.isEmpty()) { - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setNewView(newView); - e.setType(VIEW_CHANGED); - for (ListenerInvocation listener : viewChangedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - public void notifyTransactionCompleted(Transaction transaction, boolean successful, InvocationContext ctx) { - if (!transactionCompletedListeners.isEmpty()) { - boolean isOriginLocal = ctx.isOriginLocal(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setOriginLocal(isOriginLocal); - e.setTransaction(transaction); - e.setSuccessful(successful); - e.setType(TRANSACTION_COMPLETED); - for (ListenerInvocation listener : transactionCompletedListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - public void notifyTransactionRegistered(Transaction transaction, InvocationContext ctx) { - if (!transactionRegisteredListeners.isEmpty()) { - boolean isOriginLocal = ctx.isOriginLocal(); - InvocationContext backup = resetInvocationContext(ctx); - EventImpl e = new EventImpl(); - e.setCache(cache); - e.setOriginLocal(isOriginLocal); - e.setTransaction(transaction); - e.setType(TRANSACTION_REGISTERED); - for (ListenerInvocation listener : transactionRegisteredListeners) listener.invoke(e); - restoreInvocationContext(backup); - } - } - - private void restoreInvocationContext(InvocationContext backup) { - InvocationContext currentIC = cache.getInvocationContext(); - backup.clearLookedUpEntries(); - backup.putLookedUpEntries(currentIC.getLookedUpEntries()); - cache.setInvocationContext(backup); - } - - /** - * Resets the current (passed-in) invocation, and returns a temp InvocationContext containing its state so it can be - * restored later using {@link #restoreInvocationContext(InvocationContext)} - * - * @param ctx the current context to be reset - * @return a clone of ctx, before it was reset - */ - private InvocationContext resetInvocationContext(InvocationContext ctx) { - // wipe current context. - cache.setInvocationContext(null); - // get a new Invocation Context - InvocationContext newContext = cache.getInvocationContext(); - newContext.putLookedUpEntries(ctx.getLookedUpEntries()); - return ctx; - } - - /** - * Class that encapsulates a valid invocation for a given registered listener - containing a reference to the method - * to be invoked as well as the target object. - */ - class ListenerInvocation { - private final Object target; - private final Method method; - private final boolean sync; - - public ListenerInvocation(Object target, Method method, boolean sync) { - this.target = target; - this.method = method; - this.sync = sync; - } - - public void invoke(final Event e) { - Runnable r = new Runnable() { - - public void run() { - try { - method.invoke(target, e); - } - catch (InvocationTargetException exception) { - Throwable cause = exception.getCause(); - if (cause != null) - throw new CacheException("Caught exception invoking method " + method + " on listener instance " + target, cause); - else - throw new CacheException("Caught exception invoking method " + method + " on listener instance " + target, exception); - } - catch (IllegalAccessException exception) { - log.warn("Unable to invoke method " + method + " on Object instance " + target + " - removing this target object from list of listeners!", exception); - removeCacheListener(target); - } - } - }; - - if (sync) - syncProcessor.execute(r); - else - asyncProcessor.execute(r); - - } - - } - - private List<ListenerInvocation> getListenerCollectionForAnnotation(Class<? extends Annotation> annotation) { - List<ListenerInvocation> list = listenersMap.get(annotation); - if (list == null) throw new CacheException("Unknown listener annotation: " + annotation); - return list; - } - - public void notifyNodeInvalidated(Object key, boolean pre, InvocationContext ctx) { - throw new UnsupportedOperationException("Not implemented");//todo please implement! - } -} Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryActivated.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryActivated.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryActivated.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -34,7 +34,7 @@ * will be thrown when registering your cache listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @see org.horizon.notifications.annotation.NodePassivated * @since 1.0 */ Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryCreated.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryCreated.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryCreated.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -34,7 +34,7 @@ * will be thrown when registering your cache listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ @Retention(RetentionPolicy.RUNTIME) Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryEvicted.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryEvicted.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryEvicted.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -34,7 +34,7 @@ * will be thrown when registering your cache listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @see org.horizon.notifications.annotation.NodeLoaded * @since 1.0 */ Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryInvalidated.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryInvalidated.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryInvalidated.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -34,7 +34,7 @@ * will be thrown when registering your cache listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see org.horizon.notifications.annotation.CacheListener + * @see Listener * @since 1.0 */ @Retention(RetentionPolicy.RUNTIME) Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryLoaded.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryLoaded.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryLoaded.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -35,7 +35,7 @@ * will be thrown when registering your cache listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ @Retention(RetentionPolicy.RUNTIME) Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryModified.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryModified.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryModified.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -35,7 +35,7 @@ * <p/> * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ @Retention(RetentionPolicy.RUNTIME) Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryPassivated.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryPassivated.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryPassivated.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -34,7 +34,7 @@ * will be thrown when registering your listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ // ensure this annotation is available at runtime. Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryRemoved.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryRemoved.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryRemoved.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -34,7 +34,7 @@ * will be thrown when registering your listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ // ensure this annotation is available at runtime. Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryVisited.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryVisited.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheEntryVisited.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -34,7 +34,7 @@ * will be thrown when registering your listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ // ensure this annotation is available at runtime. Deleted: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheListener.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheListener.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheListener.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,246 +0,0 @@ -/* - * JBoss, Home of Professional Open Source. - * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors - * as indicated by the @author tags. See the copyright.txt file in the - * distribution for a full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.horizon.notifications.annotation; - -import java.lang.annotation.ElementType; -import java.lang.annotation.Retention; -import java.lang.annotation.RetentionPolicy; -import java.lang.annotation.Target; - -/** - * Class-level annotation used to annotate an object as being a valid cache listener. Used with the {@link - * org.horizon.Cache_Legacy#addCacheListener(Object)} and related APIs. <p/> Note that even if a class is annotated with - * this annotation, it still needs method-level annotation (such as {@link org.horizon.notifications.annotation.CacheStarted}) - * to actually receive notifications. <p/> Objects annotated with this annotation - listeners - can be attached to a - * running {@link org.horizon.Cache_Legacy} so users can be notified of {@link org.horizon.Cache_Legacy} events. <p/> - * <p/> There can be multiple methods that are annotated to receive the same event, and a method may receive multiple - * events by using a super type. </p> <p/> <h4>Delivery Semantics</h4> <p/> An event is delivered immediately after the - * respective operation, but before the underlying cache call returns. For this reason it is important to keep listener - * processing logic short-lived. If a long running task needs to be performed, it's recommended to use another thread. - * </p> <p/> <h4>Transactional Semantics</h4> <p/> Since the event is delivered during the actual cache call, the - * transactional outcome is not yet known. For this reason, <i>events are always delivered, even if the changes they - * represent are discarded by their containing transaction</i>. For applications that must only process events that - * represent changes in a completed transaction, {@link org.horizon.notifications.event.TransactionalEvent#getTransaction()} - * can be used, along with {@link org.horizon.notifications.event.TransactionCompletedEvent#isSuccessful()} to record - * events and later process them once the transaction has been successfully committed. Example 4 demonstrates this. </p> - * <p/> <h4>Threading Semantics</h4> <p/> A listener implementation must be capable of handling concurrent invocations. - * Local notifications reuse the calling thread; remote notifications reuse the network thread. </p> <p/> Since - * notifications reuse the calling or network thread, it is important to realise that if your listener implementation - * blocks or performs a long-running task, the original caller which triggered the cache event may block until the - * listener callback completes. It is therefore a good idea to use the listener to be notified of an event but to - * perform any long running tasks in a separate thread so as not to block the original caller. </p> <p/> In addition, - * any locks acquired for the operation being performed will still be held for the callback. This needs to be kept in - * mind as locks may be held longer than necessary or intended to and may cause deadlocking in certain situations. See - * above paragraph on long-running tasks that should be run in a separate thread. </p> <b>Note</b>: Since 3.0, a new - * parameter, <tt>sync</tt>, has been introduced on this annotation. This defaults to <tt>true</tt> which provides the - * above semantics. Alternatively, if you set <tt>sync</tt> to <tt>false</tt>, then invocations are made in a - * <i>separate</i> thread, which will not cause any blocking on the caller or network thread. The separate thread is - * taken from a pool, which can be configured using {@link org.horizon.config.Configuration#setListenerAsyncPoolSize(int)}. - * <p/> - * <b>Summary of Notification Annotations</b> <table border="1" cellpadding="1" cellspacing="1" summary="Summary of - * notification annotations"> <tr> <th bgcolor="#CCCCFF" align="left">Annotation</th> <th bgcolor="#CCCCFF" - * align="left">Event</th> <th bgcolor="#CCCCFF" align="left">Description</th> </tr> <tr> <td valign="top">{@link - * CacheStarted}</td> <td valign="top">{@link org.horizon.notifications.event.CacheStartedEvent}</td> <td valign="top">A - * cache was started</td> </tr> <tr> <td valign="top">{@link CacheStopped}</td> <td valign="top">{@link - * org.horizon.notifications.event.CacheStoppedEvent}</td> <td valign="top">A cache was stopped</td> </tr> <tr> <td - * valign="top">{@link CacheEntryModified}</td> <td valign="top">{@link org.horizon.notifications.event.NodeModifiedEvent}</td> - * <td valign="top">A node was modified</td> </tr> <tr> <td valign="top">{@link NodeMoved}</td> <td valign="top">{@link - * org.horizon.notifications.event.NodeMovedEvent}</td> <td valign="top">A node was moved</td> </tr> <tr> <td - * valign="top">{@link CacheEntryCreated}</td> <td valign="top">{@link org.horizon.notifications.event.NodeCreatedEvent}</td> - * <td valign="top">A node was created</td> </tr> <tr> <td valign="top">{@link CacheEntryRemoved}</td> <td - * valign="top">{@link org.horizon.notifications.event.NodeRemovedEvent}</td> <td valign="top">A node was removed</td> - * </tr> <tr> <td valign="top">{@link CacheEntryVisited}</td> <td valign="top">{@link - * org.horizon.notifications.event.NodeVisitedEvent}</td> <td valign="top">A node was visited</td> </tr> <tr> <td - * valign="top">{@link CacheEntryLoaded}</td> <td valign="top">{@link org.horizon.notifications.event.NodeLoadedEvent}</td> - * <td valign="top">A node was loaded</td> </tr> <tr> <td valign="top">{@link org.horizon.notifications.annotation.NodeEvicted}</td> - * <td valign="top">{@link org.horizon.notifications.event.NodeEvictedEvent}</td> <td valign="top">A node was - * evicted</td> </tr> <tr> <td valign="top">{@link org.horizon.notifications.annotation.NodeActivated}</td> <td - * valign="top">{@link org.horizon.notifications.event.NodeActivatedEvent}</td> <td valign="top">A node was - * activated</td> </tr> <tr> <td valign="top">{@link org.horizon.notifications.annotation.NodePassivated}</td> <td - * valign="top">{@link org.horizon.notifications.event.NodePassivatedEvent}</td> <td valign="top">A node was - * passivated</td> </tr> <tr> <td valign="top">{@link org.horizon.notifications.annotation.ViewChanged}</td> <td - * valign="top">{@link org.horizon.notifications.event.ViewChangedEvent}</td> <td valign="top">A view change event was - * detected</td> </tr> <tr> <td valign="top">{@link org.horizon.notifications.annotation.CacheBlocked}</td> <td - * valign="top">{@link org.horizon.notifications.event.CacheBlockedEvent}</td> <td valign="top">A cache block event was - * detected</td> </tr> <tr> <td valign="top">{@link CacheUnblocked}</td> <td valign="top">{@link - * org.horizon.notifications.event.CacheUnblockedEvent}</td> <td valign="top">A cache unblock event was detected</td> - * </tr> <tr> <td valign="top">{@link TransactionRegistered}</td> <td valign@="top">{@link - * org.horizon.notifications.event.TransactionRegisteredEvent}</td> <td valign="top">The cache has started to - * participate in a transaction</td> </tr> <tr> <td valign="top">{@link TransactionCompleted}</td> <td - * valign=@"top">{@link org.horizon.notifications.event.TransactionCompletedEvent}</td> <td valign="top">The cache has - * completed its participation in a transaction</td> </tr> <tr> <td valign="top">{@link BuddyGroupChanged}</td> <td - * valign=@"top">{@link org.horizon.notifications.event.BuddyGroupChangedEvent}</td> <td valign="top">Buddy replication - * is enabled and one of the buddy groups that the instance is a member of has changed its membership.</td> </tr> <tr> - * <td valign="top">{@link CacheEntryInvalidated}</td> <td valign=@"top">{@link org.horizon.notifications.event.NodeInvalidatedEvent}</td> - * <td valign="top">A node was invalidated by a remote cache. Only if cache mode is INVALIDATION_SYNC or - * INVALIDATION_ASYNC.</td> </tr> - * <p/> - * </table> - * <p/> - * <h4>Example 1 - Method receiving a single event</h4> - * <pre> - * &#064;CacheListener - * public class SingleEventListener - * { - * &#064;CacheStarted - * public void doSomething(Event event) - * { - * System.out.println(&quot;Cache started. Details = &quot; + event); - * } - * } - * </pre> - * <p/> - * <h4>Example 2 - Method receiving multiple events</h4> - * <pre> - * &#064;CacheListener - * public class MultipleEventListener - * { - * &#064;CacheStarted - * &#064;CacheStopped - * public void doSomething(Event event) - * { - * if (event.getType() == Event.Type.CACHE_STARTED) - * System.out.println(&quot;Cache started. Details = &quot; + event); - * else if (event.getType() == Event.Type.CACHE_STOPPED) - * System.out.println(&quot;Cache stopped. Details = &quot; + event); - * } - * } - * </pre> - * <p/> - * <h4>Example 3 - Multiple methods receiving the same event</h4> - * <pre> - * &#064;CAcheListener - * public class SingleEventListener - * { - * &#064;CacheStarted - * public void handleStart(Event event) - * { - * System.out.println(&quot;Cache started&quot;); - * } - * <p/> - * &#064;CacheStarted - * &#064;CacheStopped - * &#064;CacheBlocked - * &#064;CacheUnblocked - * &#064;ViewChanged - * public void logEvent(Event event) - * { - * logSystem.logEvent(event.getType()); - * } - * } - * </pre> - * <p/> - * <p/> - * <b>Example 4 - Processing only events with a committed transaction.</b> - * <p/> - * <pre> - * &#064;CacheListener - * public class TxGuaranteedListener - * { - * private class TxEventQueue - * { - * private ConcurrentMap&lt;Transaction, Queue&lt;Event&gt;&gt; map = new ConcurrentHashMap&lt;Transaction, - * Queue&lt;Event&gt;&gt;(); - * <p/> - * public void offer(Event event) - * { - * Queue&lt;Event&gt; queue = getQueue(event.getContext().getTransaction()); - * queue.offer(event); - * } - * <p/> - * private Queue&lt;Event&gt; getQueue(Transaction transaction) - * { - * Queue&lt;Event&gt; queue = map.get(transaction); - * if (queue == null) - * { - * queue = new ConcurrentLinkedQueue&lt;Event&gt;(); - * map.putIfAbsent(transaction, queue); - * } - * <p/> - * return queue; - * } - * <p/> - * public Queue&lt;Event&gt; takeAll(Transaction transaction) - * { - * return map.remove(transaction); - * } - * } - * <p/> - * private TxEventQueue events = new TxEventQueue(); - * <p/> - * &#064;NodeModified - * &#064;NodeMoved - * &#064;NodeCreated - * &#064;NodeRemoved - * public void handle(Event event) - * { - * events.offer(event); - * } - * <p/> - * &#064;TransactionCompleted - * public void handleTx(TransactionCompletedEvent event) - * { - * Queue&lt;Event&gt; completed = events.takeAll(event.getTransaction()); - * if (completed != null &amp;&amp; event.isSuccessful()) - * System.out.println("Comitted events = " + completed); - * } - * } - * </pre> - * - * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @author Jason T. Greene - * @see CacheStarted - * @see CacheStopped - * @see CacheEntryModified - * @see NodeMoved - * @see CacheEntryCreated - * @see CacheEntryRemoved - * @see CacheEntryVisited - * @see CacheEntryLoaded - * @see CacheEntryEvicted - * @see CacheEntryActivated - * @see CacheEntryPassivated - * @see ViewChanged - * @see CacheBlocked - * @see CacheUnblocked - * @see TransactionCompleted - * @see TransactionRegistered - * @see BuddyGroupChanged - * @see CacheEntryInvalidated - * @see org.horizon.Cache_Legacy#addCacheListener(Object) - * @see org.horizon.Cache_Legacy#removeCacheListener(Object) - * @see org.horizon.Cache_Legacy#getCacheListeners() - * @since 1.0 - */ -(a)Retention(RetentionPolicy.RUNTIME) -(a)Target(ElementType.TYPE) -public @interface CacheListener { - /** - * Specifies whether callbacks on any class annotated with this annotation happens synchronously (in the caller's - * thread) or asynchronously (using a separate thread). Defaults to <tt>true</tt>. - * - * @return true if the expectation is that callbacks are called using the caller's thread; false if they are to be - * made in a separate thread. - * @since 1.0 - */ - boolean sync() default true; -} Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheStarted.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheStarted.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheStarted.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -34,7 +34,7 @@ * will be thrown when registering your listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ @Retention(RetentionPolicy.RUNTIME) Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheStopped.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheStopped.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheStopped.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -34,7 +34,7 @@ * will be thrown when registering your listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ @Retention(RetentionPolicy.RUNTIME) Copied: core/branches/flat/src/main/java/org/horizon/notifications/annotation/Listener.java (from rev 7505, core/branches/flat/src/main/java/org/horizon/notifications/annotation/CacheListener.java) =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/Listener.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/Listener.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,239 @@ +/* + * JBoss, Home of Professional Open Source. + * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.horizon.notifications.annotation; + +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +/** + * Class-level annotation used to annotate an object as being a valid cache listener. Used with the {@link + * org.horizon.Cache_Legacy#addCacheListener(Object)} and related APIs. <p/> Note that even if a class is annotated with + * this annotation, it still needs method-level annotation (such as {@link org.horizon.notifications.annotation.CacheStarted}) + * to actually receive notifications. <p/> Objects annotated with this annotation - listeners - can be attached to a + * running {@link org.horizon.Cache_Legacy} so users can be notified of {@link org.horizon.Cache_Legacy} events. <p/> + * <p/> There can be multiple methods that are annotated to receive the same event, and a method may receive multiple + * events by using a super type. </p> <p/> <h4>Delivery Semantics</h4> <p/> An event is delivered immediately after the + * respective operation, but before the underlying cache call returns. For this reason it is important to keep listener + * processing logic short-lived. If a long running task needs to be performed, it's recommended to use another thread. + * </p> <p/> <h4>Transactional Semantics</h4> <p/> Since the event is delivered during the actual cache call, the + * transactional outcome is not yet known. For this reason, <i>events are always delivered, even if the changes they + * represent are discarded by their containing transaction</i>. For applications that must only process events that + * represent changes in a completed transaction, {@link org.horizon.notifications.event.TransactionalEvent#getTransaction()} + * can be used, along with {@link org.horizon.notifications.event.TransactionCompletedEvent#isSuccessful()} to record + * events and later process them once the transaction has been successfully committed. Example 4 demonstrates this. </p> + * <p/> <h4>Threading Semantics</h4> <p/> A listener implementation must be capable of handling concurrent invocations. + * Local notifications reuse the calling thread; remote notifications reuse the network thread. </p> <p/> Since + * notifications reuse the calling or network thread, it is important to realise that if your listener implementation + * blocks or performs a long-running task, the original caller which triggered the cache event may block until the + * listener callback completes. It is therefore a good idea to use the listener to be notified of an event but to + * perform any long running tasks in a separate thread so as not to block the original caller. </p> <p/> In addition, + * any locks acquired for the operation being performed will still be held for the callback. This needs to be kept in + * mind as locks may be held longer than necessary or intended to and may cause deadlocking in certain situations. See + * above paragraph on long-running tasks that should be run in a separate thread. </p> <b>Note</b>: Since 3.0, a new + * parameter, <tt>sync</tt>, has been introduced on this annotation. This defaults to <tt>true</tt> which provides the + * above semantics. Alternatively, if you set <tt>sync</tt> to <tt>false</tt>, then invocations are made in a + * <i>separate</i> thread, which will not cause any blocking on the caller or network thread. The separate thread is + * taken from a pool, which can be configured using {@link org.horizon.config.Configuration#setListenerAsyncPoolSize(int)}. + * <p/> + * <b>Summary of Notification Annotations</b> <table border="1" cellpadding="1" cellspacing="1" summary="Summary of + * notification annotations"> <tr> <th bgcolor="#CCCCFF" align="left">Annotation</th> <th bgcolor="#CCCCFF" + * align="left">Event</th> <th bgcolor="#CCCCFF" align="left">Description</th> </tr> <tr> <td valign="top">{@link + * CacheStarted}</td> <td valign="top">{@link org.horizon.notifications.event.CacheStartedEvent}</td> <td valign="top">A + * cache was started</td> </tr> <tr> <td valign="top">{@link CacheStopped}</td> <td valign="top">{@link + * org.horizon.notifications.event.CacheStoppedEvent}</td> <td valign="top">A cache was stopped</td> </tr> <tr> <td + * valign="top">{@link CacheEntryModified}</td> <td valign="top">{@link org.horizon.notifications.event.NodeModifiedEvent}</td> + * <td valign="top">A node was modified</td> </tr> <tr> <td valign="top">{@link NodeMoved}</td> <td valign="top">{@link + * org.horizon.notifications.event.NodeMovedEvent}</td> <td valign="top">A node was moved</td> </tr> <tr> <td + * valign="top">{@link CacheEntryCreated}</td> <td valign="top">{@link org.horizon.notifications.event.NodeCreatedEvent}</td> + * <td valign="top">A node was created</td> </tr> <tr> <td valign="top">{@link CacheEntryRemoved}</td> <td + * valign="top">{@link org.horizon.notifications.event.NodeRemovedEvent}</td> <td valign="top">A node was removed</td> + * </tr> <tr> <td valign="top">{@link CacheEntryVisited}</td> <td valign="top">{@link + * org.horizon.notifications.event.NodeVisitedEvent}</td> <td valign="top">A node was visited</td> </tr> <tr> <td + * valign="top">{@link CacheEntryLoaded}</td> <td valign="top">{@link org.horizon.notifications.event.NodeLoadedEvent}</td> + * <td valign="top">A node was loaded</td> </tr> <tr> <td valign="top">{@link org.horizon.notifications.annotation.NodeEvicted}</td> + * <td valign="top">{@link org.horizon.notifications.event.NodeEvictedEvent}</td> <td valign="top">A node was + * evicted</td> </tr> <tr> <td valign="top">{@link org.horizon.notifications.annotation.NodeActivated}</td> <td + * valign="top">{@link org.horizon.notifications.event.NodeActivatedEvent}</td> <td valign="top">A node was + * activated</td> </tr> <tr> <td valign="top">{@link org.horizon.notifications.annotation.NodePassivated}</td> <td + * valign="top">{@link org.horizon.notifications.event.NodePassivatedEvent}</td> <td valign="top">A node was + * passivated</td> </tr> <tr> <td valign="top">{@link org.horizon.notifications.annotation.ViewChanged}</td> <td + * valign="top">{@link org.horizon.notifications.event.ViewChangedEvent}</td> <td valign="top">A view change event was + * detected</td> </tr> <tr> <td valign="top">{@link org.horizon.notifications.annotation.CacheBlocked}</td> <td + * valign="top">{@link org.horizon.notifications.event.CacheBlockedEvent}</td> <td valign="top">A cache block event was + * detected</td> </tr> <tr> <td valign="top">{@link CacheUnblocked}</td> <td valign="top">{@link + * org.horizon.notifications.event.CacheUnblockedEvent}</td> <td valign="top">A cache unblock event was detected</td> + * </tr> <tr> <td valign="top">{@link TransactionRegistered}</td> <td valign@="top">{@link + * org.horizon.notifications.event.TransactionRegisteredEvent}</td> <td valign="top">The cache has started to + * participate in a transaction</td> </tr> <tr> <td valign="top">{@link TransactionCompleted}</td> <td + * valign=@"top">{@link org.horizon.notifications.event.TransactionCompletedEvent}</td> <td valign="top">The cache has + * completed its participation in a transaction</td> </tr> <tr> <td valign="top">{@link BuddyGroupChanged}</td> <td + * valign=@"top">{@link org.horizon.notifications.event.BuddyGroupChangedEvent}</td> <td valign="top">Buddy replication + * is enabled and one of the buddy groups that the instance is a member of has changed its membership.</td> </tr> <tr> + * <td valign="top">{@link CacheEntryInvalidated}</td> <td valign=@"top">{@link org.horizon.notifications.event.NodeInvalidatedEvent}</td> + * <td valign="top">A node was invalidated by a remote cache. Only if cache mode is INVALIDATION_SYNC or + * INVALIDATION_ASYNC.</td> </tr> + * <p/> + * </table> + * <p/> + * <h4>Example 1 - Method receiving a single event</h4> + * <pre> + * &#064;CacheListener + * public class SingleEventListener + * { + * &#064;CacheStarted + * public void doSomething(Event event) + * { + * System.out.println(&quot;Cache started. Details = &quot; + event); + * } + * } + * </pre> + * <p/> + * <h4>Example 2 - Method receiving multiple events</h4> + * <pre> + * &#064;CacheListener + * public class MultipleEventListener + * { + * &#064;CacheStarted + * &#064;CacheStopped + * public void doSomething(Event event) + * { + * if (event.getType() == Event.Type.CACHE_STARTED) + * System.out.println(&quot;Cache started. Details = &quot; + event); + * else if (event.getType() == Event.Type.CACHE_STOPPED) + * System.out.println(&quot;Cache stopped. Details = &quot; + event); + * } + * } + * </pre> + * <p/> + * <h4>Example 3 - Multiple methods receiving the same event</h4> + * <pre> + * &#064;CAcheListener + * public class SingleEventListener + * { + * &#064;CacheStarted + * public void handleStart(Event event) + * { + * System.out.println(&quot;Cache started&quot;); + * } + * <p/> + * &#064;CacheStarted + * &#064;CacheStopped + * &#064;CacheBlocked + * &#064;CacheUnblocked + * &#064;ViewChanged + * public void logEvent(Event event) + * { + * logSystem.logEvent(event.getType()); + * } + * } + * </pre> + * <p/> + * <p/> + * <b>Example 4 - Processing only events with a committed transaction.</b> + * <p/> + * <pre> + * &#064;CacheListener + * public class TxGuaranteedListener + * { + * private class TxEventQueue + * { + * private ConcurrentMap&lt;Transaction, Queue&lt;Event&gt;&gt; map = new ConcurrentHashMap&lt;Transaction, + * Queue&lt;Event&gt;&gt;(); + * <p/> + * public void offer(Event event) + * { + * Queue&lt;Event&gt; queue = getQueue(event.getContext().getTransaction()); + * queue.offer(event); + * } + * <p/> + * private Queue&lt;Event&gt; getQueue(Transaction transaction) + * { + * Queue&lt;Event&gt; queue = map.get(transaction); + * if (queue == null) + * { + * queue = new ConcurrentLinkedQueue&lt;Event&gt;(); + * map.putIfAbsent(transaction, queue); + * } + * <p/> + * return queue; + * } + * <p/> + * public Queue&lt;Event&gt; takeAll(Transaction transaction) + * { + * return map.remove(transaction); + * } + * } + * <p/> + * private TxEventQueue events = new TxEventQueue(); + * <p/> + * &#064;NodeModified + * &#064;NodeMoved + * &#064;NodeCreated + * &#064;NodeRemoved + * public void handle(Event event) + * { + * events.offer(event); + * } + * <p/> + * &#064;TransactionCompleted + * public void handleTx(TransactionCompletedEvent event) + * { + * Queue&lt;Event&gt; completed = events.takeAll(event.getTransaction()); + * if (completed != null &amp;&amp; event.isSuccessful()) + * System.out.println("Comitted events = " + completed); + * } + * } + * </pre> + * + * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> + * @author Jason T. Greene + * @see CacheStarted + * @see CacheStopped + * @see CacheEntryModified + * @see CacheEntryCreated + * @see CacheEntryRemoved + * @see CacheEntryVisited + * @see CacheEntryLoaded + * @see CacheEntryEvicted + * @see CacheEntryActivated + * @see CacheEntryPassivated + * @see ViewChanged + * @see TransactionCompleted + * @see TransactionRegistered + * @see CacheEntryInvalidated + * @since 1.0 + */ +(a)Retention(RetentionPolicy.RUNTIME) +(a)Target(ElementType.TYPE) +public @interface Listener { + /** + * Specifies whether callbacks on any class annotated with this annotation happens synchronously (in the caller's + * thread) or asynchronously (using a separate thread). Defaults to <tt>true</tt>. + * + * @return true if the expectation is that callbacks are called using the caller's thread; false if they are to be + * made in a separate thread. + * @since 1.0 + */ + boolean sync() default true; +} Property changes on: core/branches/flat/src/main/java/org/horizon/notifications/annotation/Listener.java ___________________________________________________________________ Name: svn:keywords + Id Revision Name: svn:eol-style + LF Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/TransactionCompleted.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/TransactionCompleted.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/TransactionCompleted.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -38,7 +38,7 @@ * be called with {@link org.horizon.notifications.event.Event#isPre()} being set to <tt>true</tt>. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ // ensure this annotation is available at runtime. Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/TransactionRegistered.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/TransactionRegistered.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/TransactionRegistered.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -39,7 +39,7 @@ * be called with {@link org.horizon.notifications.event.Event#isPre()} being set to <tt>true</tt>. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ // ensure this annotation is available at runtime. Modified: core/branches/flat/src/main/java/org/horizon/notifications/annotation/ViewChanged.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/annotation/ViewChanged.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/annotation/ViewChanged.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -35,7 +35,7 @@ * will be thrown when registering your listener. * * @author <a href="mailto:manik@jboss.org">Manik Surtani</a> - * @see CacheListener + * @see Listener * @since 1.0 */ // ensure this annotation is available at runtime. Modified: core/branches/flat/src/main/java/org/horizon/notifications/event/EventImpl.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/event/EventImpl.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/event/EventImpl.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -22,10 +22,11 @@ package org.horizon.notifications.event; import org.horizon.Cache; +import org.horizon.remoting.transport.Address; import org.horizon.tree.Fqn; -import org.jgroups.View; import javax.transaction.Transaction; +import java.util.List; /** * Basic implementation of an event that covers all event types. @@ -44,10 +45,11 @@ private boolean originLocal = true; // by default events all originate locally private Fqn targetFqn; private boolean successful; - private View newView; private Type type; + private List<Address> members; + Address localAddress; - public EventImpl(boolean pre, Cache cache, Object key, Transaction transaction, boolean originLocal, Fqn targetFqn, boolean successful, View newView, Type type) { + public EventImpl(boolean pre, Cache cache, Object key, Transaction transaction, boolean originLocal, Fqn targetFqn, boolean successful, List<Address> members, Address localAddress, Type type) { this.pre = pre; this.cache = cache; this.key = key; @@ -55,8 +57,9 @@ this.originLocal = originLocal; this.targetFqn = targetFqn; this.successful = successful; - this.newView = newView; this.type = type; + this.members = members; + this.localAddress = localAddress; } public EventImpl() { @@ -94,10 +97,6 @@ return successful; } - public View getNewView() { - return newView; - } - // ------------------------------ setters ----------------------------- public void setPre(boolean pre) { @@ -128,10 +127,6 @@ this.successful = successful; } - public void setNewView(View newView) { - this.newView = newView; - } - public void setType(Type type) { this.type = type; } @@ -148,30 +143,30 @@ if (successful != event.successful) return false; if (cache != null ? !cache.equals(event.cache) : event.cache != null) return false; if (key != null ? !key.equals(event.key) : event.key != null) return false; + if (localAddress != null ? !localAddress.equals(event.localAddress) : event.localAddress != null) return false; + if (members != null ? !members.equals(event.members) : event.members != null) return false; if (targetFqn != null ? !targetFqn.equals(event.targetFqn) : event.targetFqn != null) return false; if (transaction != null ? !transaction.equals(event.transaction) : event.transaction != null) return false; - if (newView != null ? !newView.equals(event.newView) : event.newView != null) return false; - if (type != null ? !type.equals(event.type) : event.type != null) return false; + if (type != event.type) return false; return true; } @Override public int hashCode() { - int result; - result = (pre ? 1 : 0); + int result = (pre ? 1 : 0); result = 31 * result + (cache != null ? cache.hashCode() : 0); result = 31 * result + (key != null ? key.hashCode() : 0); result = 31 * result + (transaction != null ? transaction.hashCode() : 0); result = 31 * result + (originLocal ? 1 : 0); result = 31 * result + (targetFqn != null ? targetFqn.hashCode() : 0); result = 31 * result + (successful ? 1 : 0); - result = 31 * result + (newView != null ? newView.hashCode() : 0); result = 31 * result + (type != null ? type.hashCode() : 0); + result = 31 * result + (members != null ? members.hashCode() : 0); + result = 31 * result + (localAddress != null ? localAddress.hashCode() : 0); return result; } - @Override public String toString() { return "EventImpl{" + @@ -183,7 +178,16 @@ ", originLocal=" + originLocal + ", targetFqn=" + targetFqn + ", successful=" + successful + - ", newView=" + newView + + ", members=" + members + + ", localAddress=" + localAddress + '}'; } + + public List<Address> getNewMemberList() { + return members; + } + + public Address getLocalAddress() { + return localAddress; + } } Modified: core/branches/flat/src/main/java/org/horizon/notifications/event/ViewChangedEvent.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/notifications/event/ViewChangedEvent.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/notifications/event/ViewChangedEvent.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -21,8 +21,10 @@ */ package org.horizon.notifications.event; -import org.jgroups.View; +import org.horizon.remoting.transport.Address; +import java.util.List; + /** * This event is passed in to any method annotated with {@link org.horizon.notifications.annotation.ViewChanged}. * @@ -33,5 +35,7 @@ /** * @return the new view associated with this view change. */ - View getNewView(); + List<Address> getNewMemberList(); + + Address getLocalAddress(); } Deleted: core/branches/flat/src/main/java/org/horizon/remoting/ChannelMessageListener.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/ChannelMessageListener.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/remoting/ChannelMessageListener.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,339 +0,0 @@ -/* - * JBoss, Home of Professional Open Source. - * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors - * as indicated by the @author tags. See the copyright.txt file in the - * distribution for a full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.horizon.remoting; - -import org.horizon.CacheException; -import org.horizon.config.Configuration; -import org.horizon.factories.annotations.Inject; -import org.horizon.factories.annotations.NonVolatile; -import org.horizon.io.ExposedByteArrayOutputStream; -import org.horizon.logging.Log; -import org.horizon.logging.LogFactory; -import org.horizon.statetransfer.DefaultStateTransferManager; -import org.horizon.statetransfer.StateTransferManager; -import org.horizon.tree.Fqn; -import org.jboss.util.stream.MarshalledValueInputStream; -import org.jboss.util.stream.MarshalledValueOutputStream; -import org.jgroups.ExtendedMessageListener; -import org.jgroups.Message; -import org.jgroups.util.Util; - -import java.io.ByteArrayInputStream; -import java.io.InputStream; -import java.io.OutputStream; - -/** - * JGroups MessageListener - * - * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) - * @since 1.0 - */ -@NonVolatile -public class ChannelMessageListener implements ExtendedMessageListener { - /** - * Reference to an exception that was raised during state installation on this node. - */ - protected volatile Exception setStateException; - private final Object stateLock = new Object(); - private static final Log log = LogFactory.getLog(ChannelMessageListener.class); - private static final boolean trace = log.isTraceEnabled(); - private StateTransferManager stateTransferManager; - private Configuration configuration; - /** - * True if state was initialized during start-up. - */ - private volatile boolean isStateSet = false; - - - @Inject - private void injectDependencies(StateTransferManager stateTransferManager, Configuration configuration) { - this.stateTransferManager = stateTransferManager; - this.configuration = configuration; - } - - public boolean isStateSet() { - return isStateSet; - } - - public void setStateSet(boolean stateSet) { - isStateSet = stateSet; - } - - public void waitForState() throws Exception { - synchronized (stateLock) { - while (!isStateSet) { - if (setStateException != null) { - throw setStateException; - } - - try { - stateLock.wait(); - } - catch (InterruptedException iex) { - } - } - } - } - - protected void stateReceivedSuccess() { - isStateSet = true; - setStateException = null; - } - - protected void stateReceivingFailed(Throwable t) { - if (t instanceof CacheException) { - log.debug(t); - } else { - log.error("failed setting state", t); - } - if (t instanceof Exception) { - setStateException = (Exception) t; - } else { - setStateException = new Exception(t); - } - } - - protected void stateProducingFailed(Throwable t) { - if (t instanceof CacheException) { - log.debug(t); - } else { - log.error("Caught " + t.getClass().getName() - + " while responding to state transfer request", t); - } - } - - /** - * Callback, does nothing. - */ - public void receive(Message msg) { - } - - public byte[] getState() { - MarshalledValueOutputStream out = null; - byte[] result; - ExposedByteArrayOutputStream baos = new ExposedByteArrayOutputStream(16 * 1024); - try { - out = new MarshalledValueOutputStream(baos); - - stateTransferManager.getState(out, Fqn.ROOT, configuration.getStateRetrievalTimeout(), true, true); - } - catch (Throwable t) { - stateProducingFailed(t); - } - finally { - result = baos.getRawBuffer(); - Util.close(out); - } - return result; - } - - public void setState(byte[] new_state) { - if (new_state == null) { - log.debug("transferred state is null (may be first member in cluster)"); - return; - } - ByteArrayInputStream bais = new ByteArrayInputStream(new_state); - MarshalledValueInputStream in = null; - try { - in = new MarshalledValueInputStream(bais); - stateTransferManager.setState(in, Fqn.ROOT); - stateReceivedSuccess(); - } - catch (Throwable t) { - stateReceivingFailed(t); - } - finally { - Util.close(in); - synchronized (stateLock) { - // Notify wait that state has been set. - stateLock.notifyAll(); - } - } - } - - public byte[] getState(String state_id) { - if (trace) log.trace("Getting state for state id " + state_id); - MarshalledValueOutputStream out = null; - String sourceRoot = state_id; - byte[] result; - - boolean hasDifferentSourceAndIntegrationRoots = state_id.indexOf(DefaultStateTransferManager.PARTIAL_STATE_DELIMITER) > 0; - if (hasDifferentSourceAndIntegrationRoots) { - sourceRoot = state_id.split(DefaultStateTransferManager.PARTIAL_STATE_DELIMITER)[0]; - } - - ExposedByteArrayOutputStream baos = new ExposedByteArrayOutputStream(16 * 1024); - try { - out = new MarshalledValueOutputStream(baos); - - stateTransferManager.getState(out, Fqn.fromString(sourceRoot), - configuration.getStateRetrievalTimeout(), true, true); - } - catch (Throwable t) { - stateProducingFailed(t); - } - finally { - result = baos.getRawBuffer(); - Util.close(out); - } - return result; - } - - public void getState(OutputStream ostream) { - MarshalledValueOutputStream out = null; - try { - out = new MarshalledValueOutputStream(ostream); - stateTransferManager.getState(out, Fqn.ROOT, configuration.getStateRetrievalTimeout(), true, true); - } - catch (Throwable t) { - stateProducingFailed(t); - } - finally { - Util.close(out); - } - } - - public void getState(String state_id, OutputStream ostream) { - if (trace) log.trace("Getting state for state id " + state_id); - String sourceRoot = state_id; - MarshalledValueOutputStream out = null; - boolean hasDifferentSourceAndIntegrationRoots = state_id.indexOf(DefaultStateTransferManager.PARTIAL_STATE_DELIMITER) > 0; - if (hasDifferentSourceAndIntegrationRoots) { - sourceRoot = state_id.split(DefaultStateTransferManager.PARTIAL_STATE_DELIMITER)[0]; - } - try { - out = new MarshalledValueOutputStream(ostream); - stateTransferManager.getState(out, Fqn.fromString(sourceRoot), configuration.getStateRetrievalTimeout(), true, true); - } - catch (Throwable t) { - stateProducingFailed(t); - } - finally { - Util.close(out); - } - } - - public void setState(InputStream istream) { - if (istream == null) { - log.debug("stream is null (may be first member in cluster)"); - return; - } - MarshalledValueInputStream in = null; - try { - in = new MarshalledValueInputStream(istream); - stateTransferManager.setState(in, Fqn.ROOT); - stateReceivedSuccess(); - } - catch (Throwable t) { - stateReceivingFailed(t); - } - finally { - Util.close(in); - synchronized (stateLock) { - // Notify wait that state has been set. - stateLock.notifyAll(); - } - } - } - - public void setState(String state_id, byte[] state) { - if (trace) log.trace("Receiving state for " + state_id); - if (state == null) { - log.debug("partial transferred state is null"); - return; - } - - MarshalledValueInputStream in = null; - String targetRoot = state_id; - boolean hasDifferentSourceAndIntegrationRoots = state_id.indexOf(DefaultStateTransferManager.PARTIAL_STATE_DELIMITER) > 0; - if (hasDifferentSourceAndIntegrationRoots) { - targetRoot = state_id.split(DefaultStateTransferManager.PARTIAL_STATE_DELIMITER)[1]; - } - try { - log.debug("Setting received partial state for subroot " + state_id); - Fqn subroot = Fqn.fromString(targetRoot); -// Region region = evictionCacheManager.getRegion(subroot, false); -// ClassLoader cl = null; -// if (region != null) -// { -// // If a classloader is registered for the node's region, use it -// cl = region.getClassLoader(); -// } - ByteArrayInputStream bais = new ByteArrayInputStream(state); - in = new MarshalledValueInputStream(bais); - //getStateTransferManager().setState(in, subroot, cl); - stateTransferManager.setState(in, subroot); - stateReceivedSuccess(); - } - catch (Throwable t) { - stateReceivingFailed(t); - } - finally { - Util.close(in); - synchronized (stateLock) { - // Notify wait that state has been set. - stateLock.notifyAll(); - } - } - } - - public void setState(String stateId, InputStream istream) { - if (trace) log.trace("Receiving state for " + stateId); - String targetRoot = stateId; - MarshalledValueInputStream in = null; - boolean hasDifferentSourceAndIntegrationRoots = stateId.indexOf(DefaultStateTransferManager.PARTIAL_STATE_DELIMITER) > 0; - if (hasDifferentSourceAndIntegrationRoots) { - targetRoot = stateId.split(DefaultStateTransferManager.PARTIAL_STATE_DELIMITER)[1]; - } - if (istream == null) { - log.debug("stream is null (may be first member in cluster). State is not set"); - return; - } - - try { - log.debug("Setting received partial state for subroot " + stateId); - in = new MarshalledValueInputStream(istream); - Fqn subroot = Fqn.fromString(targetRoot); -// Region region = evictionCacheManager.getRegion(subroot, false); -// ClassLoader cl = null; -// if (region != null) -// { -// // If a classloader is registered for the node's region, use it -// cl = region.getClassLoader(); -// } - //getStateTransferManager().setState(in, subroot, cl); - stateTransferManager.setState(in, subroot); - stateReceivedSuccess(); - } - catch (Throwable t) { - if (log.isTraceEnabled()) log.trace("Unknown error while integrating state", t); - stateReceivingFailed(t); - } - finally { - Util.close(in); - synchronized (stateLock) { - // Notify wait that state has been set. - stateLock.notifyAll(); - } - } - } -} \ No newline at end of file Added: core/branches/flat/src/main/java/org/horizon/remoting/InboundInvocationHandler.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/InboundInvocationHandler.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/remoting/InboundInvocationHandler.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,20 @@ +package org.horizon.remoting; + +import org.horizon.commands.ReplicableCommand; + +/** + * // TODO: Manik: Document this! + * + * @author Manik Surtani + * @since 1.0 + */ +public interface InboundInvocationHandler { + + /** + * Invokes a command on the cache, from a remote source. + * + * @param command command to invoke + * @return results, if any, from the invocation + */ + Object handle(ReplicableCommand command); +} Added: core/branches/flat/src/main/java/org/horizon/remoting/InboundInvocationHandlerImpl.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/InboundInvocationHandlerImpl.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/remoting/InboundInvocationHandlerImpl.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,39 @@ +package org.horizon.remoting; + +import org.horizon.commands.ReplicableCommand; +import org.horizon.factories.ComponentRegistry; +import org.horizon.interceptors.InterceptorChain; +import org.horizon.invocation.InvocationContextContainer; + +/** + * // TODO: Manik: Document this! + * + * @author Manik Surtani + * @since 1.0 + */ +public class InboundInvocationHandlerImpl implements InboundInvocationHandler { + InvocationContextContainer invocationContextContainer; + ComponentRegistry componentRegistry; + InterceptorChain interceptorChain; + + + public Object handle(ReplicableCommand command) { + + throw new RuntimeException("Implement me!"); +// if (cmd instanceof VisitableCommand) { +// InvocationContext ctx = invocationContextContainer.get(); +// ctx.setOriginLocal(false); +// if (!componentRegistry.invocationsAllowed(false)) { +// return null; +// } +// return interceptorChain.invoke(ctx, (VisitableCommand) command); +// } else { +// if (trace) log.trace("This is a non-visitable command - so performing directly and not via the invoker."); +// +// // need to check cache status for all except buddy replication commands. +// if (!componentRegistry.invocationsAllowed(false)) return null; +// +// return cmd.perform(null); +// } + } +} Modified: core/branches/flat/src/main/java/org/horizon/remoting/RPCManager.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/RPCManager.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/remoting/RPCManager.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -22,89 +22,71 @@ package org.horizon.remoting; import org.horizon.commands.ReplicableCommand; +import org.horizon.factories.annotations.NonVolatile; import org.horizon.factories.scopes.Scope; import org.horizon.factories.scopes.Scopes; -import org.jgroups.Address; -import org.jgroups.Channel; -import org.jgroups.blocks.RspFilter; +import org.horizon.lifecycle.Lifecycle; +import org.horizon.remoting.transport.Address; import java.util.List; -import java.util.Vector; /** - * Provides a mechanism for communicating with other caches in the cluster. For now this is based on JGroups as an - * underlying transport, and in future more transport options may become available. + * Provides a mechanism for communicating with other caches in the cluster. * <p/> - * Implementations have a simple lifecycle: <ul> <li>start() - starts the underlying channel based on configuration - * options injected, and connects the channel</li> <li>disconnect() - disconnects the channel</li> <li>stop() - stops - * the dispatcher and releases resources</li> </ul> + * Implementations have a simple lifecycle: <ul> <li>start() - starts the underlying communication channel based on + * configuration options injected, and connects the channel</li> <li>stop() - stops the dispatcher and releases + * resources</li> </ul> * * @author Manik Surtani * @since 1.0 */ @Scope(Scopes.GLOBAL) -public interface RPCManager { - /** - * Disconnects and closes the underlying JGroups channel. - */ - void disconnect(); +@NonVolatile +public interface RPCManager extends Lifecycle { /** - * Stops the RPCDispatcher and frees resources. Closes and disconnects the underlying JGroups channel if this is - * still open/connected. - */ - void stop(); - - /** - * Starts the RPCManager by connecting the underlying JGroups channel (if configured for replication). Connecting - * the channel may also involve state transfer (if configured) so the interceptor chain should be started and - * available before this method is called. - */ - void start(); - - /** * Invokes an RPC call on other caches in the cluster. * - * @param recipients a list of Addresses to invoke the call on. If this is null, the call is broadcast to - * the entire cluster. - * @param cacheCommand the cache command to invoke - * @param mode the group request mode to use. See {@link org.jgroups.blocks.GroupRequest}. - * @param timeout a timeout after which to throw a replication exception. - * @param responseFilter a response filter with which to filter out failed/unwanted/invalid responses. - * @param useOutOfBandMessage if true, the message is put on JGroups' OOB queue. See JGroups docs for more info. + * @param recipients a list of Addresses to invoke the call on. If this is null, the call is broadcast to the + * entire cluster. + * @param cacheCommand the cache command to invoke + * @param mode the response mode to use + * @param timeout a timeout after which to throw a replication exception. + * @param usePriorityQueue if true, a priority queue is used to deliver messages. May not be supported by all + * implementations. + * @param responseFilter a response filter with which to filter out failed/unwanted/invalid responses. * @return a list of responses from each member contacted. * @throws Exception in the event of problems. */ - List<Object> callRemoteMethods(Vector<Address> recipients, ReplicableCommand cacheCommand, int mode, long timeout, RspFilter responseFilter, boolean useOutOfBandMessage) throws Exception; + List<Object> invokeRemotely(List<Address> recipients, ReplicableCommand cacheCommand, ResponseMode mode, long timeout, boolean usePriorityQueue, ResponseFilter responseFilter) throws Exception; /** * Invokes an RPC call on other caches in the cluster. * - * @param recipients a list of Addresses to invoke the call on. If this is null, the call is broadcast to - * the entire cluster. - * @param cacheCommand the cache command to invoke - * @param mode the group request mode to use. See {@link org.jgroups.blocks.GroupRequest}. - * @param timeout a timeout after which to throw a replication exception. - * @param useOutOfBandMessage if true, the message is put on JGroups' OOB queue. See JGroups docs for more info. + * @param recipients a list of Addresses to invoke the call on. If this is null, the call is broadcast to the + * entire cluster. + * @param cacheCommand the cache command to invoke + * @param mode the response mode to use + * @param timeout a timeout after which to throw a replication exception. + * @param usePriorityQueue if true, a priority queue is used to deliver messages. May not be supported by all + * implementations. * @return a list of responses from each member contacted. * @throws Exception in the event of problems. */ - List<Object> callRemoteMethods(Vector<Address> recipients, ReplicableCommand cacheCommand, int mode, long timeout, boolean useOutOfBandMessage) throws Exception; + List<Object> invokeRemotely(List<Address> recipients, ReplicableCommand cacheCommand, ResponseMode mode, long timeout, boolean usePriorityQueue) throws Exception; /** * Invokes an RPC call on other caches in the cluster. * - * @param recipients a list of Addresses to invoke the call on. If this is null, the call is broadcast to - * the entire cluster. - * @param cacheCommand the cache command to invoke - * @param synchronous if true, sets group request mode to {@link org.jgroups.blocks.GroupRequest#GET_ALL}, - * and if false sets it to {@link org.jgroups.blocks.GroupRequest#GET_NONE}. - * @param timeout a timeout after which to throw a replication exception. - * @param useOutOfBandMessage if true, the message is put on JGroups' OOB queue. See JGroups docs for more info. + * @param recipients a list of Addresses to invoke the call on. If this is null, the call is broadcast to the + * entire cluster. + * @param cacheCommand the cache command to invoke + * @param mode the response mode to use + * @param timeout a timeout after which to throw a replication exception. * @return a list of responses from each member contacted. * @throws Exception in the event of problems. */ - List<Object> callRemoteMethods(Vector<Address> recipients, ReplicableCommand cacheCommand, boolean synchronous, long timeout, boolean useOutOfBandMessage) throws Exception; + List<Object> invokeRemotely(List<Address> recipients, ReplicableCommand cacheCommand, ResponseMode mode, long timeout) throws Exception; /** * @return true if the current Channel is the coordinator of the cluster. @@ -117,22 +99,16 @@ Address getCoordinator(); /** - * Retrieves the local JGroups channel's address + * Retrieves the current cache instance's network address * * @return an Address */ - Address getLocalAddress(); + Address getAddress(); /** - * Returns a defensively copied list of members in the current cluster view. - */ - List<Address> getMembers(); - - - /** - * Retrieves the Channel + * Returns a list of members in the current cluster view. * - * @return a channel + * @return a list of members. Typically, this would be defensively copied. */ - Channel getChannel(); + List<Address> getMembers(); } \ No newline at end of file Deleted: core/branches/flat/src/main/java/org/horizon/remoting/RPCManagerImpl.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/RPCManagerImpl.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/remoting/RPCManagerImpl.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,553 +0,0 @@ -/* - * JBoss, Home of Professional Open Source. - * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors - * as indicated by the @author tags. See the copyright.txt file in the - * distribution for a full listing of individual contributors. - * - * This is free software; you can redistribute it and/or modify it - * under the terms of the GNU Lesser General Public License as - * published by the Free Software Foundation; either version 2.1 of - * the License, or (at your option) any later version. - * - * This software is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this software; if not, write to the Free - * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA - * 02110-1301 USA, or see the FSF site: http://www.fsf.org. - */ -package org.horizon.remoting; - -import org.horizon.CacheException; -import org.horizon.CacheSPI; -import org.horizon.annotations.MBean; -import org.horizon.annotations.ManagedAttribute; -import org.horizon.annotations.ManagedOperation; -import org.horizon.commands.ReplicableCommand; -import org.horizon.config.Configuration; -import org.horizon.config.RuntimeConfig; -import org.horizon.context.InvocationContext; -import org.horizon.factories.ComponentRegistry; -import org.horizon.factories.annotations.Inject; -import org.horizon.factories.annotations.Start; -import org.horizon.factories.annotations.Stop; -import org.horizon.interceptors.InterceptorChain; -import org.horizon.invocation.InvocationContextContainer; -import org.horizon.lock.LockManager; -import org.horizon.lock.TimeoutException; -import org.horizon.logging.Log; -import org.horizon.logging.LogFactory; -import org.horizon.marshall.CommandAwareRpcDispatcher; -import org.horizon.marshall.Marshaller; -import org.horizon.notifications.Notifier; -import org.horizon.transaction.TransactionTable; -import org.horizon.util.ReflectionUtil; -import org.horizon.util.concurrent.ReclosableLatch; -import org.jgroups.Address; -import org.jgroups.Channel; -import org.jgroups.ChannelException; -import org.jgroups.ChannelFactory; -import org.jgroups.ExtendedMembershipListener; -import org.jgroups.JChannel; -import org.jgroups.View; -import org.jgroups.blocks.GroupRequest; -import org.jgroups.blocks.RspFilter; -import org.jgroups.protocols.TP; -import org.jgroups.stack.ProtocolStack; -import org.jgroups.util.Rsp; -import org.jgroups.util.RspList; - -import javax.transaction.TransactionManager; -import java.text.NumberFormat; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Vector; -import java.util.concurrent.TimeUnit; - -/** - * Manager that handles all RPC calls between JBoss Cache instances - * - * @author <a href="mailto:manik@jboss.org">Manik Surtani (manik(a)jboss.org)&lt;/a&gt; - * @since 1.0 - */ -@MBean(objectName = "RPCManager") -public class RPCManagerImpl implements RPCManager { - private Channel channel; - private final Log log = LogFactory.getLog(RPCManagerImpl.class); - private List<Address> members; - private long replicationCount; - private long replicationFailures; - private boolean statisticsEnabled = false; - - private final Object coordinatorLock = new Object(); - /** - * True if this Cache is the coordinator. - */ - private volatile boolean coordinator = false; - /** - * Thread gate used to block Dispatcher during JGroups FLUSH protocol - */ - private final ReclosableLatch flushBlockGate = new ReclosableLatch(); - /** - * JGroups RpcDispatcher in use. - */ - private CommandAwareRpcDispatcher rpcDispatcher = null; - - /** - * JGroups message listener. - */ - private ChannelMessageListener messageListener; - private Configuration configuration; - private Notifier notifier; - private CacheSPI spi; - private InvocationContextContainer invocationContextContainer; - private final boolean trace = log.isTraceEnabled(); - private Marshaller marshaller; - private TransactionManager txManager; - private TransactionTable txTable; - private InterceptorChain interceptorChain; - - private boolean isInLocalMode; - private ComponentRegistry componentRegistry; - private LockManager lockManager; - - @Inject - public void setupDependencies(ChannelMessageListener messageListener, Configuration configuration, Notifier notifier, - Marshaller extendedMarshaller, TransactionTable txTable, - TransactionManager txManager, InvocationContextContainer container, InterceptorChain interceptorChain, - ComponentRegistry componentRegistry, LockManager lockManager, CacheSPI spi) { - this.messageListener = messageListener; - this.configuration = configuration; - this.notifier = notifier; - this.spi = spi; - this.marshaller = extendedMarshaller; - this.txManager = txManager; - this.txTable = txTable; - this.invocationContextContainer = container; - this.interceptorChain = interceptorChain; - this.componentRegistry = componentRegistry; - this.lockManager = lockManager; - } - - // ------------ START: Lifecycle methods ------------ - - @Start(priority = 15) - public void start() { - if (configuration.getCacheMode().equals(Configuration.CacheMode.LOCAL)) { - log.debug("cache mode is local, will not create the channel"); - isInLocalMode = true; - return; - } - isInLocalMode = false; - if (log.isDebugEnabled()) log.debug("Cache mode is " + configuration.getCacheMode()); - - initialiseChannelAndRpcDispatcher(); - - //otherwise just connect - try { - channel.connect(configuration.getClusterName()); - } - catch (ChannelException e) { - throw new CacheException("Unable to connect to JGroups channel", e); - } - if (log.isInfoEnabled()) log.info("Cache local address is " + getLocalAddress()); - } - - public void disconnect() { - if (channel != null && channel.isOpen()) { - log.info("Disconnecting and closing the Channel"); - channel.disconnect(); - channel.close(); - } - } - - @Stop(priority = 8) - public void stop() { - try { - disconnect(); - } - catch (Exception toLog) { - log.error("Problem closing channel; setting it to null", toLog); - } - - channel = null; - configuration.getRuntimeConfig().setChannel(null); - if (rpcDispatcher != null) { - log.info("Stopping the RpcDispatcher"); - rpcDispatcher.stop(); - } - - if (members != null) members = null; - - coordinator = false; - - rpcDispatcher = null; - } - - @SuppressWarnings("deprecation") - private void initialiseChannelAndRpcDispatcher() throws CacheException { - buildChannel(); - // Channel.LOCAL *must* be set to false so we don't see our own messages - otherwise invalidations targeted at - // remote instances will be received by self. - channel.setOpt(Channel.LOCAL, false); - channel.setOpt(Channel.AUTO_RECONNECT, true); - channel.setOpt(Channel.AUTO_GETSTATE, false); - channel.setOpt(Channel.BLOCK, true); - rpcDispatcher = new CommandAwareRpcDispatcher(channel, messageListener, new MembershipListenerAdaptor(), - invocationContextContainer, invocationContextContainer, interceptorChain, componentRegistry); - checkAppropriateConfig(); - rpcDispatcher.setRequestMarshaller(marshaller); - rpcDispatcher.setResponseMarshaller(marshaller); - } - - private void buildChannel() { - channel = configuration.getRuntimeConfig().getChannel(); - if (channel == null) { - // Try to create a multiplexer channel - channel = getMultiplexerChannel(); - - if (channel != null) { - ReflectionUtil.setValue(configuration, "accessible", true); - configuration.setUsingMultiplexer(true); - if (log.isDebugEnabled()) - log.debug("Created Multiplexer Channel for cache cluster " + configuration.getClusterName() + " using stack " + configuration.getMultiplexerStack()); - } else { - try { - if (configuration.getClusterConfig() == null) { - log.debug("setting cluster properties to default value"); - channel = new JChannel(configuration.getDefaultClusterConfig()); - } else { - if (trace) { - log.trace("Cache cluster properties: " + configuration.getClusterConfig()); - } - channel = new JChannel(configuration.getClusterConfig()); - } - } - catch (ChannelException e) { - throw new CacheException(e); - } - } - - configuration.getRuntimeConfig().setChannel(channel); - } - } - - public Channel getChannel() { - return channel; - } - - - private JChannel getMultiplexerChannel() throws CacheException { - String stackName = configuration.getMultiplexerStack(); - - RuntimeConfig rtc = configuration.getRuntimeConfig(); - ChannelFactory channelFactory = rtc.getMuxChannelFactory(); - JChannel muxchannel = null; - - if (channelFactory != null) { - try { - muxchannel = (JChannel) channelFactory.createMultiplexerChannel(stackName, configuration.getClusterName()); - } - catch (Exception e) { - throw new CacheException("Failed to create multiplexed channel using stack " + stackName, e); - } - } - - return muxchannel; - } - - /** - * Only used with MVCC. - */ - // TODO: Fix me - /* - private void removeLocksForDeadMembers(InternalNode<?, ?> node, List deadMembers) - { - Set<GlobalTransaction> deadOwners = new HashSet<GlobalTransaction>(); - Object owner = lockManager.getWriteOwner(node.getKey()); - - if (isLockOwnerDead(owner, deadMembers)) deadOwners.add((GlobalTransaction) owner); - - // MVCC won't have any read locks. - - for (GlobalTransaction deadOwner : deadOwners) - { - boolean localTx = deadOwner.getAddress().equals(getLocalAddress()); -// boolean broken = LockUtil.breakTransactionLock(node.getKey(), lockManager, deadOwner, localTx, txTable, txManager); - boolean broken = true; // TODO fix me!! - - if (broken && trace) log.trace("Broke lock for node " + node.getKey() + " held by " + deadOwner); - } - - // Recursively unlock children - for (InternalNode child : node.getChildren()) removeLocksForDeadMembers(child, deadMembers); - } - - private boolean isLockOwnerDead(Object owner, List deadMembers) - { - boolean result = false; - if (owner != null && owner instanceof GlobalTransaction) - { - Object addr = ((GlobalTransaction) owner).getAddress(); - result = deadMembers.contains(addr); - } - return result; - } - */ - - // ------------ END: Lifecycle methods ------------ - - // ------------ START: RPC call methods ------------ - public List<Object> callRemoteMethods(Vector<Address> recipients, ReplicableCommand command, int mode, long timeout, boolean useOutOfBandMessage) throws Exception { - return callRemoteMethods(recipients, command, mode, timeout, null, useOutOfBandMessage); - } - - public List<Object> callRemoteMethods(Vector<Address> recipients, ReplicableCommand command, boolean synchronous, long timeout, boolean useOutOfBandMessage) throws Exception { - return callRemoteMethods(recipients, command, synchronous ? GroupRequest.GET_ALL : GroupRequest.GET_NONE, timeout, useOutOfBandMessage); - } - - public List<Object> callRemoteMethods(Vector<Address> recipients, ReplicableCommand command, int mode, long timeout, RspFilter responseFilter, boolean useOutOfBandMessage) throws Exception { - boolean success = true; - try { - // short circuit if we don't have an RpcDispatcher! - if (rpcDispatcher == null) return null; - int modeToUse = mode; - if (trace) - log.trace("callRemoteMethods(): valid members are " + recipients + " methods: " + command + " Using OOB? " + useOutOfBandMessage); - if (channel.flushSupported()) { - if (!flushBlockGate.await(configuration.getStateRetrievalTimeout(), TimeUnit.MILLISECONDS)) - throw new TimeoutException("State retrieval timed out waiting for flush unblock."); - } - useOutOfBandMessage = false; - RspList rsps = rpcDispatcher.invokeRemoteCommands(recipients, command, modeToUse, timeout, useOutOfBandMessage, responseFilter); - if (mode == GroupRequest.GET_NONE) return Collections.emptyList();// async case - if (trace) - log.trace("(" + getLocalAddress() + "): responses for method " + command.getClass().getSimpleName() + ":\n" + rsps); - // short-circuit no-return-value calls. - if (rsps == null) return Collections.emptyList(); - List<Object> retval = new ArrayList<Object>(rsps.size()); - for (Rsp rsp : rsps.values()) { - if (rsp.wasSuspected() || !rsp.wasReceived()) { - CacheException ex; - if (rsp.wasSuspected()) { - ex = new SuspectException("Suspected member: " + rsp.getSender()); - } else { - ex = new TimeoutException("Replication timeout for " + rsp.getSender()); - } - retval.add(new ReplicationException("rsp=" + rsp, ex)); - success = false; - } else { - Object value = rsp.getValue(); - if (value instanceof Exception && !(value instanceof ReplicationException)) { - // if we have any application-level exceptions make sure we throw them!! - if (trace) log.trace("Recieved exception'" + value + "' from " + rsp.getSender()); - throw (Exception) value; - } - retval.add(value); - success = true; - } - } - return retval; - } - catch (Exception e) { - success = false; - throw e; - } - finally { - computeStats(success); - } - } - - // ------------ END: Partial state transfer methods ------------ - - // ------------ START: Informational methods ------------ - - public Address getLocalAddress() { - return channel != null ? channel.getLocalAddress() : null; - } - - public List<Address> getMembers() { - if (isInLocalMode) return null; - if (members == null) - return Collections.emptyList(); - else - return members; - } - - public boolean isCoordinator() { - return coordinator; - } - - public Address getCoordinator() { - if (channel == null) { - return null; - } - - synchronized (coordinatorLock) { - while (members == null || members.isEmpty()) { - log.debug("getCoordinator(): waiting on viewAccepted()"); - try { - coordinatorLock.wait(); - } - catch (InterruptedException e) { - log.error("getCoordinator(): Interrupted while waiting for members to be set", e); - break; - } - } - return members != null && members.size() > 0 ? members.get(0) : null; - } - } - - // ------------ END: Informational methods ------------ - - /*----------------------- MembershipListener ------------------------*/ - - protected class MembershipListenerAdaptor implements ExtendedMembershipListener { - - public void viewAccepted(View newView) { - Vector<Address> newMembers = newView.getMembers(); - if (log.isInfoEnabled()) log.info("Received new cluster view: " + newView); - synchronized (coordinatorLock) { - boolean needNotification = false; - if (newMembers != null) { - if (members != null) { - // we had a membership list before this event. Check to make sure we haven't lost any members, - // and if so, determine what members have been removed - // and roll back any tx and break any locks - List<Address> removed = new ArrayList<Address>(members); - removed.removeAll(newMembers); - spi.getInvocationContext().getOptionOverrides().setSkipCacheStatusCheck(true); - // todo fix me -// NodeSPI root = null; // spi.getRoot(); -// if (root != null) -// { - // todo fix me - //removeLocksForDeadMembers(root.getDelegationTarget(), removed); -// } - } - - members = new ArrayList<Address>(newMembers); // defensive copy. - - needNotification = true; - } - - // Now that we have a view, figure out if we are the coordinator - coordinator = (members != null && members.size() != 0 && members.get(0).equals(getLocalAddress())); - - // now notify listeners - *after* updating the coordinator. - JBCACHE-662 - if (needNotification && notifier != null) { - InvocationContext ctx = invocationContextContainer.get(); - notifier.notifyViewChange(newView, ctx); - } - - // Wake up any threads that are waiting to know about who the coordinator is - coordinatorLock.notifyAll(); - } - } - - /** - * Called when a member is suspected. - */ - public void suspect(Address suspected_mbr) { - } - - /** - * Indicates that a channel has received a BLOCK event from FLUSH protocol. - */ - public void block() { - flushBlockGate.close(); - if (log.isDebugEnabled()) log.debug("Block received at " + getLocalAddress()); - - if (log.isDebugEnabled()) log.debug("Block processed at " + getLocalAddress()); - } - - /** - * Indicates that a channel has received a UNBLOCK event from FLUSH protocol. - */ - public void unblock() { - if (log.isDebugEnabled()) log.debug("UnBlock received at " + getLocalAddress()); - if (log.isDebugEnabled()) log.debug("UnBlock processed at " + getLocalAddress()); - flushBlockGate.open(); - } - - } - - //jmx operations - private void computeStats(boolean success) { - if (statisticsEnabled && rpcDispatcher != null) { - if (success) { - replicationCount++; - } else { - replicationFailures++; - } - } - } - - @ManagedOperation - public void resetStatistics() { - this.replicationCount = 0; - this.replicationFailures = 0; - } - - @ManagedAttribute(description = "number of successful replications") - public long getReplicationCount() { - return replicationCount; - } - - @ManagedAttribute(description = "number of failed replications") - public long getReplicationFailures() { - return replicationFailures; - } - - @ManagedAttribute(description = "whether or not jmx statistics are enabled") - public boolean isStatisticsEnabled() { - return statisticsEnabled; - } - - @ManagedAttribute - public void setStatisticsEnabled(boolean statisticsEnabled) { - this.statisticsEnabled = statisticsEnabled; - } - - @ManagedAttribute - public String getSuccessRatio() { - if (replicationCount == 0 || !statisticsEnabled) { - return "N/A"; - } - double totalCount = replicationCount + replicationFailures; - double ration = (double) replicationCount / totalCount * 100d; - return NumberFormat.getInstance().format(ration) + "%"; - } - - /** - * Checks to see whether the cache is using an appropriate JGroups config. - */ - private void checkAppropriateConfig() { - //if we use a shared transport do not log any warn message - if (configuration.getMultiplexerStack() != null) - return; - //bundling is not good for sync caches - Configuration.CacheMode cacheMode = configuration.getCacheMode(); - if (!cacheMode.equals(Configuration.CacheMode.LOCAL) && configuration.getCacheMode().isSynchronous()) { - ProtocolStack stack = ((JChannel) channel).getProtocolStack(); - TP transport = stack.getTransport(); - if (transport.isEnableBundling()) { - log.warn("You have enabled jgroups's message bundling, which is not recommended for sync replication. If there is no particular " + - "reason for this we strongly recommend to disable message bundling in JGroups config (enable_bundling=\"false\")."); - } - } - //bundling is good for async caches - if (!cacheMode.isSynchronous()) { - ProtocolStack stack = ((JChannel) channel).getProtocolStack(); - TP transport = stack.getTransport(); - if (!transport.isEnableBundling()) { - log.warn("You have disabled jgroups's message bundling, which is not recommended for async replication. If there is no particular " + - "reason for this we strongly recommend to enable message bundling in JGroups config (enable_bundling=\"true\")."); - } - } - } -} \ No newline at end of file Added: core/branches/flat/src/main/java/org/horizon/remoting/RPCManagerImpl.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/RPCManagerImpl.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/remoting/RPCManagerImpl.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,120 @@ +package org.horizon.remoting; + +import org.horizon.annotations.MBean; +import org.horizon.annotations.ManagedAttribute; +import org.horizon.annotations.ManagedOperation; +import org.horizon.commands.ReplicableCommand; +import org.horizon.config.GlobalConfiguration; +import org.horizon.factories.KnownComponentNames; +import org.horizon.factories.annotations.ComponentName; +import org.horizon.factories.annotations.Start; +import org.horizon.factories.annotations.Stop; +import org.horizon.marshall.Marshaller; +import org.horizon.notifications.CacheManagerNotifier; +import org.horizon.remoting.transport.Address; +import org.horizon.remoting.transport.Transport; + +import java.text.NumberFormat; +import java.util.List; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.atomic.AtomicLong; + +/** + * This component really is just a wrapper around a {@link org.horizon.remoting.transport.Transport} implementation, and + * is used to set up the transport and provide lifecycle and dependency hooks into external transport implementations. + * + * @author Manik Surtani + * @since 1.0 + */ +@MBean(objectName = "RPCManager") +public class RPCManagerImpl implements RPCManager { + + Transport t; + AtomicLong replicationCount = new AtomicLong(0); + AtomicLong replicationFailures = new AtomicLong(0); + boolean statisticsEnabled = false; // by default, don't gather statistics. + + public void injectDependencies(GlobalConfiguration globalConfiguration, Transport t, InboundInvocationHandler handler, + Marshaller marshaller, + @ComponentName(KnownComponentNames.ASYNC_SERIALIZATION_EXECUTOR) ExecutorService e, + CacheManagerNotifier notifier) { + this.t = t; + this.t.initialize(globalConfiguration, globalConfiguration.getTransportProperties(), marshaller, e, handler, notifier); + } + + @Start + public void start() { + t.start(); + } + + @Stop + public void stop() { + t.stop(); + } + + public List<Object> invokeRemotely(List<Address> recipients, ReplicableCommand cacheCommand, ResponseMode mode, long timeout, boolean usePriorityQueue, ResponseFilter responseFilter) throws Exception { + return t.invokeRemotely(recipients, cacheCommand, mode, timeout, usePriorityQueue, responseFilter); + } + + public List<Object> invokeRemotely(List<Address> recipients, ReplicableCommand cacheCommand, ResponseMode mode, long timeout, boolean usePriorityQueue) throws Exception { + return t.invokeRemotely(recipients, cacheCommand, mode, timeout, usePriorityQueue, null); + } + + public List<Object> invokeRemotely(List<Address> recipients, ReplicableCommand cacheCommand, ResponseMode mode, long timeout) throws Exception { + return t.invokeRemotely(recipients, cacheCommand, mode, timeout, false, null); + } + + public boolean isCoordinator() { + return t.isCoordinator(); + } + + public Address getCoordinator() { + return t.getCoordinator(); + } + + public Address getAddress() { + return t.getAddress(); + } + + public List<Address> getMembers() { + return t.getMembers(); + } + + // -------------------------------------------- JMX information ----------------------------------------------- + + @ManagedOperation + public void resetStatistics() { + replicationCount.set(0); + replicationFailures.set(0); + } + + @ManagedAttribute(description = "number of successful replications") + public long getReplicationCount() { + return replicationCount.get(); + } + + @ManagedAttribute(description = "number of failed replications") + public long getReplicationFailures() { + return replicationFailures.get(); + } + + @ManagedAttribute(description = "whether or not jmx statistics are enabled") + public boolean isStatisticsEnabled() { + return statisticsEnabled; + } + + @ManagedAttribute + public void setStatisticsEnabled(boolean statisticsEnabled) { + this.statisticsEnabled = statisticsEnabled; + } + + @ManagedAttribute + public String getSuccessRatio() { + if (replicationCount.get() == 0 || !statisticsEnabled) { + return "N/A"; + } + double totalCount = replicationCount.get() + replicationFailures.get(); + double ration = (double) replicationCount.get() / totalCount * 100d; + return NumberFormat.getInstance().format(ration) + "%"; + } +} Added: core/branches/flat/src/main/java/org/horizon/remoting/ResponseFilter.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/ResponseFilter.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/remoting/ResponseFilter.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,30 @@ +package org.horizon.remoting; + +import org.horizon.remoting.transport.Address; + +/** + * // TODO: Manik: Document this! + * + * @author Manik Surtani + * @since 1.0 + */ +public interface ResponseFilter { + /** + * Determines whether a response from a given sender should be added to the response list of the request + * + * @param response The response (usually a serializable value) + * @param sender The sender of response + * @return True if we should add the response to the response list ({@link org.jgroups.util.RspList}) of a request, + * otherwise false. In the latter case, we don't add the response to the response list. + */ + boolean isAcceptable(Object response, Address sender); + + /** + * Right after calling {@link #isAcceptable(Object, Address)}, this method is called to see whether we are done with + * the request and can unblock the caller + * + * @return False if the request is done, otherwise true + */ + boolean needMoreResponses(); + +} Added: core/branches/flat/src/main/java/org/horizon/remoting/ResponseMode.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/ResponseMode.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/remoting/ResponseMode.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,11 @@ +package org.horizon.remoting; + +/** + * // TODO: Manik: Document this! + * + * @author Manik Surtani + * @since 1.0 + */ +public enum ResponseMode { + SYNCHRONOUS, ASYNCHRONOUS, WAIT_FOR_VALID_RESPONSE +} Deleted: core/branches/flat/src/main/java/org/horizon/remoting/transport/Transport.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/transport/Transport.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/remoting/transport/Transport.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,22 +0,0 @@ -package org.horizon.remoting.transport; - -import org.horizon.lifecycle.Lifecycle; - -import java.util.List; -import java.util.Properties; - -/** - * // TODO: Manik: Document this! - * - * @author Manik Surtani - * @since 1.0 - */ -public interface Transport extends Lifecycle { - void setProperties(Properties p); - - Address getAddress(); - - boolean isCoordinator(); - - List<Address> getMembers(); -} Added: core/branches/flat/src/main/java/org/horizon/remoting/transport/Transport.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/transport/Transport.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/remoting/transport/Transport.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,76 @@ +package org.horizon.remoting.transport; + +import org.horizon.commands.ReplicableCommand; +import org.horizon.config.GlobalConfiguration; +import org.horizon.lifecycle.Lifecycle; +import org.horizon.marshall.Marshaller; +import org.horizon.notifications.CacheManagerNotifier; +import org.horizon.remoting.InboundInvocationHandler; +import org.horizon.remoting.ResponseFilter; +import org.horizon.remoting.ResponseMode; + +import java.util.List; +import java.util.Properties; +import java.util.concurrent.ExecutorService; + +/** + * An interface that provides a communication link with remote caches. Also allows remote caches to invoke commands on + * this cache instance. + * + * @author Manik Surtani + * @since 1.0 + */ +public interface Transport extends Lifecycle { + + /** + * Initializes the transport with global cache configuration and transport-specific properties. + * + * @param c global cache-wide configuration + * @param p properties to set + * @param marshaller marshaller to use for marshalling and unmarshalling + * @param asyncExecutor executor to use for asynchronous calls + * @param handler handler for invoking remotely originating calls on the local cache + */ + void initialize(GlobalConfiguration c, Properties p, Marshaller marshaller, ExecutorService asyncExecutor, + InboundInvocationHandler handler, CacheManagerNotifier notifier); + + /** + * Invokes an RPC call on other caches in the cluster. + * + * @param recipients a list of Addresses to invoke the call on. If this is null, the call is broadcast to the + * entire cluster. + * @param cacheCommand the cache command to invoke + * @param mode the response mode to use + * @param timeout a timeout after which to throw a replication exception. + * @param usePriorityQueue if true, a priority queue is used to deliver messages. May not be supported by all + * implementations. + * @param responseFilter a response filter with which to filter out failed/unwanted/invalid responses. + * @return a list of responses from each member contacted. + * @throws Exception in the event of problems. + */ + List<Object> invokeRemotely(List<Address> recipients, ReplicableCommand cacheCommand, ResponseMode mode, long timeout, boolean usePriorityQueue, ResponseFilter responseFilter) throws Exception; + + /** + * @return true if the current Channel is the coordinator of the cluster. + */ + boolean isCoordinator(); + + /** + * @return the Address of the current coordinator. + */ + Address getCoordinator(); + + /** + * Retrieves the current cache instance's network address + * + * @return an Address + */ + Address getAddress(); + + /** + * Returns a list of members in the current cluster view. + * + * @return a list of members. Typically, this would be defensively copied. + */ + List<Address> getMembers(); +} Added: core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/CommandAwareRpcDispatcher.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/CommandAwareRpcDispatcher.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/CommandAwareRpcDispatcher.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,188 @@ +/* + * JBoss, Home of Professional Open Source. + * Copyright 2000 - 2008, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.horizon.remoting.transport.jgroups; + +import org.horizon.CacheException; +import org.horizon.commands.ReplicableCommand; +import org.horizon.logging.Log; +import org.horizon.logging.LogFactory; +import org.horizon.remoting.InboundInvocationHandler; +import org.jgroups.Address; +import org.jgroups.Channel; +import org.jgroups.Message; +import org.jgroups.blocks.GroupRequest; +import org.jgroups.blocks.RpcDispatcher; +import org.jgroups.blocks.RspFilter; +import org.jgroups.util.Buffer; +import org.jgroups.util.Rsp; +import org.jgroups.util.RspList; + +import java.io.NotSerializableException; +import java.util.Vector; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; + +/** + * A JGroups RPC dispatcher that knows how to deal with {@link ReplicableCommand}s. + * + * @author Manik Surtani (<a href="mailto:manik@jboss.org">manik@jboss.org</a>) + * @since 1.0 + */ +public class CommandAwareRpcDispatcher extends RpcDispatcher { + protected boolean trace; + ExecutorService asyncExecutor; + InboundInvocationHandler inboundInvocationHandler; + private Log log = LogFactory.getLog(CommandAwareRpcDispatcher.class); + + public CommandAwareRpcDispatcher() { + } + + public CommandAwareRpcDispatcher(Channel channel, + JGroupsTransport transport, + ExecutorService asyncExecutor, + InboundInvocationHandler inboundInvocationHandler) { + super(channel, transport, transport, transport); + this.asyncExecutor = asyncExecutor; + this.inboundInvocationHandler = inboundInvocationHandler; + + trace = log.isTraceEnabled(); + } + + protected boolean isValid(Message req) { + if (req == null || req.getLength() == 0) { + log.error("message or message buffer is null"); + return false; + } + + return true; + } + + /** + * Similar to {@link #callRemoteMethods(java.util.Vector, org.jgroups.blocks.MethodCall, int, long, boolean, boolean, + * org.jgroups.blocks.RspFilter)} except that this version is aware of {@link ReplicableCommand} objects. + */ + public RspList invokeRemoteCommands(Vector<Address> dests, ReplicableCommand command, int mode, long timeout, + boolean anycasting, boolean oob, RspFilter filter) + throws NotSerializableException, ExecutionException, InterruptedException { + ReplicationTask task = new ReplicationTask(command, oob, dests, mode, timeout, anycasting, filter); + + if (mode == GroupRequest.GET_NONE) { + asyncExecutor.submit(task); + return null; // don't wait for a response! + } else { + RspList response = null; + try { + response = task.call(); + } catch (Exception e) { + throw new CacheException(e); + } + if (response.isEmpty() || containsOnlyNulls(response)) + return null; + else + return response; + } + } + + private boolean containsOnlyNulls(RspList l) { + for (Rsp r : l.values()) { + if (r.getValue() != null || !r.wasReceived() || r.wasSuspected()) return false; + } + return true; + } + + /** + * Message contains a Command. Execute it against *this* object and return result. + */ + @Override + public Object handle(Message req) { + if (isValid(req)) { + try { + return executeCommand((ReplicableCommand) req_marshaller.objectFromByteBuffer(req.getBuffer(), req.getOffset(), req.getLength()), req); + } + catch (Throwable x) { + if (trace) log.trace("Problems invoking command.", x); + return x; + } + } else { + return null; + } + } + + protected Object executeCommand(ReplicableCommand cmd, Message req) throws Throwable { + if (cmd == null) throw new NullPointerException("Unable to execute a null command! Message was " + req); + if (trace) log.trace("Executing command: {0} [sender={1}]", cmd, req.getSrc()); + return inboundInvocationHandler.handle(cmd); + } + + @Override + public String toString() { + return getClass().getSimpleName() + "[Outgoing marshaller: " + req_marshaller + "; incoming marshaller: " + rsp_marshaller + "]"; + } + + private class ReplicationTask implements Callable<RspList> { + + private ReplicableCommand command; + private boolean oob; + private Vector<Address> dests; + private int mode; + private long timeout; + private boolean anycasting; + private RspFilter filter; + + private ReplicationTask(ReplicableCommand command, boolean oob, Vector<Address> dests, + int mode, long timeout, + boolean anycasting, RspFilter filter) { + this.command = command; + this.oob = oob; + this.dests = dests; + this.mode = mode; + this.timeout = timeout; + this.anycasting = anycasting; + this.filter = filter; + } + + public RspList call() throws Exception { + Buffer buf; + try { + buf = req_marshaller.objectToBuffer(command); + } + catch (Exception e) { + throw new RuntimeException("Failure to marshal argument(s)", e); + } + + Message msg = new Message(); + msg.setBuffer(buf); + if (oob) msg.setFlag(Message.OOB); + RspList retval = castMessage(dests, msg, mode, timeout, anycasting, filter); + if (trace) log.trace("responses: {0}", retval); + + // a null response is 99% likely to be due to a marshalling problem - we throw a NSE, this needs to be changed when + // JGroups supports http://jira.jboss.com/jira/browse/JGRP-193 + // the serialization problem could be on the remote end and this is why we cannot catch this above, when marshalling. + if (retval == null) + throw new NotSerializableException("RpcDispatcher returned a null. This is most often caused by args for " + + command.getClass().getSimpleName() + " not being serializable."); + return retval; + } + } +} \ No newline at end of file Added: core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsAddress.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsAddress.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsAddress.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,59 @@ +package org.horizon.remoting.transport.jgroups; + +import org.horizon.remoting.transport.Address; + +import java.io.Externalizable; +import java.io.IOException; +import java.io.ObjectInput; +import java.io.ObjectOutput; + +/** + * An encapsulation of a JGroups Address + * + * @author Manik Surtani + * @since 1.0 + */ +public class JGroupsAddress implements Address, Externalizable { + org.jgroups.Address address; + + public JGroupsAddress() { + } + + public JGroupsAddress(org.jgroups.Address address) { + this.address = address; + } + + @Override + public boolean equals(Object o) { + if (this == o) return true; + if (o == null || getClass() != o.getClass()) return false; + + JGroupsAddress that = (JGroupsAddress) o; + + if (address != null ? !address.equals(that.address) : that.address != null) return false; + + return true; + } + + @Override + public int hashCode() { + return address != null ? address.hashCode() : 0; + } + + @Override + public String toString() { + if (address != null) { + return address.toString(); + } else { + return "null"; + } + } + + public void writeExternal(ObjectOutput out) throws IOException { + out.writeObject(address); + } + + public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException { + address = (org.jgroups.Address) in.readObject(); + } +} Added: core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsResponseFilterAdapter.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsResponseFilterAdapter.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsResponseFilterAdapter.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,33 @@ +package org.horizon.remoting.transport.jgroups; + +import org.horizon.remoting.ResponseFilter; +import org.jgroups.Address; +import org.jgroups.blocks.RspFilter; + +/** + * Acts as a bridge between JGroups RspFilter and {@link org.horizon.remoting.ResponseFilter}. + * + * @author Manik Surtani + * @since 1.0 + */ +public class JGroupsResponseFilterAdapter implements RspFilter { + + ResponseFilter r; + + /** + * Creates an instance of the adapter + * + * @param r response filter to wrap + */ + public JGroupsResponseFilterAdapter(ResponseFilter r) { + this.r = r; + } + + public boolean isAcceptable(Object response, Address sender) { + return r.isAcceptable(response, new JGroupsAddress(sender)); + } + + public boolean needMoreResponses() { + return r.needMoreResponses(); + } +} Added: core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsTransport.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsTransport.java (rev 0) +++ core/branches/flat/src/main/java/org/horizon/remoting/transport/jgroups/JGroupsTransport.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -0,0 +1,371 @@ +package org.horizon.remoting.transport.jgroups; + +import org.horizon.CacheException; +import org.horizon.commands.ReplicableCommand; +import org.horizon.config.GlobalConfiguration; +import org.horizon.config.parsing.XmlConfigHelper; +import org.horizon.lock.TimeoutException; +import org.horizon.logging.Log; +import org.horizon.logging.LogFactory; +import org.horizon.marshall.Marshaller; +import org.horizon.notifications.CacheManagerNotifier; +import org.horizon.remoting.InboundInvocationHandler; +import org.horizon.remoting.ReplicationException; +import org.horizon.remoting.ResponseFilter; +import org.horizon.remoting.ResponseMode; +import org.horizon.remoting.transport.Address; +import org.horizon.remoting.transport.Transport; +import org.horizon.util.FileLookup; +import org.jgroups.Channel; +import org.jgroups.ChannelException; +import org.jgroups.JChannel; +import org.jgroups.MembershipListener; +import org.jgroups.Message; +import org.jgroups.MessageListener; +import org.jgroups.View; +import org.jgroups.blocks.GroupRequest; +import org.jgroups.blocks.RspFilter; +import org.jgroups.util.Rsp; +import org.jgroups.util.RspList; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Properties; +import java.util.Vector; +import java.util.concurrent.ExecutorService; + +/** + * An encapsulation of a JGroups transport + * + * @author Manik Surtani + * @since 1.0 + */ +public class JGroupsTransport implements Transport, MembershipListener, MessageListener { + public static final String CONFIGURATION_STRING = "configurationString"; + public static final String CONFIGURATION_XML = "configurationXml"; + public static final String CONFIGURATION_FILE = "configurationFile"; + private static final String DEFAULT_JGROUPS_CONFIGURATION_FILE = "flush-udp.xml"; + + Channel channel; + Address address; + volatile List<Address> members = Collections.emptyList(); + volatile boolean coordinator = false; + final Object membersListLock = new Object(); // guards members + CommandAwareRpcDispatcher dispatcher; + static final Log log = LogFactory.getLog(JGroupsTransport.class); + static final boolean trace = log.isTraceEnabled(); + GlobalConfiguration c; + Properties p; + InboundInvocationHandler inboundInvocationHandler; + Marshaller marshaller; + ExecutorService asyncExecutor; + CacheManagerNotifier notifier; + + // ------------------------------------------------------------------------------------------------------------------ + // Lifecycle and setup stuff + // ------------------------------------------------------------------------------------------------------------------ + + public void initialize(GlobalConfiguration c, Properties p, Marshaller marshaller, ExecutorService asyncExecutor, + InboundInvocationHandler inboundInvocationHandler, CacheManagerNotifier notifier) { + this.c = c; + this.p = p; + this.marshaller = marshaller; + this.asyncExecutor = asyncExecutor; + this.inboundInvocationHandler = inboundInvocationHandler; + this.notifier = notifier; + } + + public void start() { + log.info("Starting JGroups Channel"); + + initChannelAndRPCDispatcher(); + + //otherwise just connect + try { + channel.connect(c.getClusterName()); + } + catch (ChannelException e) { + throw new CacheException("Unable to start JGroups Channel", e); + } + log.info("Cache local address is {0}", getAddress()); + } + + public void stop() { + try { + if (channel != null && channel.isOpen()) { + log.info("Disconnecting and closing JGroups Channel"); + channel.disconnect(); + channel.close(); + } + } + catch (Exception toLog) { + log.error("Problem closing channel; setting it to null", toLog); + } + + channel = null; + if (dispatcher != null) { + log.info("Stopping the RpcDispatcher"); + dispatcher.stop(); + } + + if (members != null) members = null; + coordinator = false; + dispatcher = null; + } + + private void initChannelAndRPCDispatcher() throws CacheException { + buildChannel(); + // Channel.LOCAL *must* be set to false so we don't see our own messages - otherwise invalidations targeted at + // remote instances will be received by self. + channel.setOpt(Channel.LOCAL, false); + channel.setOpt(Channel.AUTO_RECONNECT, true); + channel.setOpt(Channel.AUTO_GETSTATE, false); + channel.setOpt(Channel.BLOCK, true); + dispatcher = new CommandAwareRpcDispatcher(channel, this, + asyncExecutor, inboundInvocationHandler); + dispatcher.setRequestMarshaller(marshaller); + dispatcher.setResponseMarshaller(marshaller); + } + + private void buildChannel() { + // TODO: Check for injected channels and for channel factories. + // in order of preference - we first look for an external JGroups file, then a set of XML properties, and + // finally the legacy JGroups String properties. + String cfg; + if (p.containsKey(CONFIGURATION_FILE)) { + cfg = p.getProperty(CONFIGURATION_FILE); + try { + channel = new JChannel(new FileLookup().lookupFileLocation(cfg)); + } catch (Exception e) { + // move on to next method to configure the channel + channel = null; + } + } + + if (channel == null && p.containsKey(CONFIGURATION_XML)) { + cfg = p.getProperty(CONFIGURATION_XML); + try { + channel = new JChannel(XmlConfigHelper.stringToElement(cfg)); + } catch (Exception e) { + // move on to next method to configure the channel + channel = null; + } + } + + if (channel == null && p.containsKey(CONFIGURATION_STRING)) { + cfg = p.getProperty(CONFIGURATION_STRING); + try { + channel = new JChannel(cfg); + } catch (Exception e) { + // move on to next method to configure the channel + channel = null; + } + } + + if (channel == null) { + log.info("Unable to use any JGroups configuration mechanisms provided in properties {0}. Using default JGroups configuration!", p); + try { + channel = new JChannel(new FileLookup().lookupFileLocation(DEFAULT_JGROUPS_CONFIGURATION_FILE)); + } catch (ChannelException e) { + throw new CacheException("Unable to start JGroups channel", e); + } + } + } + + // ------------------------------------------------------------------------------------------------------------------ + // querying cluster status + // ------------------------------------------------------------------------------------------------------------------ + + public boolean isCoordinator() { + return coordinator; + } + + public Address getCoordinator() { + if (channel == null) return null; + synchronized (membersListLock) { + while (members == null || members.isEmpty()) { + log.debug("Waiting on view being accepted"); + try { + membersListLock.wait(); + } catch (InterruptedException e) { + log.error("getCoordinator(): Interrupted while waiting for members to be set", e); + break; + } + } + return members == null || members.isEmpty() ? null : members.get(0); + } + } + + public List<Address> getMembers() { + return members; + } + + public Address getAddress() { + if (address == null) { + address = new JGroupsAddress(channel.getLocalAddress()); + } + return address; + } + + + // ------------------------------------------------------------------------------------------------------------------ + // outbound RPC + // ------------------------------------------------------------------------------------------------------------------ + + public List<Object> invokeRemotely(List<Address> recipients, ReplicableCommand cacheCommand, ResponseMode mode, long timeout, boolean usePriorityQueue, ResponseFilter responseFilter) + throws Exception { + + if (recipients != null && recipients.isEmpty()) { + // don't send if dest list is empty + log.trace("Destination list is empty: no need to send message"); + return Collections.emptyList(); + } + + log.trace("dests={0}, command={1}, mode={2}, timeout={3}", recipients, cacheCommand, mode, timeout); + + RspList rsps = dispatcher.invokeRemoteCommands(toJGroupsAddressVector(recipients), cacheCommand, toJGroupsMode(mode), + timeout, false, usePriorityQueue, + toJGroupsFilter(responseFilter)); + + if (mode == ResponseMode.ASYNCHRONOUS) return Collections.emptyList();// async case + + if (trace) + log.trace("Cache [{0}]: responses for command {1}:\n{2}", getAddress(), cacheCommand.getClass().getSimpleName(), rsps); + + // short-circuit no-return-value calls. + if (rsps == null) return Collections.emptyList(); + List<Object> retval = new ArrayList<Object>(rsps.size()); + + for (Rsp rsp : rsps.values()) { + if (rsp.wasSuspected() || !rsp.wasReceived()) { + CacheException ex; + if (rsp.wasSuspected()) { + ex = new SuspectException("Suspected member: " + rsp.getSender()); + } else { + ex = new TimeoutException("Replication timeout for " + rsp.getSender()); + } + retval.add(new ReplicationException("rsp=" + rsp, ex)); + } else { + Object value = rsp.getValue(); + if (value instanceof Exception && !(value instanceof ReplicationException)) { + // if we have any application-level exceptions make sure we throw them!! + if (trace) log.trace("Recieved exception'" + value + "' from " + rsp.getSender()); + throw (Exception) value; + } + retval.add(value); + } + } + return retval; + } + + private int toJGroupsMode(ResponseMode mode) { + switch (mode) { + case ASYNCHRONOUS: + return GroupRequest.GET_NONE; + case SYNCHRONOUS: + return GroupRequest.GET_ALL; + case WAIT_FOR_VALID_RESPONSE: + return GroupRequest.GET_MAJORITY; + } + throw new CacheException("Unknown response mode " + mode); + } + + private RspFilter toJGroupsFilter(ResponseFilter responseFilter) { + return responseFilter == null ? null : new JGroupsResponseFilterAdapter(responseFilter); + } + + // ------------------------------------------------------------------------------------------------------------------ + // Implementations of JGroups interfaces + // ------------------------------------------------------------------------------------------------------------------ + + public void viewAccepted(View newView) { + Vector<org.jgroups.Address> newMembers = newView.getMembers(); + if (log.isInfoEnabled()) log.info("Received new cluster view: {0}", newView); + synchronized (membersListLock) { + boolean needNotification = false; + if (newMembers != null) { + + // TODO: Implement breaking stale locks for dead members +// if (members != null) { + // we had a membership list before this event. Check to make sure we haven't lost any members, + // and if so, determine what members have been removed + // and roll back any tx and break any locks +// List<org.jgroups.Address> removed = toJGroupsAddressList(members); +// removed.removeAll(newMembers); +// spi.getInvocationContext().getOptionOverrides().setSkipCacheStatusCheck(true); +// NodeSPI root = null; // spi.getRoot(); +// if (root != null) +// { + //removeLocksForDeadMembers(root.getDelegationTarget(), removed); +// } +// } + + // we need a defensive copy anyway + members = fromJGroupsAddressList(newMembers); + needNotification = true; + } + + // Now that we have a view, figure out if we are the coordinator + coordinator = (members != null && members.size() != 0 && members.get(0).equals(address)); + + // now notify listeners - *after* updating the coordinator. - JBCACHE-662 + if (needNotification && notifier != null) { + notifier.notifyViewChange(members, getAddress()); + } + + // Wake up any threads that are waiting to know about who the coordinator is + membersListLock.notifyAll(); + } + } + + public void suspect(org.jgroups.Address suspected_mbr) { + // no-op + } + + public void block() { + // TODO: Do we need these for state transfer? + throw new UnsupportedOperationException("Implement me!"); + } + + public void receive(Message msg) { + // no-op + } + + public byte[] getState() { + // TODO: Do we need these for state transfer? + throw new UnsupportedOperationException("Implement me!"); + } + + public void setState(byte[] state) { + // TODO: Do we need these for state transfer? + throw new UnsupportedOperationException("Implement me!"); + } + + + // ------------------------------------------------------------------------------------------------------------------ + // Helpers to convert between Address types + // ------------------------------------------------------------------------------------------------------------------ + + private Vector<org.jgroups.Address> toJGroupsAddressVector(List<Address> list) { + if (list == null) return null; + if (list.isEmpty()) return new Vector<org.jgroups.Address>(); + + Vector<org.jgroups.Address> retval = new Vector<org.jgroups.Address>(list.size()); + for (Address a : list) { + JGroupsAddress ja = (JGroupsAddress) a; + retval.add(ja.address); + } + return retval; + } + + private List<Address> fromJGroupsAddressList(List<org.jgroups.Address> list) { + if (list == null || list.isEmpty()) return Collections.emptyList(); + + List<Address> retval = new ArrayList<Address>(list.size()); + for (org.jgroups.Address a : list) { + retval.add(new JGroupsAddress(a)); + } + return retval; + } +} Modified: core/branches/flat/src/main/java/org/horizon/transaction/GlobalTransaction.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/transaction/GlobalTransaction.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/transaction/GlobalTransaction.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -21,9 +21,8 @@ */ package org.horizon.transaction; +import org.horizon.remoting.transport.Address; -import org.jgroups.Address; - import java.io.Externalizable; import java.io.IOException; import java.io.ObjectInput; Modified: core/branches/flat/src/main/java/org/horizon/transaction/TransactionTable.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/transaction/TransactionTable.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/transaction/TransactionTable.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -30,7 +30,7 @@ import org.horizon.logging.Log; import org.horizon.logging.LogFactory; import org.horizon.remoting.RPCManager; -import org.jgroups.Address; +import org.horizon.remoting.transport.Address; import javax.transaction.Status; import javax.transaction.SystemException; @@ -293,7 +293,7 @@ // GlobalTransaction gtx = get(tx); if (gtx == null && createIfNotExists) { - Address addr = rpcManager.getLocalAddress(); + Address addr = rpcManager.getAddress(); gtx = GlobalTransaction.create(addr); if (trace) log.trace("Creating new GlobalTransaction " + gtx); TransactionContext transactionContext; Modified: core/branches/flat/src/main/java/org/horizon/tree/TreeCache.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/tree/TreeCache.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/tree/TreeCache.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -28,7 +28,8 @@ import org.horizon.context.InvocationContext; import org.horizon.interceptors.base.CommandInterceptor; import org.horizon.lifecycle.Lifecycle; -import org.jgroups.Address; +import org.horizon.notifications.Listenable; +import org.horizon.remoting.transport.Address; import java.util.List; import java.util.Map; @@ -54,7 +55,7 @@ * @see Node * @since 1.0 */ -public interface TreeCache<K, V> extends Lifecycle { +public interface TreeCache<K, V> extends Lifecycle, Listenable { /** * Returns the root node of this cache. * @@ -63,33 +64,6 @@ Node<K, V> getRoot(); /** - * Adds a {@link org.horizon.notifications.annotation.CacheListener}-annotated object to the entire cache. The - * object passed in needs to be properly annotated with the {@link org.horizon.notifications.annotation.CacheListener} - * annotation otherwise an {@link org.horizon.notifications.IncorrectCacheListenerException} will be thrown. - * - * @param listener listener to add - */ - void addCacheListener(Object listener); - - /** - * Removes a {@link org.horizon.notifications.annotation.CacheListener}-annotated object from the cache. The object - * passed in needs to be properly annotated with the {@link org.horizon.notifications.annotation.CacheListener} - * annotation otherwise an {@link org.horizon.notifications.IncorrectCacheListenerException} will be thrown. - * - * @param listener listener to remove - */ - void removeCacheListener(Object listener); - - /** - * Retrieves an immutable {@link List} of objects annotated as {@link org.horizon.notifications.annotation.CacheListener}s - * attached to the cache. - * - * @return an immutable {@link List} of objects annotated as {@link org.horizon.notifications.annotation.CacheListener}s - * attached to the cache. - */ - Set<Object> getCacheListeners(); - - /** * Associates the specified value with the specified key for a {@link Node} in this cache. If the {@link Node} * previously contained a mapping for this key, the old value is replaced by the specified value. * @@ -287,7 +261,7 @@ * * @return the local address of this cache in a cluster, or <code>null</code> if running in local mode. */ - Address getLocalAddress(); + Address getAddress(); /** * Returns a list of members in the cluster, or <code>null</code> if running in local mode. Modified: core/branches/flat/src/main/java/org/horizon/tree/TreeCacheImpl.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/tree/TreeCacheImpl.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/tree/TreeCacheImpl.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -30,7 +30,7 @@ import org.horizon.interceptors.base.CommandInterceptor; import org.horizon.logging.Log; import org.horizon.logging.LogFactory; -import org.jgroups.Address; +import org.horizon.remoting.transport.Address; import java.util.List; import java.util.Map; @@ -287,16 +287,16 @@ } // ------------------ nothing different; just delegate to the cache - public void addCacheListener(Object listener) { - cache.addCacheListener(listener); + public void addListener(Object listener) { + cache.addListener(listener); } - public void removeCacheListener(Object listener) { - cache.removeCacheListener(listener); + public void removeListener(Object listener) { + cache.removeListener(listener); } - public Set<Object> getCacheListeners() { - return cache.getCacheListeners(); + public Set<Object> getListeners() { + return cache.getListeners(); } public void startBatch() { @@ -343,12 +343,12 @@ cache.setInvocationContext(ctx); } - public Address getLocalAddress() { - return cache.getLocalAddress(); + public Address getAddress() { + return cache.getCacheManager().getAddress(); } public List<Address> getMembers() { - return cache.getMembers(); + return cache.getCacheManager().getMembers(); } public void start() throws CacheException { Modified: core/branches/flat/src/main/java/org/horizon/util/TestingUtil.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/util/TestingUtil.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/util/TestingUtil.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -18,7 +18,6 @@ import org.horizon.interceptors.base.CommandInterceptor; import org.horizon.lock.LockManager; import org.horizon.tree.TreeCache; -import org.jgroups.JChannel; import javax.transaction.TransactionManager; import java.io.File; @@ -252,7 +251,7 @@ } else if (memberCount < members.size()) { // This is an exceptional condition StringBuilder sb = new StringBuilder("Cache at address "); - sb.append(cache.getRPCManager().getLocalAddress()); + sb.append(cache.getCacheManager().getAddress()); sb.append(" had "); sb.append(members.size()); sb.append(" members; expecting "); @@ -282,14 +281,14 @@ public static boolean isCacheViewComplete(Cache c, int memberCount, boolean barfIfTooManyMembers) { CacheSPI cache = (CacheSPI) c; - List members = cache.getMembers(); + List members = cache.getCacheManager().getMembers(); if (members == null || memberCount > members.size()) { return false; } else if (memberCount < members.size()) { if (barfIfTooManyMembers) { // This is an exceptional condition StringBuilder sb = new StringBuilder("Cache at address "); - sb.append(cache.getRPCManager().getLocalAddress()); + sb.append(cache.getCacheManager().getAddress()); sb.append(" had "); sb.append(members.size()); sb.append(" members; expecting "); @@ -534,25 +533,26 @@ } public static String getJGroupsAttribute(Cache cache, String protocol, String attribute) { - String s = ((JChannel) ((CacheSPI) cache).getRPCManager().getChannel()).getProperties(); - String[] protocols = s.split(":"); - String attribs = null; - for (String p : protocols) { - boolean hasAttribs = p.contains("("); - String name = hasAttribs ? p.substring(0, p.indexOf('(')) : p; - attribs = hasAttribs ? p.substring(p.indexOf('(') + 1, p.length() - 1) : null; - - if (name.equalsIgnoreCase(protocol)) break; - } - - if (attribs != null) { - String[] attrArray = attribs.split(";"); - for (String a : attrArray) { - String[] kvPairs = a.split("="); - if (kvPairs[0].equalsIgnoreCase(attribute)) return kvPairs[1]; - } - } - return null; + throw new RuntimeException("Implement me"); +// String s = ((JChannel) ((CacheSPI) cache).getRPCManager().getChannel()).getProperties(); +// String[] protocols = s.split(":"); +// String attribs = null; +// for (String p : protocols) { +// boolean hasAttribs = p.contains("("); +// String name = hasAttribs ? p.substring(0, p.indexOf('(')) : p; +// attribs = hasAttribs ? p.substring(p.indexOf('(') + 1, p.length() - 1) : null; +// +// if (name.equalsIgnoreCase(protocol)) break; +// } +// +// if (attribs != null) { +// String[] attrArray = attribs.split(";"); +// for (String a : attrArray) { +// String[] kvPairs = a.split("="); +// if (kvPairs[0].equalsIgnoreCase(attribute)) return kvPairs[1]; +// } +// } +// return null; } public static void dumpCacheContents(List caches) { @@ -563,7 +563,7 @@ if (c == null) { System.out.println(" ** Cache " + count + " is null!"); } else { - System.out.println(" ** Cache " + count + " is " + c.getRPCManager().getLocalAddress()); + System.out.println(" ** Cache " + count + " is " + c.getCacheManager().getAddress()); // System.out.println(" " + CachePrinter.printCacheDetails(c)); } count++; Modified: core/branches/flat/src/main/java/org/horizon/util/TypedProperties.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/util/TypedProperties.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/util/TypedProperties.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -15,10 +15,33 @@ public class TypedProperties extends Properties { private static final Log log = LogFactory.getLog(TypedProperties.class); + /** + * Copy constructor + * + * @param p properties instance to from. If null, then it is treated as an empty Properties instance. + */ public TypedProperties(Properties p) { - if (p != null) putAll(p); + if (p != null && !p.isEmpty()) putAll(p); } + /** + * Default constructor that returns an empty instance + */ + public TypedProperties() { + + } + + /** + * Factory method that converts a JDK {@link Properties} instance to an instance of TypedProperties, if needed. + * + * @param p properties to convert. + * @return A TypedProperties object. Returns an empty TypedProperties instance if p is null. + */ + public static TypedProperties toTypedProperties(Properties p) { + if (p instanceof TypedProperties) return (TypedProperties) p; + return new TypedProperties(p); + } + public int getIntProperty(String key, int defaultValue) { String value = getProperty(key); if (value == null) return defaultValue; Modified: core/branches/flat/src/main/java/org/horizon/util/internals/ViewChangeListener.java =================================================================== --- core/branches/flat/src/main/java/org/horizon/util/internals/ViewChangeListener.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/main/java/org/horizon/util/internals/ViewChangeListener.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -1,7 +1,7 @@ package org.horizon.util.internals; import org.horizon.Cache; -import org.horizon.notifications.annotation.CacheListener; +import org.horizon.notifications.annotation.Listener; import org.horizon.notifications.annotation.ViewChanged; import org.horizon.notifications.event.ViewChangedEvent; @@ -22,7 +22,7 @@ * @author (various) * @since 1.0 */ -@CacheListener +@Listener public class ViewChangeListener { CountDownLatch latch; @@ -33,7 +33,7 @@ */ public ViewChangeListener(Cache cache) { this.latch = new CountDownLatch(1); - cache.addCacheListener(this); + cache.addListener(this); } @ViewChanged Modified: core/branches/flat/src/test/java/org/horizon/BasicTest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/BasicTest.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/test/java/org/horizon/BasicTest.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -24,7 +24,7 @@ import org.horizon.config.Configuration; import org.horizon.logging.Log; import org.horizon.logging.LogFactory; -import org.horizon.manager.CacheManager; +import org.horizon.manager.DefaultCacheManager; import org.horizon.manager.NamedCacheNotFoundException; import org.horizon.util.TestingUtil; import org.testng.annotations.Test; @@ -35,7 +35,7 @@ // create a cache manager Configuration c = new Configuration(); // LOCAL mode c.setFetchInMemoryState(false); - CacheManager cm = new CacheManager(c); + DefaultCacheManager cm = new DefaultCacheManager(c); try { cm.start(); Cache cache = cm.getCache("test"); @@ -67,8 +67,8 @@ Configuration configuration = new Configuration(); configuration.setCacheMode(Configuration.CacheMode.REPL_SYNC); - CacheManager firstManager = new CacheManager(configuration); - CacheManager secondManager = new CacheManager(configuration); + DefaultCacheManager firstManager = new DefaultCacheManager(configuration); + DefaultCacheManager secondManager = new DefaultCacheManager(configuration); try { firstManager.start(); Modified: core/branches/flat/src/test/java/org/horizon/UnitTestCacheFactory.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/UnitTestCacheFactory.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/test/java/org/horizon/UnitTestCacheFactory.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -11,7 +11,7 @@ import org.horizon.config.parsing.XmlConfigurationParserJBC3; import org.horizon.logging.Log; import org.horizon.logging.LogFactory; -import org.horizon.manager.CacheManager; +import org.horizon.manager.DefaultCacheManager; import org.horizon.util.TestingUtil; import org.jgroups.conf.XmlConfigurator; @@ -134,7 +134,7 @@ } // Cache<K, V> cache = new DefaultCacheFactory<K, V>().createCache(configuration, start); - CacheManager cm = new CacheManager(configuration); + DefaultCacheManager cm = new DefaultCacheManager(configuration); Cache<K, V> cache = cm.getCache(); List<Cache> caches = threadCaches.get(); @@ -189,19 +189,20 @@ * @param configuration Configuration to update. */ public void mangleConfiguration(Configuration configuration) { + // TODO fix this to use the cache manager - configuration.setClusterConfig(mangleClusterConfiguration(configuration.getClusterConfig())); +// configuration.setClusterConfig(mangleClusterConfiguration(configuration.getClusterConfig())); // Check if the cluster name contains thread id. If not, append. // We can not just append the threadId, since some of the tests are crating instances // using configurations derived from configurations returned by this factory. - String clusterName = configuration.getClusterName(); +// String clusterName = configuration.getClusterName(); // append thread id - if (clusterName.indexOf(Thread.currentThread().getName()) == -1) { - clusterName = clusterName + "-" + Thread.currentThread().getName(); -// System.out.println(getThreadId() + " Setting cluster name " + newClusterName); - } +// if (clusterName.indexOf(Thread.currentThread().getName()) == -1) { +// clusterName = clusterName + "-" + Thread.currentThread().getName(); +//// System.out.println(getThreadId() + " Setting cluster name " + newClusterName); +// } // String testName = extractTestName(); @@ -212,7 +213,7 @@ } */ - configuration.setClusterName(clusterName); +// configuration.setClusterName(clusterName); } @@ -291,7 +292,9 @@ } private String getDefaultClusterConfiguration() { - return getClusterConfigFromFile(new Configuration().getDefaultClusterConfig()); + // TODO: fix this to use the cache manager +// return getClusterConfigFromFile(new Configuration().getDefaultClusterConfig()); + return null; } /** Modified: core/branches/flat/src/test/java/org/horizon/api/CacheAPITest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/api/CacheAPITest.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/test/java/org/horizon/api/CacheAPITest.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -71,7 +71,7 @@ public void testGetMembersInLocalMode() { CacheSPI<String, String> cache = cacheTL.get(); - assert cache.getRPCManager().getLocalAddress() == null : "Cache members should be null if running in LOCAL mode"; + assert cache.getRPCManager().getAddress() == null : "Cache members should be null if running in LOCAL mode"; } public void testConvenienceMethods() { Modified: core/branches/flat/src/test/java/org/horizon/api/mvcc/PutForExternalReadTest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/api/mvcc/PutForExternalReadTest.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/test/java/org/horizon/api/mvcc/PutForExternalReadTest.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -11,11 +11,12 @@ import org.horizon.config.Configuration; import org.horizon.factories.ComponentRegistry; import org.horizon.remoting.RPCManager; +import org.horizon.remoting.ResponseMode; +import org.horizon.remoting.transport.Address; import org.horizon.transaction.DummyTransactionManagerLookup; import org.horizon.transaction.TransactionTable; import org.horizon.util.TestingUtil; import org.horizon.util.internals.ReplicationListener; -import org.jgroups.Address; import static org.testng.AssertJUnit.*; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; @@ -24,7 +25,6 @@ import javax.transaction.Transaction; import javax.transaction.TransactionManager; import java.util.List; -import java.util.Vector; @Test(groups = {"functional", "jgroups", "transaction"}) public class PutForExternalReadTest { @@ -48,7 +48,8 @@ c.setCacheMode(Configuration.CacheMode.REPL_SYNC); cache1 = (CacheSPI<String, String>) cf.createCache(c, false); cache1.getConfiguration().setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); - cache1.getConfiguration().setSerializationExecutorPoolSize(0);//this is very important for async tests! + // TODO fix this +// cache1.getConfiguration().setSerializationExecutorPoolSize(0);//this is very important for async tests! cache1.start(); tm1 = cache1.getConfiguration().getRuntimeConfig().getTransactionManager(); @@ -95,11 +96,16 @@ assertEquals("PFER should have been a no-op", value, cache2.get(key)); } - private Vector<Address> anyAddresses() { + private List<Address> anyAddresses() { anyObject(); return null; } + private ResponseMode anyResponseMode() { + anyObject(); + return null; + } + public void testAsyncForce() throws Exception { RPCManager rpcManager = EasyMock.createNiceMock(RPCManager.class); RPCManager originalRpcManager = cache1.getConfiguration().getRuntimeConfig().getRPCManager(); @@ -112,7 +118,7 @@ // specify what we expectWithTx called on the mock Rpc Manager. For params we don't care about, just use ANYTHING. // setting the mock object to expectWithTx the "sync" param to be false. - expect(rpcManager.callRemoteMethods(anyAddresses(), (ReplicableCommand) anyObject(), eq(false), anyLong(), anyBoolean())).andReturn(null); + expect(rpcManager.invokeRemotely(anyAddresses(), (ReplicableCommand) anyObject(), eq(ResponseMode.ASYNCHRONOUS), anyLong(), anyBoolean())).andReturn(null); replay(rpcManager); @@ -155,8 +161,8 @@ try { List<Address> memberList = originalRpcManager.getMembers(); expect(barfingRpcManager.getMembers()).andReturn(memberList).anyTimes(); - expect(barfingRpcManager.getLocalAddress()).andReturn(originalRpcManager.getLocalAddress()).anyTimes(); - expect(barfingRpcManager.callRemoteMethods(anyAddresses(), (ReplicableCommand) anyObject(), anyBoolean(), anyLong(), anyBoolean())).andThrow(new RuntimeException("Barf!")).anyTimes(); + expect(barfingRpcManager.getAddress()).andReturn(originalRpcManager.getAddress()).anyTimes(); + expect(barfingRpcManager.invokeRemotely(anyAddresses(), (ReplicableCommand) anyObject(), anyResponseMode(), anyLong(), anyBoolean())).andThrow(new RuntimeException("Barf!")).anyTimes(); replay(barfingRpcManager); TestingUtil.extractComponentRegistry(cache1).registerComponent(barfingRpcManager, RPCManager.class); Modified: core/branches/flat/src/test/java/org/horizon/api/tree/SyncReplTxTest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/api/tree/SyncReplTxTest.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/test/java/org/horizon/api/tree/SyncReplTxTest.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -109,6 +109,6 @@ } private void assertClusterSize(String message, int size, Cache<Object, Object> c) { - assertEquals(message, size, c.getMembers().size()); + assertEquals(message, size, c.getCacheManager().getMembers().size()); } } \ No newline at end of file Modified: core/branches/flat/src/test/java/org/horizon/api/tree/TreeCacheAPITest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/api/tree/TreeCacheAPITest.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/test/java/org/horizon/api/tree/TreeCacheAPITest.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -3,7 +3,7 @@ import org.horizon.Cache; import org.horizon.CacheSPI; import org.horizon.config.Configuration; -import org.horizon.manager.CacheManager; +import org.horizon.manager.DefaultCacheManager; import org.horizon.transaction.DummyTransactionManagerLookup; import org.horizon.tree.Fqn; import org.horizon.tree.Node; @@ -36,7 +36,7 @@ Configuration c = new Configuration(); c.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); c.setInvocationBatchingEnabled(true); - CacheManager cm = new CacheManager(c); + DefaultCacheManager cm = new DefaultCacheManager(c); Cache flatcache = cm.getCache(); cache = new TreeCacheImpl(flatcache); @@ -222,7 +222,7 @@ } public void testRpcManagerElements() { - assertEquals("CacheMode.LOCAL cache has no address", null, cache.getLocalAddress()); + assertEquals("CacheMode.LOCAL cache has no address", null, cache.getAddress()); assertEquals("CacheMode.LOCAL cache has no members list", null, cache.getMembers()); } } Modified: core/branches/flat/src/test/java/org/horizon/replication/AsyncReplTest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/replication/AsyncReplTest.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/test/java/org/horizon/replication/AsyncReplTest.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -67,16 +67,6 @@ // default does nothing } - /** - * Provides a hook to check that the cache's channel came from the multiplexer, or not, as expected. This default - * impl asserts that the channel did not come from the multiplexer. - * - * @param cache a cache that has already been started - */ - protected void validateMultiplexer(Cache cache) { - assertFalse("Cache is not using multiplexer", cache.getConfiguration().isUsingMultiplexer()); - } - @AfterMethod(alwaysRun = true) public void tearDown() throws Exception { AsyncReplTestTL tl = threadLocal.get(); @@ -135,7 +125,8 @@ CacheSPI<Object, Object> cache3 = null, cache4 = null; try { - tl.configuration.setClusterName("otherTest"); + // TODO: fix this +// tl.configuration.setClusterName("otherTest"); cache3 = (CacheSPI<Object, Object>) new UnitTestCacheManager(tl.configuration).createCache("testCache"); cache4 = (CacheSPI<Object, Object>) new UnitTestCacheManager(tl.configuration).createCache("testCache"); replListener2.expectAny(); Modified: core/branches/flat/src/test/java/org/horizon/replication/ExceptionTest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/replication/ExceptionTest.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/test/java/org/horizon/replication/ExceptionTest.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -43,7 +43,8 @@ c.setSyncCommitPhase(true); c.setSyncRollbackPhase(true); c.setCacheMode(Configuration.CacheMode.REPL_SYNC); - if (jgroupsConfig != null) c.setClusterConfig(jgroupsConfig); + // TODO fix this +// if (jgroupsConfig != null) c.setClusterConfig(jgroupsConfig); c.setTransactionManagerLookupClass(DummyTransactionManagerLookup.class.getName()); Cache cache = new UnitTestCacheManager(c).createCache("testCache"); return cache; Modified: core/branches/flat/src/test/java/org/horizon/replication/SyncCacheListenerTest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/replication/SyncCacheListenerTest.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/test/java/org/horizon/replication/SyncCacheListenerTest.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -16,7 +16,6 @@ import org.horizon.logging.LogFactory; import org.horizon.notifications.annotation.CacheEntryModified; import org.horizon.notifications.annotation.CacheEntryRemoved; -import org.horizon.notifications.annotation.CacheListener; import org.horizon.notifications.event.Event; import org.horizon.notifications.event.TransactionalEvent; import org.horizon.transaction.DummyTransactionManager; @@ -89,7 +88,7 @@ tm.begin(); Transaction tx = tm.getTransaction(); Listener lis = new Listener(); - cache1.getNotifier().addCacheListener(lis); + cache1.getNotifier().addListener(lis); lis.put("age", 38); tm.suspend(); @@ -106,7 +105,7 @@ public void testRemoteCacheListener() throws Exception { Integer age; RemoteListener lis = new RemoteListener(); - cache2.getNotifier().addCacheListener(lis); + cache2.getNotifier().addListener(lis); cache1.put("age", 38); // value on cache2 must be 38 @@ -119,7 +118,7 @@ public void testSyncRepl() throws Exception { Integer age; Listener lis = new Listener(); - cache1.addCacheListener(lis); + cache1.addListener(lis); lis.put("age", 38); // value on cache2 must be 38 @@ -169,7 +168,7 @@ Integer age; Listener lis = new Listener(); - cache1.getNotifier().addCacheListener(lis); + cache1.getNotifier().addListener(lis); lis.put("age", 38); cache1.put("name", "Ben"); @@ -179,7 +178,7 @@ assertTrue("\"age\" must be 38", age == 38); } - @CacheListener + @org.horizon.notifications.annotation.Listener public class Listener { Object key = null; @@ -213,7 +212,7 @@ } - @CacheListener + @org.horizon.notifications.annotation.Listener public class RemoteListener { @CacheEntryRemoved Modified: core/branches/flat/src/test/java/org/horizon/replication/SyncReplTest.java =================================================================== --- core/branches/flat/src/test/java/org/horizon/replication/SyncReplTest.java 2009-01-19 15:43:55 UTC (rev 7512) +++ core/branches/flat/src/test/java/org/horizon/replication/SyncReplTest.java 2009-01-19 19:02:51 UTC (rev 7513) @@ -79,6 +79,6 @@ } private void assertClusterSize(String message, int size, Cache c) { - assertEquals(message, size, c.getMembers().size()); + assertEquals(message, size, c.getCacheManager().getMembers().size()); } }