From dc5d821aeeab15a0541f8ca9866c709ceb18f708 Mon Sep 17 00:00:00 2001 From: Josh Long Date: Wed, 17 Aug 2011 02:17:51 -0700 Subject: [PATCH] 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