From 3e8a774411597f17e8981ee0d689b0d4a657865d Mon Sep 17 00:00:00 2001 From: Josh Long Date: Wed, 17 Aug 2011 02:17:24 -0700 Subject: [PATCH] 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); - } - } - -}