diff --git a/src/main/java/org/springframework/data/gemfire/config/GemfireListenerContainerParser.java b/src/main/java/org/springframework/data/gemfire/config/GemfireListenerContainerParser.java new file mode 100644 index 00000000..89ac1ab4 --- /dev/null +++ b/src/main/java/org/springframework/data/gemfire/config/GemfireListenerContainerParser.java @@ -0,0 +1,126 @@ +/* + * 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.config; + +import java.util.List; + +import org.springframework.beans.factory.config.BeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.support.ManagedSet; +import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.data.gemfire.listener.ContinousQueryDefinition; +import org.springframework.data.gemfire.listener.ContinousQueryListenerContainer; +import org.springframework.data.gemfire.listener.adapter.ContinousQueryListenerAdapter; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; +import org.springframework.util.xml.DomUtils; +import org.w3c.dom.Attr; +import org.w3c.dom.Element; +import org.w3c.dom.NamedNodeMap; + +/** + * Parser for SGF <cq-listener-container> element. + * + * @author Costin Leau + */ +class GemfireListenerContainerParser extends AbstractSimpleBeanDefinitionParser { + + @Override + protected Class getBeanClass(Element element) { + return ContinousQueryListenerContainer.class; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + // parse attributes (but replace the value assignment with references) + NamedNodeMap attributes = element.getAttributes(); + + for (int x = 0; x < attributes.getLength(); x++) { + Attr attribute = (Attr) attributes.item(x); + if (isEligibleAttribute(attribute, parserContext)) { + String propertyName = extractPropertyName(attribute.getLocalName()); + Assert.state(StringUtils.hasText(propertyName), + "Illegal property name returned from 'extractPropertyName(String)': cannot be null or empty."); + builder.addPropertyReference(propertyName, attribute.getValue()); + } + } + + String phase = element.getAttribute("phase"); + if (StringUtils.hasText(phase)) { + builder.addPropertyValue("phase", phase); + } + + postProcess(builder, element); + + // parse nested listeners + List listDefs = DomUtils.getChildElementsByTagName(element, "listener"); + + if (!listDefs.isEmpty()) { + ManagedSet listeners = new ManagedSet(listDefs.size()); + for (Element listElement : listDefs) { + listeners.add(parseListener(listElement)); + } + + builder.addPropertyValue("queryListeners", listeners); + } + } + + @Override + protected boolean isEligibleAttribute(String attributeName) { + return (!"phase".equals(attributeName)); + } + + /** + * Parses a listener definition. Returns the listener bean reference definition (of a {@link ContinousQueryDefinition}). + * + * @param element + * @return + */ + private BeanDefinition parseListener(Element element) { + + BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ContinousQueryListenerAdapter.class); + builder.addConstructorArgReference(element.getAttribute("ref")); + + String attr = element.getAttribute("method"); + if (StringUtils.hasText(attr)) { + builder.addPropertyValue("defaultListenerMethod", attr); + } + + BeanDefinitionBuilder defBuilder = BeanDefinitionBuilder.genericBeanDefinition(ContinousQueryDefinition.class); + + attr = element.getAttribute("name"); + if (StringUtils.hasText(attr)) { + defBuilder.addConstructorArgValue(attr); + } + + defBuilder.addConstructorArgValue(element.getAttribute("query")); + defBuilder.addConstructorArgValue(builder.getBeanDefinition()); + + attr = element.getAttribute("durable"); + if (StringUtils.hasText(attr)) { + defBuilder.addConstructorArgValue(attr); + } + + return defBuilder.getBeanDefinition(); + } + + @Override + protected boolean shouldGenerateId() { + return true; + } +} \ No newline at end of file diff --git a/src/main/java/org/springframework/data/gemfire/config/GemfireNamespaceHandler.java b/src/main/java/org/springframework/data/gemfire/config/GemfireNamespaceHandler.java index 7882ab01..62a93ae1 100644 --- a/src/main/java/org/springframework/data/gemfire/config/GemfireNamespaceHandler.java +++ b/src/main/java/org/springframework/data/gemfire/config/GemfireNamespaceHandler.java @@ -37,5 +37,7 @@ class GemfireNamespaceHandler extends NamespaceHandlerSupport { registerBeanDefinitionParser("cache-server", new CacheServerParser()); registerBeanDefinitionParser("transaction-manager", new TransactionManagerParser()); + + registerBeanDefinitionParser("cq-listener-container", new GemfireListenerContainerParser()); } } \ No newline at end of file diff --git a/src/main/java/org/springframework/data/gemfire/listener/CqQueryDefinition.java b/src/main/java/org/springframework/data/gemfire/listener/ContinousQueryDefinition.java similarity index 70% rename from src/main/java/org/springframework/data/gemfire/listener/CqQueryDefinition.java rename to src/main/java/org/springframework/data/gemfire/listener/ContinousQueryDefinition.java index dcc6bddc..25c591fd 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/CqQueryDefinition.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinousQueryDefinition.java @@ -22,36 +22,36 @@ import org.springframework.util.Assert; import com.gemstone.gemfire.cache.query.CqQuery; /** - * Basic holder class for defining an CqQuery. Useful for configuring GemFire {@link CqQuery}s through XML + * Basic holder class for defining an {@link CqQuery}. Useful for configuring GemFire {@link CqQuery}s through XML * and or JavaBeans means. * * @author Costin Leau */ -public class CqQueryDefinition implements InitializingBean { +public class ContinousQueryDefinition implements InitializingBean { private String name = null, query = null; - private QueryListener listener = null; + private ContinuousQueryListener listener = null; private boolean durable = false; - public CqQueryDefinition() { + public ContinousQueryDefinition() { } - public CqQueryDefinition(String query, QueryListener listener) { + public ContinousQueryDefinition(String query, ContinuousQueryListener listener) { this(query, listener, false); } - public CqQueryDefinition(String query, QueryListener listener, boolean durable) { + public ContinousQueryDefinition(String query, ContinuousQueryListener listener, boolean durable) { this.query = query; this.listener = listener; this.durable = durable; afterPropertiesSet(); } - public CqQueryDefinition(String name, String query, QueryListener listener) { + public ContinousQueryDefinition(String name, String query, ContinuousQueryListener listener) { this(name, query, listener, false); } - public CqQueryDefinition(String name, String query, QueryListener listener, boolean durable) { + public ContinousQueryDefinition(String name, String query, ContinuousQueryListener listener, boolean durable) { this.name = name; this.query = query; this.listener = listener; @@ -81,7 +81,7 @@ public class CqQueryDefinition implements InitializingBean { /** * @return the listener */ - public QueryListener getListener() { + public ContinuousQueryListener getListener() { return listener; } diff --git a/src/main/java/org/springframework/data/gemfire/listener/QueryListenerContainer.java b/src/main/java/org/springframework/data/gemfire/listener/ContinousQueryListenerContainer.java similarity index 89% rename from src/main/java/org/springframework/data/gemfire/listener/QueryListenerContainer.java rename to src/main/java/org/springframework/data/gemfire/listener/ContinousQueryListenerContainer.java index b25b3290..5b6c017e 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/QueryListenerContainer.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinousQueryListenerContainer.java @@ -47,12 +47,12 @@ import com.gemstone.gemfire.internal.concurrent.ConcurrentHashSet; * * @author Costin Leau */ -public class QueryListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle { +public class ContinousQueryListenerContainer implements InitializingBean, DisposableBean, BeanNameAware, SmartLifecycle { private class EventDispatcherAdapter implements CqListener { - private final QueryListener delegate; + private final ContinuousQueryListener delegate; - EventDispatcherAdapter(QueryListener delegate) { + EventDispatcherAdapter(ContinuousQueryListener delegate) { this.delegate = delegate; } @@ -72,9 +72,9 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, protected final Log logger = LogFactory.getLog(getClass()); /** - * Default thread name prefix: "QueryListenerContainer-". + * Default thread name prefix: "ContinousQueryListenerContainer-". */ - public static final String DEFAULT_THREAD_NAME_PREFIX = ClassUtils.getShortName(QueryListenerContainer.class) + "-"; + public static final String DEFAULT_THREAD_NAME_PREFIX = ClassUtils.getShortName(ContinousQueryListenerContainer.class) + "-"; private Executor subscriptionExecutor; private Executor taskExecutor; @@ -160,7 +160,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, doStart(); if (logger.isDebugEnabled()) { - logger.debug("Started QueryListenerContainer"); + logger.debug("Started ContinousQueryListenerContainer"); } } } @@ -172,7 +172,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, } if (logger.isDebugEnabled()) { - logger.debug("Stopped QueryListenerContainer"); + logger.debug("Stopped ContinousQueryListenerContainer"); } } @@ -215,7 +215,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, * * @see #handleListenerException */ - protected void executeListener(QueryListener listener, CqEvent event) { + protected void executeListener(ContinuousQueryListener listener, CqEvent event) { try { listener.onEvent(event); } catch (Throwable ex) { @@ -298,6 +298,11 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, this.queryService = cache.getQueryService(); } + /** + * Set the query service to be used by this container. + * + * @param service query service used by the container + */ public void setQueryService(QueryService service) { this.queryService = service; } @@ -307,7 +312,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, * * @param queries set of queries */ - public void setQueryListeners(Set queries) { + public void setQueryListeners(Set queries) { initMapping(queries); } @@ -317,11 +322,11 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, * * @param listener event cqQuery */ - public void addListener(CqQueryDefinition cqQuery) { + public void addListener(ContinousQueryDefinition cqQuery) { doAddListener(cqQuery); } - private void initMapping(Set queryDefinitions) { + private void initMapping(Set queryDefinitions) { // stop the listener if currently running if (isRunning()) { stop(); @@ -329,7 +334,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, closeQueries(); - for (CqQueryDefinition def : queryDefinitions) { + for (ContinousQueryDefinition def : queryDefinitions) { doAddListener(def); } @@ -339,7 +344,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, } } - private void doAddListener(CqQueryDefinition def) { + private void doAddListener(ContinousQueryDefinition def) { CqQuery cq = null; try { @@ -375,7 +380,7 @@ public class QueryListenerContainer implements InitializingBean, DisposableBean, } } - private void dispatchEvent(final QueryListener listener, final CqEvent event) { + private void dispatchEvent(final ContinuousQueryListener listener, final CqEvent event) { taskExecutor.execute(new Runnable() { public void run() { executeListener(listener, event); diff --git a/src/main/java/org/springframework/data/gemfire/listener/QueryListener.java b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListener.java similarity index 95% rename from src/main/java/org/springframework/data/gemfire/listener/QueryListener.java rename to src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListener.java index fdd36d1a..c96581c5 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/QueryListener.java +++ b/src/main/java/org/springframework/data/gemfire/listener/ContinuousQueryListener.java @@ -23,7 +23,7 @@ import com.gemstone.gemfire.cache.query.CqEvent; * * @author Costin Leau */ -public interface QueryListener { +public interface ContinuousQueryListener { void onEvent(CqEvent event); } diff --git a/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java b/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java index 3cbab3f3..0295480d 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java +++ b/src/main/java/org/springframework/data/gemfire/listener/GemfireListenerExecutionFailedException.java @@ -17,13 +17,13 @@ package org.springframework.data.gemfire.listener; import org.springframework.dao.InvalidDataAccessApiUsageException; -import org.springframework.data.gemfire.listener.adapter.QueryListenerAdapter; +import org.springframework.data.gemfire.listener.adapter.ContinousQueryListenerAdapter; /** * Exception thrown when the execution of a listener method failed. * * @author Costin Leau - * @see QueryListenerAdapter + * @see ContinousQueryListenerAdapter */ public class GemfireListenerExecutionFailedException extends InvalidDataAccessApiUsageException { diff --git a/src/main/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapter.java b/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinousQueryListenerAdapter.java similarity index 93% rename from src/main/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapter.java rename to src/main/java/org/springframework/data/gemfire/listener/adapter/ContinousQueryListenerAdapter.java index 40ea9a02..044271a0 100644 --- a/src/main/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapter.java +++ b/src/main/java/org/springframework/data/gemfire/listener/adapter/ContinousQueryListenerAdapter.java @@ -27,7 +27,7 @@ import org.apache.commons.logging.LogFactory; import org.springframework.dao.DataAccessException; import org.springframework.dao.InvalidDataAccessApiUsageException; import org.springframework.data.gemfire.listener.GemfireListenerExecutionFailedException; -import org.springframework.data.gemfire.listener.QueryListener; +import org.springframework.data.gemfire.listener.ContinuousQueryListener; import org.springframework.util.Assert; import org.springframework.util.ReflectionUtils; import org.springframework.util.ReflectionUtils.MethodCallback; @@ -69,7 +69,7 @@ import com.gemstone.gemfire.cache.query.CqQuery; * @author Costin Leau * @see org.springframework.jms.listener.adapter.MessageListenerAdapter */ -public class QueryListenerAdapter implements QueryListener { +public class ContinousQueryListenerAdapter implements ContinuousQueryListener { private class MethodInvoker { private final Object delegate; @@ -192,18 +192,18 @@ public class QueryListenerAdapter implements QueryListener { private MethodInvoker invoker; /** - * Create a new {@link QueryListenerAdapter} with default settings. + * Create a new {@link ContinousQueryListenerAdapter} with default settings. */ - public QueryListenerAdapter() { + public ContinousQueryListenerAdapter() { setDelegate(this); } /** - * Create a new {@link QueryListenerAdapter} for the given delegate. + * Create a new {@link ContinousQueryListenerAdapter} for the given delegate. * * @param delegate the delegate object */ - public QueryListenerAdapter(Object delegate) { + public ContinousQueryListenerAdapter(Object delegate) { setDelegate(delegate); } @@ -251,7 +251,7 @@ public class QueryListenerAdapter implements QueryListener { } /** - * Standard {@link QueryListener} entry point. + * Standard {@link ContinuousQueryListener} entry point. *

Delegates the event to the target listener method, with appropriate * conversion of the event argument. In case of an exception, the * {@link #handleListenerException(Throwable)} method will be invoked. @@ -263,11 +263,11 @@ public class QueryListenerAdapter implements QueryListener { public void onEvent(CqEvent event) { try { - // Check whether the delegate is a QueryListener impl itself. + // Check whether the delegate is a ContinuousQueryListener impl itself. // In that case, the adapter will simply act as a pass-through. if (delegate != this) { - if (delegate instanceof QueryListener) { - ((QueryListener) delegate).onEvent(event); + if (delegate instanceof ContinuousQueryListener) { + ((ContinuousQueryListener) delegate).onEvent(event); } } diff --git a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.1.xsd b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.1.xsd index 47039d54..c9fc5cfb 100644 --- a/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.1.xsd +++ b/src/main/resources/org/springframework/data/gemfire/config/spring-gemfire-1.1.xsd @@ -1032,4 +1032,101 @@ The server groups that this server will be a member of given as a comma separate + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/src/test/java/org/springframework/data/gemfire/ForkUtil.java b/src/test/java/org/springframework/data/gemfire/ForkUtil.java index c50f940d..e240ef7b 100644 --- a/src/test/java/org/springframework/data/gemfire/ForkUtil.java +++ b/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); + } + } } \ No newline at end of file diff --git a/src/test/java/org/springframework/data/gemfire/listener/GemfireMDP.java b/src/test/java/org/springframework/data/gemfire/listener/GemfireMDP.java new file mode 100644 index 00000000..f64cec97 --- /dev/null +++ b/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); + } +} 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 c296f83a..d7f2d17e 100644 --- a/src/test/java/org/springframework/data/gemfire/listener/ListenerContainerTests.java +++ b/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 bag = new LinkedBlockingDeque(); - 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(); } } \ No newline at end of file diff --git a/src/test/java/org/springframework/data/gemfire/listener/StubErrorHandler.java b/src/test/java/org/springframework/data/gemfire/listener/StubErrorHandler.java new file mode 100644 index 00000000..6d2454cf --- /dev/null +++ b/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 throwables = new LinkedBlockingDeque(); + + + public void handleError(Throwable t) { + throwables.add(t); + } + +} diff --git a/src/test/java/org/springframework/data/gemfire/listener/ThrowableEventListener.java b/src/test/java/org/springframework/data/gemfire/listener/ThrowableEventListener.java new file mode 100644 index 00000000..19e8f405 --- /dev/null +++ b/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); + } +} 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 new file mode 100644 index 00000000..44942eac --- /dev/null +++ b/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(); + } +} diff --git a/src/test/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapterTest.java b/src/test/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapterTest.java index 43adbeb9..f68c23eb 100644 --- a/src/test/java/org/springframework/data/gemfire/listener/adapter/QueryListenerAdapterTest.java +++ b/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()); diff --git a/src/test/resources/org/springframework/data/gemfire/listener/container.xml b/src/test/resources/org/springframework/data/gemfire/listener/container.xml new file mode 100644 index 00000000..e54d7ddb --- /dev/null +++ b/src/test/resources/org/springframework/data/gemfire/listener/container.xml @@ -0,0 +1,39 @@ + + + + + + 0 + cq-client + warning + + + + + + + + + + + + + + + + + + + + + + +