SGF-57

+ integration test forking a VM for a cache server to test the CQ functionality

.classpath

@@ -1,10 +1,30 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <classpath>
-	<classpathentry kind="src" output="target/classes" path="src/main/java"/>
-	<classpathentry kind="src" output="target/classes" path="src/main/resources"/>
-	<classpathentry kind="src" output="target/test-classes" path="src/test/java"/>
-	<classpathentry kind="src" output="target/test-classes" path="src/test/resources"/>
-	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER/org.eclipse.jdt.internal.debug.ui.launcher.StandardVMType/J2SE-1.5"/>
-	<classpathentry kind="con" path="org.maven.ide.eclipse.MAVEN2_CLASSPATH_CONTAINER"/>
-	<classpathentry kind="output" path="target/classes"/>
+	<classpathentry kind="output" path="bin"/>
+	<classpathentry kind="src" path="src/main/java"/>
+	<classpathentry kind="src" path="src/main/resources"/>
+	<classpathentry kind="src" path="src/test/java"/>
+	<classpathentry kind="src" path="src/test/resources"/>
+	<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.mockito/mockito-core/sources/mockito-core-1.8.5-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.mockito/mockito-core/jars/mockito-core-1.8.5.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.springframework/spring-core/sources/spring-core-3.1.0.M2-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.springframework/spring-core/jars/spring-core-3.1.0.M2.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.springframework/spring-aop/sources/spring-aop-3.1.0.M2-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.springframework/spring-aop/jars/spring-aop-3.1.0.M2.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.slf4j/slf4j-log4j12/sources/slf4j-log4j12-1.6.1-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.slf4j/slf4j-log4j12/jars/slf4j-log4j12-1.6.1.jar" exported="true"/>
+	<classpathentry kind="lib" path="C:/Users/costin/.gradle/cache/antlr/antlr/jars/antlr-2.7.7.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/log4j/log4j/sources/log4j-1.2.16-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/log4j/log4j/bundles/log4j-1.2.16.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.springframework/spring-beans/sources/spring-beans-3.1.0.M2-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.springframework/spring-beans/jars/spring-beans-3.1.0.M2.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.springframework/spring-asm/sources/spring-asm-3.1.0.M2-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.springframework/spring-asm/jars/spring-asm-3.1.0.M2.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.springframework/spring-context-support/sources/spring-context-support-3.1.0.M2-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.springframework/spring-context-support/jars/spring-context-support-3.1.0.M2.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.hamcrest/hamcrest-core/sources/hamcrest-core-1.1-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.hamcrest/hamcrest-core/jars/hamcrest-core-1.1.jar" exported="true"/>
+	<classpathentry kind="lib" path="C:/Users/costin/.gradle/cache/com.gemstone.gemfire/gemfire/jars/gemfire-6.6.RC.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.springframework/spring-expression/sources/spring-expression-3.1.0.M2-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.springframework/spring-expression/jars/spring-expression-3.1.0.M2.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.objenesis/objenesis/sources/objenesis-1.0-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.objenesis/objenesis/jars/objenesis-1.0.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.springframework/spring-test/sources/spring-test-3.1.0.M2-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.springframework/spring-test/jars/spring-test-3.1.0.M2.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/javax.annotation/jsr250-api/sources/jsr250-api-1.0-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/javax.annotation/jsr250-api/jars/jsr250-api-1.0.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/aopalliance/aopalliance/sources/aopalliance-1.0-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/aopalliance/aopalliance/jars/aopalliance-1.0.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.slf4j/slf4j-api/sources/slf4j-api-1.6.1-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.slf4j/slf4j-api/jars/slf4j-api-1.6.1.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/junit/junit/sources/junit-4.8.1-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/junit/junit/jars/junit-4.8.1.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.slf4j/jcl-over-slf4j/sources/jcl-over-slf4j-1.6.1-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.slf4j/jcl-over-slf4j/jars/jcl-over-slf4j-1.6.1.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.springframework/spring-tx/sources/spring-tx-3.1.0.M2-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.springframework/spring-tx/jars/spring-tx-3.1.0.M2.jar" exported="true"/>
+	<classpathentry sourcepath="C:/Users/costin/.gradle/cache/org.springframework/spring-context/sources/spring-context-3.1.0.M2-sources.jar" kind="lib" path="C:/Users/costin/.gradle/cache/org.springframework/spring-context/jars/spring-context-3.1.0.M2.jar" exported="true"/>
 </classpath>

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<CqQuery> queries;
+	private Set<CqQuery> queries = new ConcurrentHashSet<CqQuery>();
+
+	private QueryService queryService;
 
 
 	public void afterPropertiesSet() {
@@ -101,7 +102,6 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean,
 			subscriptionExecutor = taskExecutor;
 		}
 
-		queries = new LinkedHashSet<CqQuery>();
 		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) {

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?");
 		}
 

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
  */

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
  */

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;
+	}
+}

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();
+	}
+}

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<CqEvent> bag = new LinkedBlockingDeque<CqEvent>();
+	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();
+	}
+}

src/test/resources/cache.xml

@@ -1,39 +0,0 @@
-<?xml version="1.0"?>
-<!DOCTYPE cache PUBLIC "-//GemStone Systems, Inc.//GemFire Declarative Caching 5.7//EN" "http://www.gemstone.com/dtd/cache5_7.dtd">
-<cache lock-lease="120" lock-timeout="60" search-timeout="300">
-  <region-attributes id="attTemplate" scope="local" data-policy="normal" initial-capacity="16" load-factor="0.75" concurrency-level="16" statistics-enabled="true">
-    <key-constraint>java.lang.String</key-constraint>
-  </region-attributes>
-  <region name="root">
-    <region-attributes refid="attTemplate" scope="distributed-no-ack">
-      <region-time-to-live>
-        <expiration-attributes timeout="0" action="invalidate"/>
-      </region-time-to-live>
-      <region-idle-time>
-        <expiration-attributes timeout="0" action="invalidate"/>
-      </region-idle-time>
-      <entry-time-to-live>
-        <expiration-attributes timeout="0" action="invalidate"/>
-      </entry-time-to-live>
-      <entry-idle-time>
-        <expiration-attributes timeout="0" action="invalidate"/>
-      </entry-idle-time>
-    </region-attributes>
-    <entry>
-      <key><string>Application Version</string></key>
-      <value><string>1.0</string></value>
-    </entry>
-    <region name="rlocal">
-      <region-attributes refid="attTemplate">
-      </region-attributes>
-    </region>
-    <region name="rdistnoack">
-      <region-attributes refid="attTemplate" scope="distributed-no-ack">
-      </region-attributes>
-    </region>
-    <region name="rglobalreplication">
-      <region-attributes refid="attTemplate" scope="global" data-policy="replicate">
-      </region-attributes>
-    </region>
-  </region>
-</cache>