diff --git a/spring-integration-gemfire/.springBeans b/spring-integration-gemfire/.springBeans index 8a3d5fe7e4..0f1d19bcc5 100644 --- a/spring-integration-gemfire/.springBeans +++ b/spring-integration-gemfire/.springBeans @@ -1,7 +1,7 @@ 1 - + @@ -9,6 +9,7 @@ src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml + src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests-context.xml diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java index 919016af3c..3572d273b7 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java @@ -16,182 +16,87 @@ package org.springframework.integration.gemfire.inbound; -import com.gemstone.gemfire.cache.Region; -import com.gemstone.gemfire.cache.client.Pool; -import com.gemstone.gemfire.cache.query.*; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; +import org.springframework.data.gemfire.listener.CqQueryDefinition; +import org.springframework.data.gemfire.listener.QueryListener; +import org.springframework.data.gemfire.listener.QueryListenerContainer; import org.springframework.integration.Message; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import com.gemstone.gemfire.cache.query.CqEvent; /** - * Responds to a continuous query (set using the #queryString field) that is - * constantly evaluated against a cache {@link com.gemstone.gemfire.cache.Region}. - * This is much faster than re-querying the cache manually. - * + * Responds to a Gemfire continuous query (set using the #query field) that is + * constantly evaluated against a cache + * {@link com.gemstone.gemfire.cache.Region}. This is much faster than + * re-querying the cache manually. + * * @author Josh Long + * @author David Turanski * @since 2.1 + * */ -public class ContinuousQueryMessageProducer extends MessageProducerSupport { - - private final Log logger = LogFactory.getLog(this.getClass()); - +public class ContinuousQueryMessageProducer extends MessageProducerSupport implements QueryListener { + private static Log logger = LogFactory.getLog(ContinuousQueryMessageProducer.class); + + private final String query; + private final QueryListenerContainer queryListenerContainer; + private volatile String name; + private boolean durable; + /** - * Not sure yet if there's a way to avoid depending on this. + * + * @param queryListenerContainer a {@link org.springframework.data.gemfire.listener.QueryListenerContainer} + * @param query the query string */ - private volatile Pool pool; - - /** - * Must be provided by the client of this class - */ - private final Region region; - - /** - * Is the queryString durable (optional) - */ - private volatile boolean durable = false; - - /** - * the {@link com.gemstone.gemfire.cache.query.CqQuery} instance created and - * registered with the server - */ - private volatile CqQuery cqQuery; - - /** - * the query to be registered against the cache - */ - private final String queryString; - - /** - * a reference to a {@link com.gemstone.gemfire.cache.query.QueryService} - * that is obtained through the #regionService instance. - */ - private volatile QueryService queryService; - - /** - * used when building the queryString itself - optional - */ - private volatile String queryName; - - /** - * a {@link com.gemstone.gemfire.cache.query.CqAttributesFactory} to generate - * the {@link com.gemstone.gemfire.cache.query.CqAttributes} that in turn - * hold the reference to the listener that we register to in turn funnel - * messages to the clients of this adapter. - */ - private final CqAttributesFactory cqAttributesFactory = new CqAttributesFactory(); - - /** - * the adapter requires a query string to continuously evaluate as well as a - * {@link com.gemstone.gemfire.cache.Region} against which to evaluate the - * query. - * - * @param region - * the region against which the query should be evaluated - * @param queryString - * the query string - */ - public ContinuousQueryMessageProducer(Region region, Pool pool, String queryString) { - this.region = region; - Assert.notNull(this.region, "You must provide a reference to a 'Region'"); - this.pool = pool; - Assert.notNull(this.pool, "You must provide a 'pool'"); - this.queryString = queryString; - Assert.hasText(this.queryString, "You must provide a queryString to evaluate against the region"); + public ContinuousQueryMessageProducer(QueryListenerContainer queryListenerContainer, String query) { + Assert.notNull(queryListenerContainer, "'queryListenerContainer' cannot be null"); + Assert.notNull(query, "'query' cannot be null"); + this.queryListenerContainer = queryListenerContainer; + this.query = query; } - + /** - * whether or not the query is durable (that is, whether or not this query - * should live beyond the registered query) - * - * @param durable - * whether or not the query is registered and saved and - * subsequently retrievable by a query name. + * + * @param name optional query name + */ + public void setName(String name) { + this.name = name; + } + + /** + * + * @param durable true if the query is a durable subscription */ public void setDurable(boolean durable) { this.durable = durable; } - /** - * Specify the name of the queryString (optional). - */ - public void setQueryName(String queryName) { - this.queryName = queryName; - } - - @Override - protected void doStart() { - try { - cqQuery.execute(); - } - catch (Throwable e) { - throw new RuntimeException("Failed to start the continuous query", e); - } - } - - @Override - protected void doStop() { - try { - this.cqQuery.stop(); - } - catch (CqException e) { - throw new RuntimeException("Failed to stop the continuous query", e); - } - } - - /** - * hook to handle registration of the query - */ - private CqQuery registerContinuousQuery(QueryService queryService, - String name, String query, boolean durable, CqListener cqListener) throws Throwable { - cqAttributesFactory.addCqListener(cqListener); - CqAttributes attrs = cqAttributesFactory.create(); - CqQuery cqQuery = queryService.newCq(name, query, attrs, durable); - return cqQuery; - } - @Override protected void onInit() { - try { - super.onInit(); - - // regionService = this.region.getRegionService(); - queryService = this.pool.getQueryService(); - String defaultName = String.format("%s-%s-query", - getComponentName() + "", getComponentType() + ""); - queryName = StringUtils.hasText(queryName) ? queryName : defaultName; - this.cqQuery = registerContinuousQuery(queryService, queryName, - this.queryString, this.durable, - new MessageProducingCqListener()); - } - catch (Throwable e) { - throw new RuntimeException("Couldn't properly setup the " + getClass().getName(), e); + super.onInit(); + if (name == null){ + queryListenerContainer.addListener(new CqQueryDefinition(this.query, this, this.durable)); + } else { + queryListenerContainer.addListener(new CqQueryDefinition(this.name, this.query, this, this.durable)); } } - - - /** - * Listener that listens for any events being broadcast as a result of the - * evaluation of a continuous query {@link CqQuery}. + + /* + * (non-Javadoc) + * + * @see + * org.springframework.data.gemfire.listener.QueryListener#onEvent(com.gemstone + * .gemfire.cache.query.CqEvent) */ - class MessageProducingCqListener implements CqListener { - - public void onEvent(CqEvent cqEvent) { - Message cqEventMessage = MessageBuilder.withPayload(cqEvent).build(); - sendMessage(cqEventMessage); - } - - public void onError(CqEvent cqEvent) { - logger.debug("error on " + getClass() + " (a CqListener) "); - throw new RuntimeException("error when interacting with region.", cqEvent.getThrowable()); - } - - public void close() { - logger.debug(getClass() + " close() called"); + public void onEvent(CqEvent event) { + if (logger.isDebugEnabled()){ + logger.debug(String.format("processing cq event key [%s] event [%s]",event.getBaseOperation().toString(),event.getKey())); } + Message cqEventMessage = MessageBuilder.withPayload(event).build(); + sendMessage(cqEventMessage); } } \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/CacheServerProcess.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/CacheServerProcess.java index 7a2a87e5c9..745a61eaa4 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/CacheServerProcess.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/CacheServerProcess.java @@ -31,14 +31,18 @@ import com.gemstone.gemfire.distributed.DistributedSystem; /** * @author Costin Leau + * @author David Turanski + * + * Runs as a standalone Java app. + * Modified from SGF implementation for testing client/server CQ features */ public class CacheServerProcess { public static void main(String[] args) throws Exception { Properties props = new Properties(); - props.setProperty("name", "CqServer"); - props.setProperty("log-level", "warning"); + props.setProperty("name", "CacheServer"); + props.setProperty("log-level", "info"); System.out.println("\nConnecting to the distributed system and creating the cache."); DistributedSystem ds = DistributedSystem.connect(props); @@ -48,28 +52,20 @@ public class CacheServerProcess { AttributesFactory factory = new AttributesFactory(); factory.setDataPolicy(DataPolicy.REPLICATE); factory.setScope(Scope.DISTRIBUTED_ACK); - Region testRegion = cache.createRegion("test-cq", factory.create()); + 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); + System.out.println("Starting server"); server.start(); - - System.out.println("Waiting for signal"); - // wait for signal + System.out.println("Waiting for shutdown"); + 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/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/ForkUtil.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/ForkUtil.java new file mode 100644 index 0000000000..57bee0bf72 --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/fork/ForkUtil.java @@ -0,0 +1,108 @@ +/* + * 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.integration.gemfire.fork; + +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. Modified from the SGF version for SI + * + * @author Costin Leau + * @author David Turanski + * + * + */ +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 ".replace("\\","/");; + 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.integration.gemfire.fork.CacheServerProcess"); + try { + Thread.sleep(8000); + } catch (InterruptedException ex) { + // ignore and move on + } + return os; + } +} \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests-context.xml new file mode 100644 index 0000000000..f91e4a42f0 --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests-context.xml @@ -0,0 +1,36 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests.java new file mode 100644 index 0000000000..7e6640fa1e --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests.java @@ -0,0 +1,94 @@ +/* + * Copyright 2002-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.integration.gemfire.inbound.cq; + +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.io.OutputStream; + +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.integration.Message; +import org.springframework.integration.core.PollableChannel; +import org.springframework.integration.gemfire.fork.ForkUtil; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; + +import com.gemstone.gemfire.cache.query.CqEvent; +import com.gemstone.gemfire.internal.cache.LocalRegion; + +/** + * @author David Turanski + * @since 2.1 + */ +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration +public class ContinuousQueryMessageProducerTests { + + @Autowired + LocalRegion region; + + @Autowired + ConfigurableApplicationContext applicationContext; + + @Autowired + PollableChannel outputChannel; + + static OutputStream os; + @BeforeClass + public static void startUp() throws Exception { + os = ForkUtil.cacheServer(); + } + + + + @Test + public void test() throws InterruptedException { + region.put("one",1); + Message msg = outputChannel.receive(1000); + assertNotNull(msg); + assertTrue(msg.getPayload() instanceof CqEvent); + /* + * Avoid shutdown errors + */ + applicationContext.close(); + } + + @AfterClass + public static void cleanUp() { + + try { + Thread.sleep(3000); + } + catch (InterruptedException e) { + // TODO Auto-generated catch block + e.printStackTrace(); + } + sendSignal(); + } + + public static void sendSignal() { + try { + os.write("\n".getBytes()); + os.flush(); + } catch (IOException ex) { + throw new IllegalStateException("Cannot communicate with forked VM", ex); + } + } +} diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml deleted file mode 100644 index a30f3ac5f9..0000000000 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml +++ /dev/null @@ -1,15 +0,0 @@ - - - - - - - - \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml deleted file mode 100644 index ee0387c201..0000000000 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml +++ /dev/null @@ -1,25 +0,0 @@ - - - - - - - - - - - - - - - - - - \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/client/CqClientConfiguration.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/client/CqClientConfiguration.java deleted file mode 100644 index 919f9dff26..0000000000 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/client/CqClientConfiguration.java +++ /dev/null @@ -1,131 +0,0 @@ -/* - * Copyright 2002-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.integration.gemfire.inbound.cq.client; - -import com.gemstone.gemfire.cache.Region; -import com.gemstone.gemfire.cache.client.*; -import com.gemstone.gemfire.cache.query.CqEvent; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.context.annotation.*; -import org.springframework.core.env.Environment; -import org.springframework.integration.Message; -import org.springframework.integration.MessageChannel; -import org.springframework.integration.MessagingException; -import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.integration.gemfire.inbound.ContinuousQueryMessageProducer; -import org.springframework.integration.gemfire.inbound.cq.server.CqServerConfiguration; - -import javax.annotation.PostConstruct; - -/** - * Simple example demonstrating the client side of a continuous query using Gemfire - * - * @author Josh Long - * - */ -@ImportResource("/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml") -@Configuration -@PropertySource("/org/springframework/integration/gemfire/inbound/cq/common.properties") -@SuppressWarnings("unused") -public class CqClientConfiguration { - - static private Log log = LogFactory.getLog(CqClientConfiguration.class); - - private String regionName, host, query; - - private int port; - - @Autowired - private Environment environment; - - @Autowired @Qualifier("cqIn") - private MessageChannel messageChannel; - - public static void main(String[] args) throws Throwable { - - if (log.isInfoEnabled()) { - log.info(String.format("Starting the %s client. Make sure to run the %s server, first.", CqServerConfiguration.class.getName(), CqClientConfiguration.class.getName())); - } - - AnnotationConfigApplicationContext annotationConfigApplicationContext = new AnnotationConfigApplicationContext(CqClientConfiguration.class); - - long timeout = 10 * 1000; - long counter = 0; - - while (((counter += 1) < timeout)) { - Thread.sleep(1000); - } - } - - @PostConstruct - public void setup() throws Throwable { - this.regionName = environment.getProperty("region-name"); - this.port = Integer.parseInt(environment.getProperty("port")); - this.host = environment.getProperty("host"); - this.query = environment.getProperty("region-query"); - } - - - @Bean - public Object cqServiceActivator() { - return new Object() { - @ServiceActivator - public void handleMessage(Message eventMessage) throws MessagingException { - CqEvent cqEvent = eventMessage.getPayload(); - System.out.println("Received an event from the continuous query adapter: " + cqEvent) ; - } - }; - } - - @Bean - public ClientCache clientCache() throws Throwable { - return new ClientCacheFactory().create(); - } - - @Bean - public Region clientRegion() throws Throwable { - ClientRegionFactory clientRegionFactory = clientCache().createClientRegionFactory(ClientRegionShortcut.PROXY); - return clientRegionFactory.create(this.regionName); - } - - @Bean - public Pool pool() throws Throwable { - return this.buildPool(this.host, this.port); - } - - @Bean - public ContinuousQueryMessageProducer continuousQueryMessageProducer() throws Throwable { - ContinuousQueryMessageProducer mp = new ContinuousQueryMessageProducer(this.clientRegion(), this.pool(), this.query); - mp.setDurable(true); - mp.setOutputChannel(this.messageChannel); - mp.setQueryName("pplQuery"); - return mp; - } - - - protected Pool buildPool(String host, int port) throws Throwable { - return PoolManager.createFactory() - .addServer(host, port) - .setSubscriptionEnabled(true) - .create(host + "Pool"); - } - - -} \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/server/CqServerConfiguration.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/server/CqServerConfiguration.java deleted file mode 100644 index 9c7ae872dd..0000000000 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/server/CqServerConfiguration.java +++ /dev/null @@ -1,116 +0,0 @@ -/* - * Copyright 2002-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.integration.gemfire.inbound.cq.server; - -import com.gemstone.gemfire.cache.Cache; -import com.gemstone.gemfire.cache.Region; -import com.gemstone.gemfire.cache.server.CacheServer; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.annotation.*; -import org.springframework.core.env.Environment; -import org.springframework.data.gemfire.GemfireTemplate; -import org.springframework.scheduling.TaskScheduler; -import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler; - -import javax.annotation.PostConstruct; - -/** - * Demonstrates the server side for a continuous query example. This must be run before - * {@link org.springframework.integration.gemfire.inbound.cq.client.CqClientConfiguration}. - * - * This must also be run in a separate VM as the {@link org.springframework.integration.gemfire.inbound.cq.client.CqClientConfiguration}. - * - * @author Josh Long - * - */ -@PropertySource("org/springframework/integration/gemfire/inbound/cq/common.properties") -@ImportResource("/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml") -@Configuration -@SuppressWarnings("unused") -public class CqServerConfiguration { - - - private static Log log = LogFactory.getLog(CqServerConfiguration.class); - - @Autowired private Environment environment; - - @Value("#{c}") private Cache cache; - - @Value("#{r}") private Region region; - - private String regionName, host; - private int port; - - public static void main(String[] args) throws Exception { - AnnotationConfigApplicationContext annotationConfigApplicationContext = new AnnotationConfigApplicationContext(CqServerConfiguration.class); - GemfireTemplate gemfireTemplate = annotationConfigApplicationContext.getBean(GemfireTemplate.class); - TaskScheduler scheduler = annotationConfigApplicationContext.getBean(TaskScheduler.class); - BusyWorkRunnable busyWorkRunnable = new BusyWorkRunnable(gemfireTemplate); - scheduler.scheduleAtFixedRate(busyWorkRunnable, 10 * 1000); - } - - @PostConstruct - public void setup() throws Throwable { - host = this.environment.getProperty("host"); - regionName = this.environment.getProperty("region-name"); - port = Integer.parseInt(this.environment.getProperty("port")); - } - - @Bean - public GemfireTemplate gemfireTemplate() { - return new GemfireTemplate(this.region); - } - - @Bean - public CacheServer cacheServer() throws Throwable { - CacheServer cacheServer = this.cache.addCacheServer(); - cacheServer.setBindAddress(this.host); - cacheServer.setPort(this.port); - cacheServer.start(); - return cacheServer; - } - - @Bean - public TaskScheduler scheduler() { - return new ThreadPoolTaskScheduler(); - } - - private static class BusyWorkRunnable implements Runnable { - - private GemfireTemplate gemfireTemplate; - - private String letters = "abcdefghijk"; - - public BusyWorkRunnable(GemfireTemplate gemfireTemplate) { - this.gemfireTemplate = gemfireTemplate; - } - - public void run() { - for (char c : letters.toCharArray()) { - if (log.isDebugEnabled()) { - log.debug("Adding '" + c + "'"); - } - gemfireTemplate.put("" + c, "value-" + c); - } - } - } - - -} \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest-context.xml index d2601f2e5d..195325bcba 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest-context.xml @@ -11,7 +11,7 @@ - + @@ -21,6 +21,6 @@ - + \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/common.properties b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/common.properties similarity index 100% rename from spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/common.properties rename to spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/common.properties diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/gfe-cache.properties similarity index 100% rename from spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties rename to spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/gfe-cache.properties diff --git a/spring-integration-gemfire/src/test/resources/log4j.xml b/spring-integration-gemfire/src/test/resources/log4j.xml index d99732eeb1..688ecd5f77 100644 --- a/spring-integration-gemfire/src/test/resources/log4j.xml +++ b/spring-integration-gemfire/src/test/resources/log4j.xml @@ -12,11 +12,11 @@ - + - +