From 3e8a774411597f17e8981ee0d689b0d4a657865d Mon Sep 17 00:00:00 2001 From: Josh Long Date: Wed, 17 Aug 2011 02:17:24 -0700 Subject: [PATCH 1/3] clean up... --- .../ContinuousQueryMessageProducer.java | 22 +-- .../store/KeyValueMessageGroupStore.java | 38 +---- .../gemfire/inbound/cq/CqClient-context.xml | 21 +-- .../gemfire/inbound/cq/CqServer-context.xml | 9 +- .../inbound/cq/CqServiceActivator.java | 32 ---- .../cq/client/CqClientConfiguration.java | 143 +++++++++--------- .../gemfire/inbound/cq/gfe-cache.properties | 2 +- .../cq/server/CqServerConfiguration.java | 88 ++++++++--- .../GemfireMessageStore-context.xml | 2 +- .../GemfireMessageStoreConfiguration.java | 70 --------- .../gemfire/store/messagegroupstore/Main.java | 41 ----- .../MessageGroupStoreActivator.java | 44 ------ .../store/messagegroupstore/Producer.java | 74 --------- 13 files changed, 157 insertions(+), 429 deletions(-) delete mode 100644 spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServiceActivator.java delete mode 100644 spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStoreConfiguration.java delete mode 100644 spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/Main.java delete mode 100644 spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/MessageGroupStoreActivator.java delete mode 100644 spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/Producer.java 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 2fcfd13871..919016af3c 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,30 +16,22 @@ 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.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 com.gemstone.gemfire.cache.Region; -import com.gemstone.gemfire.cache.client.Pool; -import com.gemstone.gemfire.cache.query.CqAttributes; -import com.gemstone.gemfire.cache.query.CqAttributesFactory; -import com.gemstone.gemfire.cache.query.CqEvent; -import com.gemstone.gemfire.cache.query.CqException; -import com.gemstone.gemfire.cache.query.CqListener; -import com.gemstone.gemfire.cache.query.CqQuery; -import com.gemstone.gemfire.cache.query.QueryService; - /** * 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. - * + * * @author Josh Long * @since 2.1 */ @@ -96,7 +88,7 @@ public class ContinuousQueryMessageProducer extends MessageProducerSupport { * 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 @@ -114,7 +106,7 @@ public class ContinuousQueryMessageProducer extends MessageProducerSupport { /** * 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. @@ -202,4 +194,4 @@ public class ContinuousQueryMessageProducer extends MessageProducerSupport { } } -} +} \ No newline at end of file diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/KeyValueMessageGroupStore.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/KeyValueMessageGroupStore.java index 324094df2b..d8228b805f 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/KeyValueMessageGroupStore.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/KeyValueMessageGroupStore.java @@ -35,23 +35,6 @@ import java.util.concurrent.ConcurrentMap; */ public class KeyValueMessageGroupStore extends AbstractMessageGroupStore { - /** - * Some operations can be done atomically and we should support them if possible - */ - // TODO: this is unused - //private boolean unmarkedIsConcurrentMap; - - /** - * Some operations can be done atomically and we should support them if possible - */ - // TODO: this is unused - //private boolean markedIsConcurrentMap; - - /** - * Some operations can be done atomically and we should support them if possible - */ - private boolean groupIdToMessageGroupIsConcurrentMap; - /** * Required {@link com.gemstone.gemfire.cache.Region} to managed the association of groups => {@link KeyValueMessageGroup} */ @@ -75,21 +58,10 @@ public class KeyValueMessageGroupStore extends AbstractMessageGroupStore { * @param marked the collection that will hold which messages are marked (delivered) * @param unmarked the collection that holds which messages are unmarked (not yet delivered) */ - public KeyValueMessageGroupStore(Map groupIdToMessageGroup, - Map> marked, Map> unmarked) { + public KeyValueMessageGroupStore(Map groupIdToMessageGroup, Map> marked, Map> unmarked) { this.marked = marked; - // TODO: these claim that a ConcurrentMap is required, but don't enforce it - Assert.notNull(this.marked, - "you must provide a ConcurrentMap to hold String => Message for marked"); this.unmarked = unmarked; - Assert.notNull(this.unmarked, - "you must provide a ConcurrentMap to hold String => Message for unmarked"); this.groupIdToMessageGroup = groupIdToMessageGroup; - Assert.notNull(this.groupIdToMessageGroup, - "you must provide a ConcurrentMap to hold associations of group ids to message groups ('groupIdToMessageGroup')"); - //this.markedIsConcurrentMap = marked instanceof ConcurrentMap; - //this.unmarkedIsConcurrentMap = unmarked instanceof ConcurrentMap; - this.groupIdToMessageGroupIsConcurrentMap = this.groupIdToMessageGroup instanceof ConcurrentMap; } @@ -142,14 +114,10 @@ public class KeyValueMessageGroupStore extends AbstractMessageGroupStore { } protected KeyValueMessageGroup getMessageGroupInternal(Object groupId) { - if (this.groupIdToMessageGroupIsConcurrentMap) { - ConcurrentMap cm = (ConcurrentMap) this.groupIdToMessageGroup; - cm.putIfAbsent(groupId, new KeyValueMessageGroup(groupId)); - } - else if (!groupIdToMessageGroup.containsKey(groupId)) { + if (!groupIdToMessageGroup.containsKey(groupId)) { groupIdToMessageGroup.put(groupId, new KeyValueMessageGroup(groupId)); } return ensureMessageGroupHasReferencesToRegions(groupIdToMessageGroup.get( groupId)); } -} +} \ No newline at end of file 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 index c1ba6b7d5e..a30f3ac5f9 100644 --- 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 @@ -1,26 +1,15 @@ + http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> - - - - - - - - - - + \ 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 index 6f37aa49ae..ee0387c201 100644 --- 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 @@ -1,18 +1,15 @@ - - + @@ -25,4 +22,4 @@ - + \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServiceActivator.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServiceActivator.java deleted file mode 100644 index 386d335ee8..0000000000 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServiceActivator.java +++ /dev/null @@ -1,32 +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; - -import org.springframework.integration.Message; -import org.springframework.integration.annotation.ServiceActivator; - -import com.gemstone.gemfire.cache.query.CqEvent; - -public class CqServiceActivator { - - @ServiceActivator - public void handleMessage(Message msg) throws Exception { - CqEvent cqEvent = msg.getPayload(); - System.out.println( "Received an event from the continuous query adapter: " +cqEvent ); - } - -} 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 index 096eb6d9ed..7124a17acd 100644 --- 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 @@ -16,48 +16,87 @@ package org.springframework.integration.gemfire.inbound.cq.client; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.context.support.ClassPathXmlApplicationContext; +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.gemfire.inbound.cq.CqServiceActivator; +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 com.gemstone.gemfire.cache.client.Pool; -import com.gemstone.gemfire.cache.client.PoolManager; +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 { - @Value("${region-name}") - private String regionName; + static private Log log = LogFactory.getLog(CqClientConfiguration.class); - @Value("${host}") - private String host; + private String regionName, host, query; - @Value("${region-query}") - private String query; - - @Value("${port}") private int port; - @Value("#{cqIn}") + @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 CqServiceActivator cqServiceActivator() { - return new CqServiceActivator(); + public Object cqServiceActivator() { + return new Object() { + @ServiceActivator + public void handleMessage(Message eventMessage) throws MessagingException { + CqEvent cqEvent = eventMessage.getPayload(); + log.info("Received an event from the continuous query adapter: " + cqEvent); + } + }; } - /* todo - protected ClientCache buildCache() throws Throwable { - return new ClientCacheFactory().create(); - } @Bean public ClientCache clientCache() throws Throwable { - return buildCache(); + return new ClientCacheFactory().create(); } @Bean @@ -65,62 +104,28 @@ public class CqClientConfiguration { 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 + @Bean public ContinuousQueryMessageProducer continuousQueryMessageProducer() throws Throwable { - ContinuousQueryMessageProducer continuousQueryMessageProducer - = new ContinuousQueryMessageProducer( this.clientRegion() , this.pool(), this.query); - continuousQueryMessageProducer.setDurable(true); - continuousQueryMessageProducer.setOutputChannel(this.messageChannel); - continuousQueryMessageProducer.setQueryName("pplQuery"); - return continuousQueryMessageProducer; + ContinuousQueryMessageProducer mp = new ContinuousQueryMessageProducer(this.clientRegion(), this.pool(), this.query); + mp.setDurable(true); + mp.setOutputChannel(this.messageChannel); + mp.setQueryName("pplQuery"); + return mp; } -*/ -/* protected CqQuery registerContinuousQuery(QueryService queryService, String name, String query, boolean durable, CqListener cqListener) throws Throwable { - CqAttributesFactory cqAttributesFactory = new CqAttributesFactory(); - cqAttributesFactory.addCqListener(cqListener); - CqAttributes attrs = cqAttributesFactory.create(); - CqQuery cqQuery = queryService.newCq(name, query, attrs, durable); - cqQuery.execute(); - return cqQuery; - }*/ + protected Pool buildPool(String host, int port) throws Throwable { - Pool pool = PoolManager.createFactory() - .addServer(host, port) - .setSubscriptionEnabled(true) - .create(host + "Pool"); - return pool; + return PoolManager.createFactory() + .addServer(host, port) + .setSubscriptionEnabled(true) + .create(host + "Pool"); } - /** - * the continuous query listener attached to the - */ - /*class MyContinuousQueryListener implements CqListener { - public void onEvent(CqEvent cqEvent) { - System.out.println("Received event: " + - new ToStringCreator(cqEvent)); - } - public void onError(CqEvent cqEvent) { - } - - public void close() { - } - }*/ - - - public static void main(String[] args) throws Exception { - new ClassPathXmlApplicationContext("org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml"); - while (true) { - Thread.sleep(1000 * 10); - } - } - -} +} \ No newline at end of file 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/inbound/cq/gfe-cache.properties index 04392af0d1..20fcecc488 100644 --- 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/inbound/cq/gfe-cache.properties @@ -1,3 +1,3 @@ log-level=warning name=Spring Integration GemFire World -bind-address=127.0.0.1 +bind-address=127.0.0.1 \ 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 index 30dc43f430..9c7ae872dd 100644 --- 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 @@ -16,35 +16,62 @@ package org.springframework.integration.gemfire.inbound.cq.server; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.context.support.ClassPathXmlApplicationContext; -import org.springframework.data.gemfire.GemfireTemplate; - 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 { - @Value("#{c}") - private Cache cache; - @Value("#{r}") - private Region region; + private static Log log = LogFactory.getLog(CqServerConfiguration.class); - @Value("${region-name}") - private String regionName; + @Autowired private Environment environment; - @Value("${host}") - private String host; + @Value("#{c}") private Cache cache; - @Value("${port}") + @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() { @@ -60,19 +87,30 @@ public class CqServerConfiguration { return cacheServer; } - public static void main(String[] args) throws Exception { - ClassPathXmlApplicationContext applicationContext = new ClassPathXmlApplicationContext( - "org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml"); - applicationContext.registerShutdownHook(); - applicationContext.start(); - GemfireTemplate gemfireTemplate = applicationContext.getBean(GemfireTemplate.class); - String letters = "abcdefghijk"; - while (true) { - Thread.sleep(1000 * 10); + @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/GemfireMessageStore-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml index cb861d2a0f..5359683ada 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml @@ -32,4 +32,4 @@ - + \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStoreConfiguration.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStoreConfiguration.java deleted file mode 100644 index 74f7b86df7..0000000000 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStoreConfiguration.java +++ /dev/null @@ -1,70 +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.store.messagegroupstore; - -import java.util.Map; - -import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.integration.Message; -import org.springframework.integration.aggregator.CorrelationStrategy; -import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy; -import org.springframework.integration.aggregator.ReleaseStrategy; -import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy; -import org.springframework.integration.gemfire.store.KeyValueMessageGroup; -import org.springframework.integration.gemfire.store.KeyValueMessageGroupStore; - -/** - * Our aggregator needs a {@link org.springframework.integration.gemfire.store.KeyValueMessageGroupStore}. - * This handles configuration of the ancillary objects. - * - * @author Josh Long - * @since 2.1 - */ -@Configuration -public class GemfireMessageStoreConfiguration { - - @Value("${correlation-header}") - private String correlationHeader; - - @Value("#{unmarkedRegion}") - private Map> unmarked; - - @Value("#{markedRegion}") - private Map> marked; - - @Value("#{messageGroupRegion}") - private Map messageGroupRegion; - - - @Bean - public ReleaseStrategy releaseStrategy() { - return new SequenceSizeReleaseStrategy(false); - } - - @Bean - public CorrelationStrategy correlationStrategy() { - return new HeaderAttributeCorrelationStrategy(this.correlationHeader); - } - - @Bean - public KeyValueMessageGroupStore gemfireMessageGroupStore() { - return new KeyValueMessageGroupStore(this.messageGroupRegion, this.marked , this.unmarked ); - } - -} diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/Main.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/Main.java deleted file mode 100644 index 1c6450e90a..0000000000 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/Main.java +++ /dev/null @@ -1,41 +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.store.messagegroupstore; - -import java.util.Arrays; - -import org.springframework.context.support.ClassPathXmlApplicationContext; - -/** - * Simple example demonstrating the use of a {@link org.springframework.integration.gemfire.store.KeyValueMessageGroupStore}. - * - * @author Josh Long - * @since 2.1 - */ -public class Main { - - public static void main(String[] args) throws Throwable { - ClassPathXmlApplicationContext classPathXmlApplicationContext = new ClassPathXmlApplicationContext( - "/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml"); - Producer producer = classPathXmlApplicationContext.getBean(Producer.class); - for(int i =0 ; i < 10 ; i++ ) { - producer.sendManyMessages(i, Arrays.asList("1,2,3,4,5".split(","))); - } - Thread.sleep( 1000 * 10 ); - } - -} diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/MessageGroupStoreActivator.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/MessageGroupStoreActivator.java deleted file mode 100644 index 988f11fc8a..0000000000 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/MessageGroupStoreActivator.java +++ /dev/null @@ -1,44 +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.store.messagegroupstore; - -import java.util.Collection; - -import org.springframework.integration.Message; -import org.springframework.integration.annotation.ServiceActivator; -import org.springframework.stereotype.Component; -import org.springframework.util.StringUtils; - -/** - * @author Josh Long - * @since 2.1 - */ -@Component -public class MessageGroupStoreActivator { - - @ServiceActivator - public void activate(Message> msg) throws Throwable { - Collection payloads = msg.getPayload(); - StringBuffer buffer = new StringBuffer(); - for (int i = 0; i < 100; i++) { - buffer.append("-"); - } - System.out.println(buffer.toString()); - System.out.println(StringUtils.collectionToCommaDelimitedString(payloads)); - } - -} diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/Producer.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/Producer.java deleted file mode 100644 index 31a8c53f32..0000000000 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/Producer.java +++ /dev/null @@ -1,74 +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.store.messagegroupstore; - -import java.util.Collection; - -import javax.annotation.PostConstruct; - -import org.springframework.beans.factory.annotation.Value; -import org.springframework.integration.Message; -import org.springframework.integration.MessageChannel; -import org.springframework.integration.core.MessagingTemplate; -import org.springframework.integration.support.MessageBuilder; -import org.springframework.stereotype.Component; -import org.springframework.util.Assert; - -/** - * Simple endpoint that we can use to send in a lot of test messages. - * - * @author Josh Long - * @since 2.1 - */ -@Component -public class Producer { - - private MessagingTemplate messagingTemplate = new MessagingTemplate(); - - @Value("#{i}") - private MessageChannel messageChannel; - - @Value("${correlation-header}") - private String correlationHeader ; - - - @PostConstruct - public void start() throws Throwable { - this.messagingTemplate.setDefaultChannel(this.messageChannel); - } - - /** - * @param lines - * @throws Throwable - */ - public void sendManyMessages(int correlationValue, Collection lines) throws Throwable { - Assert.notNull( lines, "the collection must be non-null"); - Assert.notEmpty( lines, "the collection must not be empty"); - int ctr = 0; - int size = lines.size() ; - for (String l : lines) { - Message msg = MessageBuilder.withPayload(l) - .setCorrelationId( this.correlationHeader) - .setHeader(this.correlationHeader , correlationValue) - .setSequenceNumber(++ctr) - .setSequenceSize(size) - .build(); - this.messagingTemplate.send(msg); - } - } - -} From dc5d821aeeab15a0541f8ca9866c709ceb18f708 Mon Sep 17 00:00:00 2001 From: Josh Long Date: Wed, 17 Aug 2011 02:17:51 -0700 Subject: [PATCH 2/3] clean up... --- .../GemfireMessageGroupStoreTest.java | 42 ++++ ...ireMessageGroupStoreTestConfiguration.java | 211 ++++++++++++++++++ .../GemfireMessageGroupStoreTest-context.xml | 24 ++ 3 files changed, 277 insertions(+) create mode 100644 spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest.java create mode 100644 spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTestConfiguration.java create mode 100644 spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest-context.xml diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest.java new file mode 100644 index 0000000000..adb90d2948 --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest.java @@ -0,0 +1,42 @@ +package org.springframework.integration.gemfire.store.messagegroupstore; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.context.support.AnnotationConfigContextLoader; +import java.util.Collection; +import java.util.List; +import java.util.Set; + +/** + * Tests the Gemfire {@link org.springframework.integration.store.MessageGroupStore} implementation, {@link org.springframework.integration.gemfire.store.GemfireMessageGroupStore} + * + * @author Josh Long + */ + +@RunWith(SpringJUnit4ClassRunner.class) +@ContextConfiguration(loader = AnnotationConfigContextLoader.class, classes = {GemfireMessageGroupStoreTestConfiguration.class}) +public class GemfireMessageGroupStoreTest { + + @Autowired private GemfireMessageGroupStoreTestConfiguration.FakeMessageConsumer consumer; + + private List letters = GemfireMessageGroupStoreTestConfiguration.LIST_OF_STRINGS; + + @Test + public void testGemfireMessageGroupStore() throws Throwable { + Set> batches = consumer.getBatches(); + Assert.assertTrue(batches.size() == 10); + for (Collection collection : batches) { + Assert.assertTrue(letters.size() == collection.size()); + for (String c : this.letters) { + Assert.assertTrue(collection.contains(c)); + } + for (Object o : collection) { + Assert.assertTrue(o instanceof String); + } + } + } +} \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTestConfiguration.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTestConfiguration.java new file mode 100644 index 0000000000..26ad2879f8 --- /dev/null +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTestConfiguration.java @@ -0,0 +1,211 @@ +/* + * 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.store.messagegroupstore; + +import com.gemstone.gemfire.cache.Cache; +import com.gemstone.gemfire.cache.Region; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.SmartLifecycle; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.ImportResource; +import org.springframework.data.gemfire.CacheFactoryBean; +import org.springframework.data.gemfire.RegionFactoryBean; +import org.springframework.integration.Message; +import org.springframework.integration.MessageChannel; +import org.springframework.integration.aggregator.CorrelationStrategy; +import org.springframework.integration.aggregator.HeaderAttributeCorrelationStrategy; +import org.springframework.integration.aggregator.ReleaseStrategy; +import org.springframework.integration.aggregator.SequenceSizeReleaseStrategy; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.core.MessagingTemplate; +import org.springframework.integration.gemfire.store.KeyValueMessageGroup; +import org.springframework.integration.gemfire.store.KeyValueMessageGroupStore; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.util.Assert; + +import java.util.*; + +/** + * Our aggregator needs a {@link org.springframework.integration.gemfire.store.KeyValueMessageGroupStore}. + * This handles configuration of the ancillary objects. + * + * @author Josh Long + * @since 2.1 + */ +@Configuration +@ImportResource("org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest-context.xml") +public class GemfireMessageGroupStoreTestConfiguration { + + public static List LIST_OF_STRINGS = Arrays.asList("1,2,3,4,5".split(",")); + + static private Log log = LogFactory.getLog(GemfireMessageGroupStoreTestConfiguration.class); + + @Value("${correlation-header}") + private String correlationHeader; + + @Bean + public Cache cache() throws Throwable { + CacheFactoryBean cacheFactoryBean = new CacheFactoryBean(); + cacheFactoryBean.afterPropertiesSet(); + return cacheFactoryBean.getObject(); + } + + + @Bean + public Region messageGroupRegion() throws Throwable { + RegionFactoryBean regionFactoryBean = new RegionFactoryBean(); + regionFactoryBean.setName("messageGroupRegion"); + regionFactoryBean.setCache(cache()); + regionFactoryBean.afterPropertiesSet(); + return regionFactoryBean.getObject(); + } + + @Bean + public Region> unmarkedRegion() throws Throwable { + RegionFactoryBean> regionFactoryBean = new RegionFactoryBean>(); + regionFactoryBean.setName("unmarkedRegion"); + regionFactoryBean.setCache(cache()); + regionFactoryBean.afterPropertiesSet(); + return regionFactoryBean.getObject(); + } + + @Bean + public Region> markedRegion() throws Throwable { + RegionFactoryBean> regionFactoryBean = new RegionFactoryBean>(); + regionFactoryBean.setName("markedRegion"); + regionFactoryBean.setCache(cache()); + regionFactoryBean.afterPropertiesSet(); + return regionFactoryBean.getObject(); + } + + + @Bean(name = "messageGroupStoreActivator") + public FakeMessageConsumer serviceActivator() { + return new FakeMessageConsumer(); + } + + @Bean + public ReleaseStrategy releaseStrategy() { + return new SequenceSizeReleaseStrategy(false); + } + + @Bean + public CorrelationStrategy correlationStrategy() { + return new HeaderAttributeCorrelationStrategy(this.correlationHeader); + } + + @Bean + public KeyValueMessageGroupStore gemfireMessageGroupStore() throws Throwable { + return new KeyValueMessageGroupStore(messageGroupRegion(), markedRegion(), unmarkedRegion()); + } + + @Bean + public FakeMessageProducer producer() { + return new FakeMessageProducer(); + } + + static public class FakeMessageConsumer { + + private Set> batches = new HashSet>(); + + public Set> getBatches() { + return this.batches; + } + + @ServiceActivator + public void activateAsMessagesArriveInBatches(Message> msg) throws Throwable { + Collection payloads = msg.getPayload(); + batches.add(payloads); + if (log.isDebugEnabled()) { + log.debug(payloads); + } + + } + + } + + static public class FakeMessageProducer implements InitializingBean, SmartLifecycle { + public boolean isAutoStartup() { + return true; + } + + public void stop(Runnable callback) { + } + + public int getPhase() { + return 0; + } + + @Autowired @Qualifier("i") + private MessageChannel messageChannel; + + private MessagingTemplate messagingTemplate = new MessagingTemplate(); + + private volatile boolean running = false; + + @Value("${correlation-header}") + private String correlationHeader; + + public void sendManyMessages(int correlationValue, Collection lines) throws Throwable { + Assert.notNull(lines, "the collection must be non-null"); + Assert.notEmpty(lines, "the collection must not be empty"); + int ctr = 0; + int size = lines.size(); + for (String l : lines) { + Message msg = MessageBuilder.withPayload(l) + .setCorrelationId(this.correlationHeader) + .setHeader(this.correlationHeader, correlationValue) + .setSequenceNumber(++ctr) + .setSequenceSize(size) + .build(); + this.messagingTemplate.send(msg); + } + } + + + public void afterPropertiesSet() throws Exception { + this.messagingTemplate.setDefaultChannel(this.messageChannel); + } + + + public void start() { + for (int i = 0; i < 10; i++) { + try { + running = true; + sendManyMessages(i, LIST_OF_STRINGS); + running = false; + } catch (Throwable throwable) { + throw new RuntimeException(throwable); + } + } + } + + public void stop() { + } + + public boolean isRunning() { + return running; + } + + } +} \ No newline at end of file diff --git a/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest-context.xml b/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest-context.xml new file mode 100644 index 0000000000..68e9974f8c --- /dev/null +++ b/spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest-context.xml @@ -0,0 +1,24 @@ + + + + + + + + + + + + + + + + \ No newline at end of file From 197021cd63ff196a26be40400b988337732b683b Mon Sep 17 00:00:00 2001 From: Josh Long Date: Wed, 17 Aug 2011 02:29:19 -0700 Subject: [PATCH 3/3] ensured everything worked after merge.. --- .../gemfire/inbound/cq/client/CqClientConfiguration.java | 2 +- .../messagegroupstore/GemfireMessageGroupStoreTest.java | 7 +++++-- 2 files changed, 6 insertions(+), 3 deletions(-) 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 index 7124a17acd..919f9dff26 100644 --- 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 @@ -89,7 +89,7 @@ public class CqClientConfiguration { @ServiceActivator public void handleMessage(Message eventMessage) throws MessagingException { CqEvent cqEvent = eventMessage.getPayload(); - log.info("Received an event from the continuous query adapter: " + cqEvent); + System.out.println("Received an event from the continuous query adapter: " + cqEvent) ; } }; } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest.java index adb90d2948..4b85af5e77 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest.java @@ -12,11 +12,14 @@ import java.util.List; import java.util.Set; /** - * Tests the Gemfire {@link org.springframework.integration.store.MessageGroupStore} implementation, {@link org.springframework.integration.gemfire.store.GemfireMessageGroupStore} + * Tests the Gemfire {@link org.springframework.integration.store.MessageGroupStore} implementation, + * {@link org.springframework.integration.gemfire.store.GemfireMessageGroupStore}. * + * It tests the {@link org.springframework.integration.store.MessageGroupStore} by sending 10 batches of letters (all of the same width), + * and then counting on the other end that indeed all 10 batches arrived and that all letters expected are there. + * * * @author Josh Long */ - @RunWith(SpringJUnit4ClassRunner.class) @ContextConfiguration(loader = AnnotationConfigContextLoader.class, classes = {GemfireMessageGroupStoreTestConfiguration.class}) public class GemfireMessageGroupStoreTest {