SGF-57
SGF-63 + improved schema + fix some bugs that prevented the client cache to reuse an already defined pool
This commit is contained in:
@@ -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<InetSocketAddress> servers = p.getServers();
|
||||
if (locators != null) {
|
||||
for (InetSocketAddress inet : servers) {
|
||||
ccf.addPoolLocator(inet.getHostName(), inet.getPort());
|
||||
ccf.addPoolServer(inet.getHostName(), inet.getPort());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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);
|
||||
|
||||
|
||||
@@ -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<CqQuery> queries = new ConcurrentHashSet<CqQuery>();
|
||||
|
||||
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.
|
||||
*
|
||||
|
||||
@@ -1083,6 +1083,13 @@ and stop as soon as possible.
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="pool-name" use="optional" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
The name of the pool used by the container.
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
|
||||
@@ -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));
|
||||
|
||||
@@ -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));
|
||||
|
||||
@@ -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());
|
||||
|
||||
|
||||
@@ -16,7 +16,7 @@
|
||||
<prop key="log-level">warning</prop>
|
||||
</util:properties>
|
||||
|
||||
<gfe:cache properties-ref="props" use-bean-factory-locator="false"/>
|
||||
<gfe:client-cache use-bean-factory-locator="false" pool-name="client"/>
|
||||
|
||||
<gfe:pool id="client" subscription-enabled="true">
|
||||
<gfe:server host="localhost" port="40404"/>
|
||||
@@ -24,7 +24,7 @@
|
||||
|
||||
<task:executor id="testTaskExecutor" />
|
||||
|
||||
<gfe:cq-listener-container cache="gemfire-cache">
|
||||
<gfe:cq-listener-container cache="gemfire-cache" pool-name="client">
|
||||
<!-- default handle method -->
|
||||
<gfe:listener ref="testBean1" query="SELECT * from /test-cq"/>
|
||||
<gfe:listener ref="testBean1" query="SELECT * from /test-cq" name="test-bean-1" method="handleQuery"/>
|
||||
|
||||
Reference in New Issue
Block a user