From ed0ca63c550c6c0f4e5c8d4f3f376689951cdc3d Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Fri, 26 Aug 2011 20:46:38 +0300 Subject: [PATCH] SGF-57 SGF-63 + improved schema + fix some bugs that prevented the client cache to reuse an already defined pool --- .../client/ClientCacheFactoryBean.java | 3 ++- .../GemfireListenerContainerParser.java | 7 +++--- .../ContinousQueryListenerContainer.java | 22 ++++++++++++++++++- .../gemfire/config/spring-gemfire-1.1.xsd | 7 ++++++ .../data/gemfire/fork/CacheServerProcess.java | 6 ++++- .../listener/ListenerContainerTests.java | 21 ++++-------------- .../adapter/ContainerXmlSetupTest.java | 6 ----- .../data/gemfire/listener/container.xml | 4 ++-- 8 files changed, 44 insertions(+), 32 deletions(-) diff --git a/src/main/java/org/springframework/data/gemfire/client/ClientCacheFactoryBean.java b/src/main/java/org/springframework/data/gemfire/client/ClientCacheFactoryBean.java index 7c5d8a14..4c660dc2 100644 --- a/src/main/java/org/springframework/data/gemfire/client/ClientCacheFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/client/ClientCacheFactoryBean.java @@ -103,6 +103,7 @@ public class ClientCacheFactoryBean extends CacheFactoryBean { + Pool.class.getName() + " found"); p = getBeanFactory().getBean(poolName, Pool.class); + Assert.notNull(p, "No pool named [" + poolName + "] found"); } if (p != null) { @@ -137,7 +138,7 @@ public class ClientCacheFactoryBean extends CacheFactoryBean { List servers = p.getServers(); if (locators != null) { for (InetSocketAddress inet : servers) { - ccf.addPoolLocator(inet.getHostName(), inet.getPort()); + ccf.addPoolServer(inet.getHostName(), inet.getPort()); } } } diff --git a/src/main/java/org/springframework/data/gemfire/config/GemfireListenerContainerParser.java b/src/main/java/org/springframework/data/gemfire/config/GemfireListenerContainerParser.java index 89ac1ab4..607d7574 100644 --- a/src/main/java/org/springframework/data/gemfire/config/GemfireListenerContainerParser.java +++ b/src/main/java/org/springframework/data/gemfire/config/GemfireListenerContainerParser.java @@ -60,10 +60,9 @@ class GemfireListenerContainerParser extends AbstractSimpleBeanDefinitionParser } } - String phase = element.getAttribute("phase"); - if (StringUtils.hasText(phase)) { - builder.addPropertyValue("phase", phase); - } + + ParsingUtils.setPropertyValue(element, builder, "phase", "phase"); + ParsingUtils.setPropertyValue(element, builder, "pool-name", "poolName"); postProcess(builder, element); diff --git a/src/main/java/org/springframework/data/gemfire/listener/ContinousQueryListenerContainer.java b/src/main/java/org/springframework/data/gemfire/listener/ContinousQueryListenerContainer.java index 5b6c017e..b847400d 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/ContinousQueryListenerContainer.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinousQueryListenerContainer.java @@ -28,11 +28,14 @@ import org.springframework.context.SmartLifecycle; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.core.task.TaskExecutor; import org.springframework.data.gemfire.GemfireQueryException; +import org.springframework.util.Assert; import org.springframework.util.ClassUtils; import org.springframework.util.ErrorHandler; import org.springframework.util.StringUtils; import com.gemstone.gemfire.cache.RegionService; +import com.gemstone.gemfire.cache.client.Pool; +import com.gemstone.gemfire.cache.client.PoolManager; import com.gemstone.gemfire.cache.query.CqAttributes; import com.gemstone.gemfire.cache.query.CqAttributesFactory; import com.gemstone.gemfire.cache.query.CqEvent; @@ -74,7 +77,8 @@ public class ContinousQueryListenerContainer implements InitializingBean, Dispos /** * Default thread name prefix: "ContinousQueryListenerContainer-". */ - public static final String DEFAULT_THREAD_NAME_PREFIX = ClassUtils.getShortName(ContinousQueryListenerContainer.class) + "-"; + public static final String DEFAULT_THREAD_NAME_PREFIX = ClassUtils.getShortName(ContinousQueryListenerContainer.class) + + "-"; private Executor subscriptionExecutor; private Executor taskExecutor; @@ -90,6 +94,7 @@ public class ContinousQueryListenerContainer implements InitializingBean, Dispos private Set queries = new ConcurrentHashSet(); private QueryService queryService; + private String poolName; public void afterPropertiesSet() { @@ -102,6 +107,12 @@ public class ContinousQueryListenerContainer implements InitializingBean, Dispos subscriptionExecutor = taskExecutor; } + if (StringUtils.hasText(poolName)) { + Pool pool = PoolManager.find(poolName); + Assert.notNull(pool, "No pool named [" + poolName + "] found"); + queryService = pool.getQueryService(); + } + initialized = true; start(); @@ -307,6 +318,15 @@ public class ContinousQueryListenerContainer implements InitializingBean, Dispos this.queryService = service; } + /** + * Set the name of the {@link Pool} used for performing the queries by this container. + * + * @param service + */ + public void setPoolName(String poolName) { + this.poolName = poolName; + } + /** * Attaches the given query definitions. * diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.1.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.1.xsd index c9fc5cfb..08d55ba0 100644 --- a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.1.xsd +++ b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.1.xsd @@ -1083,6 +1083,13 @@ and stop as soon as possible. ]]> + + + + + diff --git a/src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java b/src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java index 245ff53d..76ff596b 100644 --- a/src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java +++ b/src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java @@ -20,6 +20,9 @@ import java.io.BufferedReader; import java.io.InputStreamReader; import java.util.Properties; + +import org.springframework.data.gemfire.ForkUtil; + import com.gemstone.gemfire.cache.AttributesFactory; import com.gemstone.gemfire.cache.Cache; import com.gemstone.gemfire.cache.CacheFactory; @@ -57,7 +60,8 @@ public class CacheServerProcess { server.setNotifyBySubscription(true); server.start(); - + ForkUtil.createControlFile(CacheServerProcess.class.getName()); + System.out.println("Waiting for signal"); // wait for signal BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in)); diff --git a/src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java b/src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java index d7f2d17e..3d295899 100644 --- a/src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java +++ b/src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java @@ -24,15 +24,12 @@ import org.junit.AfterClass; import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; -import org.springframework.data.gemfire.CacheFactoryBean; import org.springframework.data.gemfire.ForkUtil; import org.springframework.data.gemfire.listener.adapter.ContinousQueryListenerAdapter; import com.gemstone.gemfire.cache.RegionService; -import com.gemstone.gemfire.cache.client.ClientCache; +import com.gemstone.gemfire.cache.client.ClientCacheFactory; import com.gemstone.gemfire.cache.client.Pool; -import com.gemstone.gemfire.cache.client.PoolFactory; -import com.gemstone.gemfire.cache.client.PoolManager; import com.gemstone.gemfire.cache.query.CqEvent; /** @@ -63,18 +60,10 @@ public class ListenerContainerTests { props.put("name", "cq-client"); props.put("log-level", "warning"); - CacheFactoryBean cacheFB = new CacheFactoryBean(); - cacheFB.setBeanName("gemfire-cache"); - cacheFB.setUseBeanFactoryLocator(false); - cacheFB.setProperties(props); - cacheFB.afterPropertiesSet(); + ClientCacheFactory ccf = new ClientCacheFactory(props); + ccf.setPoolSubscriptionEnabled(true); + cache = ccf.create(); - cache = cacheFB.getObject(); - - PoolFactory pf = PoolManager.createFactory(); - pf.addServer("localhost", 40404); - pf.setSubscriptionEnabled(true); - pool = pf.create("client"); } @@ -99,8 +88,6 @@ public class ListenerContainerTests { String query = "SELECT * from /test-cq"; container = new ContinousQueryListenerContainer(); - //container.setQueryService(pool.getQueryService()); - System.out.println(cache instanceof ClientCache); container.setCache(cache); container.setBeanName("container"); container.addListener(new ContinousQueryDefinition("test", query, adapter)); diff --git a/src/test/java/org/springframework/data/gemfire/listener/adapter/ContainerXmlSetupTest.java b/src/test/java/org/springframework/data/gemfire/listener/adapter/ContainerXmlSetupTest.java index 44942eac..1414e4cb 100644 --- a/src/test/java/org/springframework/data/gemfire/listener/adapter/ContainerXmlSetupTest.java +++ b/src/test/java/org/springframework/data/gemfire/listener/adapter/ContainerXmlSetupTest.java @@ -26,7 +26,6 @@ import org.springframework.data.gemfire.ForkUtil; import org.springframework.data.gemfire.listener.ContinousQueryListenerContainer; import com.gemstone.gemfire.cache.Cache; -import com.gemstone.gemfire.cache.client.Pool; import com.gemstone.gemfire.cache.client.PoolFactory; import com.gemstone.gemfire.cache.client.PoolManager; import com.gemstone.gemfire.cache.query.CqQuery; @@ -52,11 +51,6 @@ public class ContainerXmlSetupTest { GenericXmlApplicationContext ctx = new GenericXmlApplicationContext( "/org/springframework/data/gemfire/listener/container.xml"); - PoolFactory pf = PoolManager.createFactory(); - pf.addServer("localhost", 40404); - pf.setSubscriptionEnabled(true); - Pool pool = pf.create("client"); - ContinousQueryListenerContainer container = ctx.getBean(ContinousQueryListenerContainer.class); assertTrue(container.isRunning()); diff --git a/src/test/resources/org/springframework/data/gemfire/listener/container.xml b/src/test/resources/org/springframework/data/gemfire/listener/container.xml index e54d7ddb..e3351047 100644 --- a/src/test/resources/org/springframework/data/gemfire/listener/container.xml +++ b/src/test/resources/org/springframework/data/gemfire/listener/container.xml @@ -16,7 +16,7 @@ warning - + @@ -24,7 +24,7 @@ - +