Author: manik.surtani(a)jboss.com Date: 2008-10-14 05:38:28 -0400 (Tue, 14 Oct 2008) New Revision: 6928 Modified: core/trunk/src/test/java/org/jboss/cache/UnitTestCacheFactory.java core/trunk/src/test/java/org/jboss/cache/util/internals/ReplicationListener.java Log: Made this quieter + fixed ReplList Modified: core/trunk/src/test/java/org/jboss/cache/UnitTestCacheFactory.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/UnitTestCacheFactory.java 2008-10-14 09:23:28 UTC (rev 6927) +++ core/trunk/src/test/java/org/jboss/cache/UnitTestCacheFactory.java 2008-10-14 09:38:28 UTC (rev 6928) @@ -6,11 +6,6 @@ */ package org.jboss.cache; -import java.io.InputStream; -import java.util.Vector; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.regex.Matcher; -import java.util.regex.Pattern; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.jboss.cache.config.Configuration; @@ -20,8 +15,13 @@ import org.jboss.cache.factories.UnitTestCacheConfigurationFactory; import org.jboss.cache.util.TestingUtil; +import java.io.InputStream; +import java.util.Vector; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.regex.Matcher; +import java.util.regex.Pattern; + /** - * * @author <a href="mailto:dpospisi@redhat.com">Dominik Pospisil (dpospisi(a)redhat.com)&lt;/a&gt; */ public class UnitTestCacheFactory<K, V> implements CacheFactory<K, V> @@ -29,64 +29,78 @@ private final Log log = LogFactory.getLog(UnitTestCacheFactory.class); - + /** - * Holds unique mcast_addr for each thread used for JGroups channel construction. + * Holds unique mcast_addr for each thread used for JGroups channel construction. */ - private static final ThreadLocal<String> threadMcastIP = new ThreadLocal<String>() { + private static final ThreadLocal<String> threadMcastIP = new ThreadLocal<String>() + { - private final AtomicInteger uniqueAddr = new AtomicInteger(11); - @Override protected String initialValue() { + private final AtomicInteger uniqueAddr = new AtomicInteger(11); + + @Override + protected String initialValue() + { return "228.10.10." + String.valueOf(uniqueAddr.getAndIncrement()); - } + } }; - + /** - * Holds unique mcast_port for each thread used for JGroups channel construction. + * Holds unique mcast_port for each thread used for JGroups channel construction. */ - private static final ThreadLocal<Integer> threadMcastPort = new ThreadLocal<Integer>() { - + private static final ThreadLocal<Integer> threadMcastPort = new ThreadLocal<Integer>() + { + private final AtomicInteger uniquePort = new AtomicInteger(45589); - - @Override protected Integer initialValue() { + + @Override + protected Integer initialValue() + { return uniquePort.getAndIncrement(); - } + } }; - + /** - * For each thread holds list of caches created using this factory. + * For each thread holds list of caches created using this factory. */ private static final ThreadLocal<Vector<Cache>> threadCaches = - new ThreadLocal<Vector<Cache>>() { - - - @Override protected Vector<Cache> initialValue() { - return new Vector<Cache>(); - } - }; - + new ThreadLocal<Vector<Cache>>() + { + + + @Override + protected Vector<Cache> initialValue() + { + return new Vector<Cache>(); + } + }; + private static Vector<Cache> allCaches = new Vector<Cache>(); - + /** * For each thread holds the name of the test class which executed createCache factory method. */ private static final ThreadLocal<String> threadTestName = new ThreadLocal<String>(); - + // factory methods - - public Cache<K, V> createCache() throws ConfigurationException { + + public Cache<K, V> createCache() throws ConfigurationException + { return createCache(true); } - public Cache<K, V> createCache(boolean start) throws ConfigurationException { + public Cache<K, V> createCache(boolean start) throws ConfigurationException + { return createCache(new Configuration(), start); } - public Cache<K, V> createCache(String configFileName) throws ConfigurationException { + public Cache<K, V> createCache(String configFileName) throws ConfigurationException + { return createCache(configFileName, true); } - public Cache<K, V> createCache(String configFileName, boolean start) throws ConfigurationException { + public Cache<K, V> createCache(String configFileName, boolean start) throws ConfigurationException + { XmlConfigurationParser parser = new XmlConfigurationParser(); Configuration c; try @@ -102,30 +116,35 @@ return createCache(c, start); } - public Cache<K, V> createCache(Configuration configuration) throws ConfigurationException { + public Cache<K, V> createCache(Configuration configuration) throws ConfigurationException + { return createCache(configuration, true); } - public Cache<K, V> createCache(InputStream is) throws ConfigurationException { + public Cache<K, V> createCache(InputStream is) throws ConfigurationException + { return createCache(is, true); } - public Cache<K, V> createCache(InputStream is, boolean start) throws ConfigurationException { + public Cache<K, V> createCache(InputStream is, boolean start) throws ConfigurationException + { XmlConfigurationParser parser = new XmlConfigurationParser(); Configuration c = parser.parseStream(is); return createCache(c, start); } - - public Cache<K, V> createCache(Configuration configuration, boolean start) throws ConfigurationException { - - if (! Thread.currentThread().getName().contains("pool")) { + + public Cache<K, V> createCache(Configuration configuration, boolean start) throws ConfigurationException + { + + if (!Thread.currentThread().getName().contains("pool")) + { System.out.println("CreateCache called from wrong thread: " + Thread.currentThread().getName()); //Thread.dumpStack(); //System.exit(-1); } - + checkCaches(); - + switch (configuration.getCacheMode()) { case LOCAL: @@ -141,74 +160,86 @@ default: log.info("Unknown cache mode!"); } - - Cache<K, V> cache = new DefaultCacheFactory<K, V>().createCache(configuration, start); + Cache<K, V> cache = new DefaultCacheFactory<K, V>().createCache(configuration, start); + Vector<Cache> caches = threadCaches.get(); caches.add(cache); - - synchronized (allCaches) { + + synchronized (allCaches) + { allCaches.add(cache); } return cache; - + } - + /** * Destroys all caches created by this factory in the current thread. - * + * * @return true if some cleanup was actually performed */ - public boolean cleanUp() { + public boolean cleanUp() + { Vector<Cache> caches = new Vector<Cache>(threadCaches.get()); boolean ret = false; - - for (Cache cache : caches) { - TestingUtil.killCaches(cache); + + for (Cache cache : caches) + { + TestingUtil.killCaches(cache); ret = true; } return ret; } - - public void removeCache(Cache c) { - - if (! Thread.currentThread().getName().contains("pool")) { + + public void removeCache(Cache c) + { + + if (!Thread.currentThread().getName().contains("pool")) + { System.out.println("RemoveCache called from wrong thread."); //Thread.dumpStack(); //System.exit(-1); } - + Vector<Cache> caches = threadCaches.get(); - synchronized (allCaches) { - if (caches.contains(c)) { + synchronized (allCaches) + { + if (caches.contains(c)) + { caches.remove(c); allCaches.remove(c); - } else if (allCaches.contains(c)) { + } + else if (allCaches.contains(c)) + { System.out.println("[" + Thread.currentThread().getName() + "] WARNING!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! Remove cache called from different thread."); Thread.dumpStack(); } } } - + /** * Updates cluster configuration to ensure mutual thread isolation. + * * @param configuration Configuration to update. */ - public void mangleConfiguration(Configuration configuration) { - - configuration.setClusterConfig(mangleClusterConfiguration(configuration.getClusterConfig())); + public void mangleConfiguration(Configuration configuration) + { + + 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(); - + // append thread id - if (clusterName.indexOf(Thread.currentThread().getName()) == -1) { + if (clusterName.indexOf(Thread.currentThread().getName()) == -1) + { clusterName = clusterName + "-" + Thread.currentThread().getName(); // System.out.println(getThreadId() + " Setting cluster name " + newClusterName); } - + String testName = extractTestName(); // prepend test name @@ -217,86 +248,101 @@ clusterName = testName + "-" + clusterName; } */ - - configuration.setClusterName(clusterName); - + + configuration.setClusterName(clusterName); + } /** * Updates cluster configuration to ensure mutual thread isolation. + * * @param configuration Configuration to update. */ - public String mangleClusterConfiguration(String clusterConfig) { - if (clusterConfig == null) { + public String mangleClusterConfiguration(String clusterConfig) + { + if (clusterConfig == null) + { // No explicit cluster configuration found. we need to resolve the default config // now in orded to be able to update it before the cache (and the channel) starts. // TODO: this does not seem to be the best solution :( clusterConfig = UnitTestCacheConfigurationFactory.getClusterConfigFromFile( - new Configuration().getDefaultClusterConfig()); - + new Configuration().getDefaultClusterConfig()); + } - - if (Thread.currentThread().getName().equals("main")) { - Exception e = new Exception("Invoked from main thread."); - e.printStackTrace(); - }; - + + // - Do not print out this exception - since tests are ALSO run from IDEs on the main thread. - MANIK +// +// if (Thread.currentThread().getName().equals("main")) { +// Exception e = new Exception("Invoked from main thread."); +// e.printStackTrace(); +// }; + // replace mcast_addr Pattern pattern = Pattern.compile("mcast_addr=[^;]*"); Matcher m = pattern.matcher(clusterConfig); - if (m.find()) { + if (m.find()) + { String origAddr = m.group().substring(m.group().indexOf("=") + 1); String newAddr = threadMcastIP.get(); // System.out.println(getThreadId() + " Replacing mcast_addr " + origAddr + " with " + newAddr); clusterConfig = m.replaceFirst("mcast_addr=" + newAddr); - } else { + } + else + { Thread.dumpStack(); System.exit(1); } - + // replace mcast_port pattern = Pattern.compile("mcast_port=[^;]*"); m = pattern.matcher(clusterConfig); - if (m.find()) { + if (m.find()) + { String origPort = m.group().substring(m.group().indexOf("=") + 1); String newPort = threadMcastPort.get().toString(); - // System.out.println(getThreadId() + " Replacing mcast_port " + origPort + " with " + newPort); + // System.out.println(getThreadId() + " Replacing mcast_port " + origPort + " with " + newPort); clusterConfig = m.replaceFirst("mcast_port=" + newPort); } - + return clusterConfig; } - - private String getThreadId() { + + private String getThreadId() + { return "[" + Thread.currentThread().getName() + "]"; } - private void checkCaches() { + private void checkCaches() + { String lastTestName = threadTestName.get(); - String currentTestName = extractTestName(); - - if ((lastTestName != null) && (! lastTestName.equals(currentTestName))) { + String currentTestName = extractTestName(); + if ((lastTestName != null) && (!lastTestName.equals(currentTestName))) + { + String threadId = "[" + Thread.currentThread().getName() + "] "; // we are running new test class // check if there is a cache(s) instance left & kill it if possitive - - if (cleanUp()) { + + if (cleanUp()) + { System.out.print(threadId + "WARNING!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! "); - System.out.print(threadId +" A test method in " + lastTestName + " did not clean all cache instances properly. "); - System.out.println(threadId +" Use UnitTestCacheFactory.cleanUp() or TestngUtil.killCaches(...) "); + System.out.print(threadId + " A test method in " + lastTestName + " did not clean all cache instances properly. "); + System.out.println(threadId + " Use UnitTestCacheFactory.cleanUp() or TestngUtil.killCaches(...) "); } - + } threadTestName.set(currentTestName); } - - private String extractTestName() { + + private String extractTestName() + { StackTraceElement[] stack = Thread.currentThread().getStackTrace(); if (stack.length == 0) return null; - for (int i=stack.length-1; i>0; i--) { + for (int i = stack.length - 1; i > 0; i--) + { StackTraceElement e = stack[i]; String className = e.getClassName(); if (className.indexOf("org.jboss.cache") != -1) return className; //+ "." + e.getMethodName(); Modified: core/trunk/src/test/java/org/jboss/cache/util/internals/ReplicationListener.java =================================================================== --- core/trunk/src/test/java/org/jboss/cache/util/internals/ReplicationListener.java 2008-10-14 09:23:28 UTC (rev 6927) +++ core/trunk/src/test/java/org/jboss/cache/util/internals/ReplicationListener.java 2008-10-14 09:38:28 UTC (rev 6928) @@ -11,6 +11,7 @@ import org.jboss.cache.io.ByteBuffer; import org.jboss.cache.marshall.AbstractMarshaller; import org.jboss.cache.marshall.CommandAwareRpcDispatcher; +import org.jboss.cache.marshall.InactiveRegionAwareRpcDispatcher; import org.jboss.cache.marshall.Marshaller; import org.jboss.cache.marshall.RegionalizedMethodCall; import org.jboss.cache.util.TestingUtil; @@ -61,7 +62,11 @@ RPCManager rpcManager = componentRegistry.getComponent(RPCManager.class); CommandAwareRpcDispatcher realDispatcher = (CommandAwareRpcDispatcher) TestingUtil.extractField(rpcManager, "rpcDispatcher"); RpcDispatcher.Marshaller2 realMarshaller = (RpcDispatcher.Marshaller2) realDispatcher.getMarshaller(); - MarshallerDelegate delegate = new MarshallerDelegate(realMarshaller); + RpcDispatcher.Marshaller2 delegate = null; + if (realDispatcher instanceof InactiveRegionAwareRpcDispatcher) + delegate = new RegionMarshallerDelegate((Marshaller) realMarshaller); + else + delegate = new MarshallerDelegate(realMarshaller); realDispatcher.setMarshaller(delegate); realDispatcher.setRequestMarshaller(delegate); realDispatcher.setResponseMarshaller(delegate); @@ -204,11 +209,6 @@ return realOne.regionalizedMethodCallFromObjectStream(in); } - public Object objectFromByteBuffer(byte[] bytes) throws Exception - { - return realOne.objectFromByteBuffer(bytes); - } - public ByteBuffer objectToBuffer(Object o) throws Exception { return realOne.objectToBuffer(o);