From 2e415e1bd61be7daba650b86d2f478436fc9fa2f Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Fri, 26 Aug 2011 21:18:37 +0300 Subject: [PATCH] SGF-57 + fix some initilization problem with the container that triggered eager registration of cqs + added more integration tests --- .../ContinousQueryListenerContainer.java | 24 +++++++++++++------ .../listener/ListenerContainerTests.java | 9 +++++-- .../adapter/ContainerXmlSetupTest.java | 13 ++++++---- .../data/gemfire/listener/container.xml | 2 +- 4 files changed, 33 insertions(+), 15 deletions(-) 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 b847400d..3cfd290d 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/ContinousQueryListenerContainer.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinousQueryListenerContainer.java @@ -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 defs = new LinkedHashSet(); private Set queries = new ConcurrentHashSet(); 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 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) { 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 3d295899..05a13382 100644 --- a/src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java +++ b/src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java @@ -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(); 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 1414e4cb..2882fb9d 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 @@ -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); } } 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 e3351047..f3791508 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 - +