diff --git a/.classpath b/.classpath index 438f0fe3..ee649b3c 100644 --- a/.classpath +++ b/.classpath @@ -1,10 +1,30 @@ - - - - - - - + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/src/main/java/org/springframework/data/gemfire/listener/QueryListenerContainer.java b/src/main/java/org/springframework/data/gemfire/listener/QueryListenerContainer.java index 3bf84a90..b25b3290 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/QueryListenerContainer.java +++ b/src/main/java/org/springframework/data/gemfire/listener/QueryListenerContainer.java @@ -16,7 +16,6 @@ package org.springframework.data.gemfire.listener; -import java.util.LinkedHashSet; import java.util.Set; import java.util.concurrent.Executor; @@ -41,6 +40,7 @@ import com.gemstone.gemfire.cache.query.CqListener; import com.gemstone.gemfire.cache.query.CqQuery; import com.gemstone.gemfire.cache.query.QueryException; import com.gemstone.gemfire.cache.query.QueryService; +import com.gemstone.gemfire.internal.concurrent.ConcurrentHashSet; /** * Container providing asynchronous behaviour for GemFire continuous queries. @@ -87,8 +87,9 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, private volatile boolean initialized = false; private volatile boolean manageExecutor = false; - private RegionService cache; - private Set queries; + private Set queries = new ConcurrentHashSet(); + + private QueryService queryService; public void afterPropertiesSet() { @@ -101,7 +102,6 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, subscriptionExecutor = taskExecutor; } - queries = new LinkedHashSet(); initialized = true; start(); @@ -195,19 +195,18 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, } private void closeQueries() { - if (queries != null) { - for (CqQuery cq : queries) { - try { - if (!cq.isClosed()) { - cq.close(); - } - } catch (QueryException ex) { - logger.warn("Cannot close query", ex); - } catch (RuntimeException ex) { - logger.warn("Cannot close query", ex); + for (CqQuery cq : queries) { + try { + if (!cq.isClosed()) { + cq.close(); } + } catch (QueryException ex) { + logger.warn("Cannot close query", ex); + } catch (RuntimeException ex) { + logger.warn("Cannot close query", ex); } } + queries.clear(); } @@ -296,7 +295,11 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, * @param cache cache used for registering queries */ public void setCache(RegionService cache) { - this.cache = cache; + this.queryService = cache.getQueryService(); + } + + public void setQueryService(QueryService service) { + this.queryService = service; } /** @@ -337,8 +340,6 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, } private void doAddListener(CqQueryDefinition def) { - QueryService qService = cache.getQueryService(); - CqQuery cq = null; try { @@ -347,10 +348,10 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, CqAttributes attr = caf.create(); if (StringUtils.hasText(def.getName())) { - cq = qService.newCq(def.getName(), def.getQuery(), attr, def.isDurable()); + cq = queryService.newCq(def.getName(), def.getQuery(), attr, def.isDurable()); } else { - cq = qService.newCq(def.getQuery(), attr, def.isDurable()); + cq = queryService.newCq(def.getQuery(), attr, def.isDurable()); } queries.add(cq); } catch (RuntimeException ex) { diff --git a/src/main/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapter.java b/src/main/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapter.java index 91c2af0b..40ea9a02 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapter.java +++ b/src/main/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapter.java @@ -85,6 +85,7 @@ public class QueryListenerAdapter implements QueryListener { ReflectionUtils.doWithMethods(c, new MethodCallback() { public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException { + ReflectionUtils.makeAccessible(method); methods.add(method); } @@ -131,7 +132,7 @@ public class QueryListenerAdapter implements QueryListener { } }); - Assert.isTrue(!methods.isEmpty(), "Cannot find a suitable methods named [" + methodName + Assert.isTrue(!methods.isEmpty(), "Cannot find a suitable method named [" + c.getName() + "#" + methodName + "] - is the method public and has the proper arguments?"); } diff --git a/src/main/java/org/springframework/data/gemfire/server/CacheServerFactoryBean.java b/src/main/java/org/springframework/data/gemfire/server/CacheServerFactoryBean.java index dea15af1..4e939433 100644 --- a/src/main/java/org/springframework/data/gemfire/server/CacheServerFactoryBean.java +++ b/src/main/java/org/springframework/data/gemfire/server/CacheServerFactoryBean.java @@ -34,7 +34,7 @@ import com.gemstone.gemfire.cache.server.ClientSubscriptionConfig; import com.gemstone.gemfire.cache.server.ServerLoadProbe; /** - * FactoryBean for easy creation and configuration of GemFire {@link CacheServer} instances. + * FactoryBean for easy creation and configuration of GemFire {@link CacheServerProcess} instances. * * @author Costin Leau */ diff --git a/src/main/java/org/springframework/data/gemfire/server/SubscriptionEvictionPolicy.java b/src/main/java/org/springframework/data/gemfire/server/SubscriptionEvictionPolicy.java index d7c1efda..b8037017 100644 --- a/src/main/java/org/springframework/data/gemfire/server/SubscriptionEvictionPolicy.java +++ b/src/main/java/org/springframework/data/gemfire/server/SubscriptionEvictionPolicy.java @@ -19,7 +19,7 @@ package org.springframework.data.gemfire.server; import com.gemstone.gemfire.cache.server.CacheServer; /** - * Enumeration of the various client subscription policies for {@link CacheServer}. + * Enumeration of the various client subscription policies for {@link CacheServerProcess}. * * @author Costin Leau */ diff --git a/src/test/java/org/springframework/data/gemfire/ForkUtil.java b/src/test/java/org/springframework/data/gemfire/ForkUtil.java new file mode 100644 index 00000000..710dc14b --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/ForkUtil.java @@ -0,0 +1,105 @@ +/* + * Copyright 2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.gemfire; + +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.util.concurrent.atomic.AtomicBoolean; + + +/** + * Utility for forking Java processes. + * + * @author Costin Leau + */ +public class ForkUtil { + + public static OutputStream cloneJVM(String argument) { + String cp = System.getProperty("java.class.path"); + String home = System.getProperty("java.home"); + + Process proc = null; + String java = home + "\\bin\\java "; + String argCp = "-cp " + cp; + String argClass = argument; + + String cmd = java + argCp + " " + argClass; + try { + //ProcessBuilder builder = new ProcessBuilder(cmd, argCp, argClass); + //builder.redirectErrorStream(true); + proc = Runtime.getRuntime().exec(cmd); + } catch (IOException ioe) { + throw new IllegalStateException("Cannot start command " + cmd, ioe); + } + + System.out.println("Started fork"); + final Process p = proc; + + final BufferedReader br = new BufferedReader(new InputStreamReader(p.getInputStream())); + final AtomicBoolean run = new AtomicBoolean(true); + + Thread reader = new Thread(new Runnable() { + + public void run() { + try { + String line = null; + do { + while ((line = br.readLine()) != null) { + System.out.println("[FORK] " + line); + } + Thread.sleep(200); + } while (run.get()); + } catch (Exception ex) { + // ignore and exit + } + } + }); + + reader.start(); + + Runtime.getRuntime().addShutdownHook(new Thread() { + @Override + public void run() { + System.out.println("Stopping fork..."); + run.set(false); + if (p != null) + p.destroy(); + + try { + p.waitFor(); + } catch (InterruptedException e) { + // ignore + } + System.out.println("Fork stopped"); + } + }); + + return proc.getOutputStream(); + } + + public static OutputStream cacheServer() { + OutputStream os = cloneJVM("org.springframework.data.gemfire.fork.CacheServerProcess"); + try { + Thread.sleep(3000); + } catch (InterruptedException ex) { + // ignore and move on + } + return os; + } +} \ No newline at end of file diff --git a/src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java b/src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java new file mode 100644 index 00000000..015b67bd --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/fork/CacheServerProcess.java @@ -0,0 +1,75 @@ +/* + * Copyright 2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.gemfire.fork; + +import java.io.BufferedReader; +import java.io.InputStreamReader; +import java.util.Properties; + +import com.gemstone.gemfire.cache.AttributesFactory; +import com.gemstone.gemfire.cache.Cache; +import com.gemstone.gemfire.cache.CacheFactory; +import com.gemstone.gemfire.cache.DataPolicy; +import com.gemstone.gemfire.cache.Region; +import com.gemstone.gemfire.cache.Scope; +import com.gemstone.gemfire.cache.server.CacheServer; +import com.gemstone.gemfire.distributed.DistributedSystem; + +/** + * @author Costin Leau + */ +public class CacheServerProcess { + + public static void main(String[] args) throws Exception { + + Properties props = new Properties(); + props.setProperty("name", "CqServer"); + props.setProperty("log-level", "warning"); + + System.out.println("\nConnecting to the distributed system and creating the cache."); + DistributedSystem ds = DistributedSystem.connect(props); + Cache cache = CacheFactory.create(ds); + + // Create region. + AttributesFactory factory = new AttributesFactory(); + factory.setDataPolicy(DataPolicy.REPLICATE); + factory.setScope(Scope.DISTRIBUTED_ACK); + Region testRegion = cache.createRegion("test", factory.create()); + System.out.println("Test region, " + testRegion.getFullPath() + ", created in cache."); + + // Start Cache Server. + CacheServer server = cache.addCacheServer(); + server.setPort(40404); + server.setNotifyBySubscription(true); + server.start(); + + + System.out.println("Waiting for signal"); + // wait for signal + BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(System.in)); + bufferedReader.readLine(); + + System.out.println("Received signal"); + + testRegion.put("one", 1); + testRegion.put("two", 2); + testRegion.put("three", 3); + + System.out.println("Waiting for shutdown"); + bufferedReader.readLine(); + } +} diff --git a/src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java b/src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java new file mode 100644 index 00000000..1419b6b4 --- /dev/null +++ b/src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java @@ -0,0 +1,127 @@ +/* + * Copyright 2011 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.data.gemfire.listener; + +import java.io.IOException; +import java.io.OutputStream; +import java.util.Properties; +import java.util.concurrent.BlockingDeque; +import java.util.concurrent.LinkedBlockingDeque; + +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.QueryListenerAdapter; + +import com.gemstone.gemfire.cache.RegionService; +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; + +/** + * @author Costin Leau + */ +public class ListenerContainerTests { + + private final BlockingDeque bag = new LinkedBlockingDeque(); + protected QueryListenerContainer container; + + private static RegionService cache = null; + private static Pool pool = null; + private static OutputStream os = null; + + private final Object handler = new Object() { + public void handleEvent(CqEvent event) { + bag.add(event); + } + }; + + private final QueryListenerAdapter adapter = new QueryListenerAdapter(handler); + + @BeforeClass + public static void startUp() throws Exception { + os = ForkUtil.cacheServer(); + + Properties props = new Properties(); + props.put("mcast-port", "0"); + 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(); + + cache = cacheFB.getObject(); + + + PoolFactory pf = PoolManager.createFactory(); + pf.addServer("localhost", 40404); + pf.setSubscriptionEnabled(true); + pool = pf.create("client"); + } + + + @AfterClass + public static void cleanUp() { + sendSignal(); + + if (pool != null) { + pool.destroy(); + pool = null; + } + + if (cache != null) { + cache.close(); + } + cache = null; + } + + + @Before + public void setUp() throws Exception { + String query = "SELECT * from /test"; + + container = new QueryListenerContainer(); + container.setQueryService(pool.getQueryService()); + container.setBeanName("container"); + container.addListener(new CqQueryDefinition("test", query, adapter)); + container.afterPropertiesSet(); + } + + private static void sendSignal() { + try { + os.write("\n".getBytes()); + os.flush(); + } catch (IOException ex) { + throw new IllegalStateException("Cannot communicate with forked VM", ex); + } + } + + @Test + public void testContainer() throws Exception { + sendSignal(); + Thread.sleep(3000); + System.out.println("Bag is " + bag); + sendSignal(); + } +} \ No newline at end of file diff --git a/src/test/resources/cache.xml b/src/test/resources/cache.xml deleted file mode 100644 index 5c221040..00000000 --- a/src/test/resources/cache.xml +++ /dev/null @@ -1,39 +0,0 @@ - - - - - java.lang.String - - - - - - - - - - - - - - - - - - Application Version - 1.0 - - - - - - - - - - - - - - -