Merge branch 'SGF_57_1'

src/test/java/org/springframework/data/gemfire/ForkUtil.java

@@ -29,6 +29,7 @@ import java.util.concurrent.atomic.AtomicBoolean;
  * @author Costin Leau
  */
 public class ForkUtil {
+	private static OutputStream os;
 
 	public static OutputStream cloneJVM(String argument) {
 		String cp = System.getProperty("java.class.path");
@@ -79,6 +80,7 @@ public class ForkUtil {
 			public void run() {
 				System.out.println("Stopping fork...");
 				run.set(false);
+				os = null;
 				if (p != null)
 					p.destroy();
 
@@ -91,7 +93,8 @@ public class ForkUtil {
 			}
 		});
 
-		return proc.getOutputStream();
+		os = proc.getOutputStream();
+		return os;
 	}
 
 	public static OutputStream cacheServer() {
@@ -103,4 +106,13 @@ public class ForkUtil {
 		}
 		return os;
 	}
+
+	public static void sendSignal() {
+		try {
+			os.write("\n".getBytes());
+			os.flush();
+		} catch (IOException ex) {
+			throw new IllegalStateException("Cannot communicate with forked VM", ex);
+		}
+	}
 }

src/test/java/org/springframework/data/gemfire/listener/GemfireMDP.java

@@ -0,0 +1,41 @@
+/*
+ * 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 com.gemstone.gemfire.cache.Operation;
+import com.gemstone.gemfire.cache.query.CqEvent;
+import com.gemstone.gemfire.cache.query.CqQuery;
+
+/**
+ * Simple GemFire-message/event-driven-pojo.
+ * 
+ * @author Costin Leau
+ */
+public class GemfireMDP {
+
+	public void handleEvent(CqEvent event) {
+		System.out.println("Received event " + event);
+	}
+
+	public void handleQuery(CqQuery query) {
+		System.out.println("Received query " + query);
+	}
+
+	public void handleOperation(Operation op) {
+		System.out.println("Received operation " + op);
+	}
+}

src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java

@@ -16,8 +16,6 @@
 
 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;
@@ -28,9 +26,10 @@ 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 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.Pool;
 import com.gemstone.gemfire.cache.client.PoolFactory;
 import com.gemstone.gemfire.cache.client.PoolManager;
@@ -42,11 +41,10 @@ import com.gemstone.gemfire.cache.query.CqEvent;
 public class ListenerContainerTests {
 
 	private final BlockingDeque<CqEvent> bag = new LinkedBlockingDeque<CqEvent>();
-	protected QueryListenerContainer container;
+	protected ContinousQueryListenerContainer 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) {
@@ -54,11 +52,11 @@ public class ListenerContainerTests {
 		}
 	};
 
