Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #18 from joshlong/INT2060

Browse Source 
ensured everything worked after merge
  general clean up

Issue: INT-2060
This commit is contained in:
Mark Fisher
2011-08-17 22:39:14 -04:00
parent 6544414544 197021cd63
commit 17c8759ceb
16 changed files with 437 additions and 429 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/ContinuousQueryMessageProducer.java
View File

@@ -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 {
}
}
}
}

38
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/KeyValueMessageGroupStore.java
View File

@@ -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<Object, KeyValueMessageGroup> groupIdToMessageGroup,
Map<String, Message<?>> marked, Map<String, Message<?>> unmarked) {
public KeyValueMessageGroupStore(Map<Object, KeyValueMessageGroup> groupIdToMessageGroup, Map<String, Message<?>> marked, Map<String, Message<?>> 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<Object, KeyValueMessageGroup> cm = (ConcurrentMap<Object, KeyValueMessageGroup>) 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));
}
}
}

21
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqClient-context.xml
View File

@@ -1,26 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:gfe="http://www.springframework.org/schema/gemfire" xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:component-scan base-package="org.springframework.integration.gemfire.inbound.cq.client"/>
<context:property-placeholder location="org/springframework/integration/gemfire/inbound/cq/common.properties"/>
<!--<util:properties id="props" location="org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties"/>-->
<int:channel id="cqIn"/>
<int:service-activator input-channel="cqIn" ref="cqServiceActivator"/>
</beans>
</beans>

9
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServer-context.xml
View File

@@ -1,18 +1,15 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:gfe="http://www.springframework.org/schema/gemfire" xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
xsi:schemaLocation="
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<!-- infrastrcture-->
<context:component-scan base-package="org.springframework.integration.gemfire.inbound.cq.server"/>
<context:property-placeholder location="org/springframework/integration/gemfire/inbound/cq/common.properties"/>
<!-- setup the cache-->
@@ -25,4 +22,4 @@
<!-- region -->
<gfe:replicated-region id="r" name="${region-name}" cache-ref="c" />
</beans>
</beans>

32
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/CqServiceActivator.java
View File

@@ -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<CqEvent> msg) throws Exception {
CqEvent cqEvent = msg.getPayload();
System.out.println( "Received an event from the continuous query adapter: " +cqEvent );
}
}

143
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/client/CqClientConfiguration.java
View File

@@ -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<CqEvent> eventMessage) throws MessagingException {
CqEvent cqEvent = eventMessage.getPayload();
System.out.println("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);
}
}
}
}

2
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties
View File

@@ -1,3 +1,3 @@
log-level=warning
name=Spring Integration GemFire World
bind-address=127.0.0.1
bind-address=127.0.0.1

88
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/server/CqServerConfiguration.java
View File

@@ -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<String, ?> 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<String, ?> 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);
}
}
}
}
}

45
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest.java Normal file
View File

@@ -0,0 +1,45 @@
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}.
*
* 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 {
@Autowired private GemfireMessageGroupStoreTestConfiguration.FakeMessageConsumer consumer;
private List<String> letters = GemfireMessageGroupStoreTestConfiguration.LIST_OF_STRINGS;
@Test
public void testGemfireMessageGroupStore() throws Throwable {
Set<Collection<Object>> batches = consumer.getBatches();
Assert.assertTrue(batches.size() == 10);
for (Collection<Object> 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);
}
}
}
}

211
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTestConfiguration.java Normal file
View File

@@ -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<String> 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<Object, KeyValueMessageGroup> messageGroupRegion() throws Throwable {
RegionFactoryBean<Object, KeyValueMessageGroup> regionFactoryBean = new RegionFactoryBean<Object, KeyValueMessageGroup>();
regionFactoryBean.setName("messageGroupRegion");
regionFactoryBean.setCache(cache());
regionFactoryBean.afterPropertiesSet();
return regionFactoryBean.getObject();
}
@Bean
public Region<String, Message<?>> unmarkedRegion() throws Throwable {
RegionFactoryBean<String, Message<?>> regionFactoryBean = new RegionFactoryBean<String, Message<?>>();
regionFactoryBean.setName("unmarkedRegion");
regionFactoryBean.setCache(cache());
regionFactoryBean.afterPropertiesSet();
return regionFactoryBean.getObject();
}
@Bean
public Region<String, Message<?>> markedRegion() throws Throwable {
RegionFactoryBean<String, Message<?>> regionFactoryBean = new RegionFactoryBean<String, Message<?>>();
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<Collection<Object>> batches = new HashSet<Collection<Object>>();
public Set<Collection<Object>> getBatches() {
return this.batches;
}
@ServiceActivator
public void activateAsMessagesArriveInBatches(Message<Collection<Object>> msg) throws Throwable {
Collection<Object> 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<String> 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;
}
}
}

2
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStore-context.xml
View File

@@ -32,4 +32,4 @@
<gfe:replicated-region id="markedRegion" cache-ref="c"/>
<gfe:replicated-region id="messageGroupRegion" cache-ref="c"/>
</beans>
</beans>

70
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageStoreConfiguration.java
View File

@@ -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<String, Message<?>> unmarked;
@Value("#{markedRegion}")
private Map<String, Message<?>> marked;
@Value("#{messageGroupRegion}")
private Map<Object, KeyValueMessageGroup> 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 );
}
}

41
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/Main.java
View File

@@ -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 );
}
}

44
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/MessageGroupStoreActivator.java
View File

@@ -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<Collection<Object>> msg) throws Throwable {
Collection<Object> 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));
}
}

74
spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/store/messagegroupstore/Producer.java
View File

@@ -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<String> 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);
}
}
}

24
spring-integration-gemfire/src/test/resources/org/springframework/integration/gemfire/store/messagegroupstore/GemfireMessageGroupStoreTest-context.xml Normal file
View File

@@ -0,0 +1,24 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:util="http://www.springframework.org/schema/util"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-2.0.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<context:property-placeholder location="org/springframework/integration/gemfire/inbound/cq/common.properties"/>
<int:channel id="i"/>
<int:aggregator release-strategy="releaseStrategy" correlation-strategy="correlationStrategy" message-store="gemfireMessageGroupStore" input-channel="i" output-channel="o" />
<int:channel id="o"/>
<int:service-activator input-channel="o" ref="messageGroupStoreActivator" />
<util:properties id="props" location="org/springframework/integration/gemfire/inbound/cq/gfe-cache.properties"/>
</beans>