+ fix some initilization problem with the container that triggered eager registration of cqs
+ added more integration tests
This commit is contained in:
Costin Leau
2011-08-26 21:18:37 +03:00
parent ed0ca63c55
commit 2e415e1bd6
4 changed files with 33 additions and 15 deletions

View File

@@ -16,6 +16,7 @@
package org.springframework.data.gemfire.listener;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.Executor;
@@ -91,6 +92,7 @@ public class ContinousQueryListenerContainer implements InitializingBean, Dispos
private volatile boolean initialized = false;
private volatile boolean manageExecutor = false;
private Set<ContinousQueryDefinition> defs = new LinkedHashSet<ContinousQueryDefinition>();
private Set<CqQuery> queries = new ConcurrentHashSet<CqQuery>();
private QueryService queryService;
@@ -113,6 +115,7 @@ public class ContinousQueryListenerContainer implements InitializingBean, Dispos
queryService = pool.getQueryService();
}
initMapping(defs);
initialized = true;
start();
@@ -333,7 +336,8 @@ public class ContinousQueryListenerContainer implements InitializingBean, Dispos
* @param queries set of queries
*/
public void setQueryListeners(Set<ContinousQueryDefinition> queries) {
initMapping(queries);
defs.clear();
defs.addAll(queries);
}
/**
@@ -355,7 +359,7 @@ public class ContinousQueryListenerContainer implements InitializingBean, Dispos
closeQueries();
for (ContinousQueryDefinition def : queryDefinitions) {
doAddListener(def);
addCQuery(def);
}
// resume activity
@@ -365,29 +369,35 @@ public class ContinousQueryListenerContainer implements InitializingBean, Dispos
}
private void doAddListener(ContinousQueryDefinition def) {
CqQuery cq = null;
CqQuery cq = addCQuery(def);
if (isRunning()) {
executeQuery(cq);
}
}
private CqQuery addCQuery(ContinousQueryDefinition def) {
try {
CqAttributesFactory caf = new CqAttributesFactory();
caf.addCqListener(new EventDispatcherAdapter(def.getListener()));
CqAttributes attr = caf.create();
CqQuery cq = null;
if (StringUtils.hasText(def.getName())) {
cq = queryService.newCq(def.getName(), def.getQuery(), attr, def.isDurable());
}
else {
cq = queryService.newCq(def.getQuery(), attr, def.isDurable());
}
queries.add(cq);
return cq;
} catch (RuntimeException ex) {
throw new GemfireQueryException("Cannot create query ", ex);
} catch (QueryException ex) {
throw new GemfireQueryException("Cannot create query ", ex);
}
if (isRunning()) {
executeQuery(cq);
}
}
private void executeQuery(CqQuery cq) {

View File

@@ -64,6 +64,11 @@ public class ListenerContainerTests {
ccf.setPoolSubscriptionEnabled(true);
cache = ccf.create();
// not really used but here just for future tests :)
// PoolFactory pf = PoolManager.createFactory();
// pf.addServer("localhost", 40404);
// pf.setSubscriptionEnabled(true);
// Pool pool = pf.create("client");
}
@@ -89,12 +94,12 @@ public class ListenerContainerTests {
container = new ContinousQueryListenerContainer();
container.setCache(cache);
//container.setPoolName("client");
container.setBeanName("container");
container.addListener(new ContinousQueryDefinition("test", query, adapter));
container.afterPropertiesSet();
container.addListener(new ContinousQueryDefinition("test", query, adapter));
}
@Test
public void testContainer() throws Exception {
ForkUtil.sendSignal();

View File

@@ -16,6 +16,7 @@
package org.springframework.data.gemfire.listener.adapter;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import org.junit.AfterClass;
@@ -26,8 +27,7 @@ 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.PoolFactory;
import com.gemstone.gemfire.cache.client.PoolManager;
import com.gemstone.gemfire.cache.client.Pool;
import com.gemstone.gemfire.cache.query.CqQuery;
/**
@@ -54,12 +54,15 @@ public class ContainerXmlSetupTest {
ContinousQueryListenerContainer container = ctx.getBean(ContinousQueryListenerContainer.class);
assertTrue(container.isRunning());
ForkUtil.sendSignal();
Thread.sleep(3000);
Cache cache = ctx.getBean("gemfire-cache", Cache.class);
Pool pool = ctx.getBean("client", Pool.class);
CqQuery[] cqs = cache.getQueryService().getCqs();
System.out.println("Cqs " + cqs.length);
CqQuery[] pcqs = pool.getQueryService().getCqs();
assertTrue(pool.getQueryService().getCq("test-bean-1") != null);
assertEquals(3, cqs.length);
assertEquals(3, pcqs.length);
ForkUtil.sendSignal();
Thread.sleep(3000);
}
}

View File

@@ -16,7 +16,7 @@
<prop key="log-level">warning</prop>
</util:properties>
<gfe:client-cache use-bean-factory-locator="false" pool-name="client"/>
<gfe:client-cache use-bean-factory-locator="false"/>
<gfe:pool id="client" subscription-enabled="true">
<gfe:server host="localhost" port="40404"/>