-	private final QueryListenerAdapter adapter = new QueryListenerAdapter(handler);
+	private final ContinousQueryListenerAdapter adapter = new ContinousQueryListenerAdapter(handler);
 
 	@BeforeClass
 	public static void startUp() throws Exception {
-		os = ForkUtil.cacheServer();
+		ForkUtil.cacheServer();
 
 		Properties props = new Properties();
 		props.put("mcast-port", "0");
@@ -73,7 +71,6 @@ public class ListenerContainerTests {
 
 		cache = cacheFB.getObject();
 
-
 		PoolFactory pf = PoolManager.createFactory();
 		pf.addServer("localhost", 40404);
 		pf.setSubscriptionEnabled(true);
@@ -83,7 +80,7 @@ public class ListenerContainerTests {
 
 	@AfterClass
 	public static void cleanUp() {
-		sendSignal();
+		ForkUtil.sendSignal();
 
 		if (pool != null) {
 			pool.destroy();
@@ -101,27 +98,21 @@ public class ListenerContainerTests {
 	public void setUp() throws Exception {
 		String query = "SELECT * from /test-cq";
 
-		container = new QueryListenerContainer();
-		container.setQueryService(pool.getQueryService());
+		container = new ContinousQueryListenerContainer();
+		//container.setQueryService(pool.getQueryService());
+		System.out.println(cache instanceof ClientCache);
+		container.setCache(cache);
 		container.setBeanName("container");
-		container.addListener(new CqQueryDefinition("test", query, adapter));
+		container.addListener(new ContinousQueryDefinition("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();
+		ForkUtil.sendSignal();
 		Thread.sleep(3000);
 		System.out.println("Bag is " + bag);
-		sendSignal();
+		ForkUtil.sendSignal();
 	}
 }

src/test/java/org/springframework/data/gemfire/listener/StubErrorHandler.java

@@ -0,0 +1,35 @@
+/*
+ * 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.util.concurrent.BlockingDeque;
+import java.util.concurrent.LinkedBlockingDeque;
+
+import org.springframework.util.ErrorHandler;
+
+/**
+ * @author Costin Leau
+ */
+public class StubErrorHandler implements ErrorHandler {
+
+	public BlockingDeque<Throwable> throwables = new LinkedBlockingDeque<Throwable>();
+
+
+	public void handleError(Throwable t) {
+		throwables.add(t);
+	}
+
+}

src/test/java/org/springframework/data/gemfire/listener/ThrowableEventListener.java

@@ -0,0 +1,30 @@
+/*
+ * 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 com.gemstone.gemfire.cache.query.CqEvent;
+
+
+/**
+ * 
+ * @author Costin Leau
+ */
+public class ThrowableEventListener implements ContinuousQueryListener {
+
+	public void onEvent(CqEvent event) {
+		throw new IllegalStateException("throwing exception for event " + event);
+	}
+}

src/test/java/org/springframework/data/gemfire/listener/adapter/ContainerXmlSetupTest.java

@@ -0,0 +1,71 @@
+/*
+ * 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.adapter;
+
+import static org.junit.Assert.assertTrue;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.springframework.context.support.GenericXmlApplicationContext;
+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;
+
+/**
+ * @author Costin Leau
+ */
+public class ContainerXmlSetupTest {
+
+	@BeforeClass
+	public static void init() {
+		ForkUtil.cacheServer();
+	}
+
+	@AfterClass
+	public static void cleanUp() {
+		ForkUtil.sendSignal();
+	}
+
+
+	@Test
+	public void testContainerSetup() throws Exception {
+		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());
+
+		ForkUtil.sendSignal();
+		Thread.sleep(3000);
+		Cache cache = ctx.getBean("gemfire-cache", Cache.class);
+
+		CqQuery[] cqs = cache.getQueryService().getCqs();
+		System.out.println("Cqs " + cqs.length);
+		ForkUtil.sendSignal();
+	}
+}

src/test/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapterTest.java

@@ -24,7 +24,7 @@ import static org.mockito.Mockito.verify;
 
 import org.junit.Before;
 import org.junit.Test;
-import org.springframework.data.gemfire.listener.QueryListener;
+import org.springframework.data.gemfire.listener.ContinuousQueryListener;
 
 import com.gemstone.gemfire.cache.Operation;
 import com.gemstone.gemfire.cache.query.CqEvent;
@@ -37,11 +37,11 @@ import com.gemstone.gemfire.cache.query.internal.CqQueryImpl;
  */
 public class QueryListenerAdapterTest {
 
-	private QueryListenerAdapter adapter;
+	private ContinousQueryListenerAdapter adapter;
 
 	@Before
 	public void setUp() {
-		adapter = new QueryListenerAdapter();
+		adapter = new ContinousQueryListenerAdapter();
 	}
 
 	CqEvent event() {
@@ -113,14 +113,14 @@ public class QueryListenerAdapterTest {
 
 	@Test
 	public void testThatTheDefaultHandlingMethodNameIsTheConstantDefault() throws Exception {
-		assertEquals(QueryListenerAdapter.ORIGINAL_DEFAULT_LISTENER_METHOD, adapter.getDefaultListenerMethod());
+		assertEquals(ContinousQueryListenerAdapter.ORIGINAL_DEFAULT_LISTENER_METHOD, adapter.getDefaultListenerMethod());
 	}
 
 	@Test
 	public void testAdapterWithListenerAndDefaultMessage() throws Exception {
-		QueryListener mock = mock(QueryListener.class);
+		ContinuousQueryListener mock = mock(ContinuousQueryListener.class);
 
-		QueryListenerAdapter adapter = new QueryListenerAdapter(mock);
+		ContinousQueryListenerAdapter adapter = new ContinousQueryListenerAdapter(mock);
 		CqEvent event = event();
 		adapter.onEvent(event);
 		verify(mock).onEvent(event);
@@ -129,7 +129,7 @@ public class QueryListenerAdapterTest {
 	@Test
 	public void testHandleEvent() throws Exception {
 		Delegate mock = mock(Delegate.class);
-		QueryListenerAdapter adapter = new QueryListenerAdapter(mock);
+		ContinousQueryListenerAdapter adapter = new ContinousQueryListenerAdapter(mock);
 		CqEvent event = event();
 
 		adapter.onEvent(event);
@@ -139,7 +139,7 @@ public class QueryListenerAdapterTest {
 	@Test
 	public void testHandleArray() throws Exception {
 		Delegate mock = mock(Delegate.class);
-		QueryListenerAdapter adapter = new QueryListenerAdapter(mock);
+		ContinousQueryListenerAdapter adapter = new ContinousQueryListenerAdapter(mock);
 		adapter.setDefaultListenerMethod("handleArray");
 		CqEvent event = event();
 		adapter.onEvent(event);
@@ -149,7 +149,7 @@ public class QueryListenerAdapterTest {
 	@Test
 	public void testHandleKey() throws Exception {
 		Delegate mock = mock(Delegate.class);
-		QueryListenerAdapter adapter = new QueryListenerAdapter(mock);
+		ContinousQueryListenerAdapter adapter = new ContinousQueryListenerAdapter(mock);
 		adapter.setDefaultListenerMethod("handleKey");
 		CqEvent event = event();
 
@@ -160,7 +160,7 @@ public class QueryListenerAdapterTest {
 	@Test
 	public void testHandleKV() throws Exception {
 		Delegate mock = mock(Delegate.class);
-		QueryListenerAdapter adapter = new QueryListenerAdapter(mock);
+		ContinousQueryListenerAdapter adapter = new ContinousQueryListenerAdapter(mock);
 		adapter.setDefaultListenerMethod("handleKV");
 		CqEvent event = event();
 
@@ -171,7 +171,7 @@ public class QueryListenerAdapterTest {
 	@Test
 	public void testHandleEx() throws Exception {
 		Delegate mock = mock(Delegate.class);
-		QueryListenerAdapter adapter = new QueryListenerAdapter(mock);
+		ContinousQueryListenerAdapter adapter = new ContinousQueryListenerAdapter(mock);
 		adapter.setDefaultListenerMethod("handleEx");
 		CqEvent event = event();
 
@@ -182,7 +182,7 @@ public class QueryListenerAdapterTest {
 	@Test
 	public void testHandleOps() throws Exception {
 		Delegate mock = mock(Delegate.class);
-		QueryListenerAdapter adapter = new QueryListenerAdapter(mock);
+		ContinousQueryListenerAdapter adapter = new ContinousQueryListenerAdapter(mock);
 		adapter.setDefaultListenerMethod("handleOps");
 		CqEvent event = event();
 
@@ -193,7 +193,7 @@ public class QueryListenerAdapterTest {
 	@Test
 	public void testHandleAll() throws Exception {
 		Delegate mock = mock(Delegate.class);
-		QueryListenerAdapter adapter = new QueryListenerAdapter(mock);
+		ContinousQueryListenerAdapter adapter = new ContinousQueryListenerAdapter(mock);
 		adapter.setDefaultListenerMethod("handleAll");
 		CqEvent event = event();
 
@@ -205,7 +205,7 @@ public class QueryListenerAdapterTest {
 	@Test
 	public void testInvalid() throws Exception {
 		Delegate mock = mock(Delegate.class);
-		QueryListenerAdapter adapter = new QueryListenerAdapter(mock);
+		ContinousQueryListenerAdapter adapter = new ContinousQueryListenerAdapter(mock);
 		adapter.setDefaultListenerMethod("handleInvalid");
 
 		adapter.onEvent(event());