INT-2032 migrated 'spring-integration-gemfire' to main branch from sandbox in preparation for 2.1 development

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java

@@ -0,0 +1,147 @@
+/*
+ * 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;
+
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.Set;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.springframework.expression.Expression;
+import org.springframework.expression.spel.standard.SpelExpressionParser;
+import org.springframework.integration.endpoint.MessageProducerSupport;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.util.Assert;
+
+import com.gemstone.gemfire.cache.CacheListener;
+import com.gemstone.gemfire.cache.EntryEvent;
+import com.gemstone.gemfire.cache.Region;
+import com.gemstone.gemfire.cache.util.CacheListenerAdapter;
+
+/**
+ * An inbound endpoint that listens to a GemFire region for events and then publishes Messages to
+ * a channel. The default supported event types are CREATED and UPDATED. See the {@link EventType}
+ * enum for all options. A SpEL expression may be provided to generate a Message payload by
+ * evaluating that expression against the {@link EntryEvent} instance as the root object. If no
+ * payloadExpression is provided, the {@link EntryEvent} itself will be the payload.
+ * 
+ * @author Mark Fisher
+ * @since 2.1
+ */
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class CacheListeningMessageProducer extends MessageProducerSupport {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	private final Region region;
+
+	private final CacheListener<?, ?> listener;
+
+	private volatile Set<EventType> supportedEventTypes =
+			new HashSet<EventType>(Arrays.asList(EventType.CREATED, EventType.UPDATED));
+
+	private volatile Expression payloadExpression;
+
+	private final SpelExpressionParser parser = new SpelExpressionParser();
+
+
+	public CacheListeningMessageProducer(Region<?, ?> region) {
+		Assert.notNull(region, "region must not be null");
+		this.region = region;
+		this.listener = new MessageProducingCacheListener();
+	}
+
+
+	public void setSupportedEventTypes(EventType... eventTypes) {
+		Assert.notEmpty(eventTypes, "eventTypes must not be empty");
+		this.supportedEventTypes = new HashSet<EventType>(Arrays.asList(eventTypes));
+	}
+
+	public void setPayloadExpression(String payloadExpression) {
+		if (payloadExpression == null) {
+			this.payloadExpression = null;
+		}
+		else {
+			this.payloadExpression = this.parser.parseExpression(payloadExpression);
+		}
+	}
+
+	@Override
+	protected void doStart() {
+		if (logger.isInfoEnabled()) {
+			logger.info("adding MessageProducingCacheListener to GemFire Region '" + this.region.getName() + "'");
+		}
+		this.region.getAttributesMutator().addCacheListener(this.listener);
+	}
+
+	@Override
+	protected void doStop() {
+		if (logger.isInfoEnabled()) {
+			logger.info("removing MessageProducingCacheListener from GemFire Region '" + this.region.getName() + "'");
+		}
+		this.region.getAttributesMutator().removeCacheListener(this.listener);
+	}
+
+
+	private class MessageProducingCacheListener extends CacheListenerAdapter {
+
+		@Override
+		public void afterCreate(EntryEvent event) {
+			if (supportedEventTypes.contains(EventType.CREATED)) {
+				this.processEvent(event);
+			}
+		}
+
+		@Override
+		public void afterUpdate(EntryEvent event) {
+			if (supportedEventTypes.contains(EventType.UPDATED)) {
+				this.processEvent(event);
+			}
+		}
+
+		@Override
+		public void afterInvalidate(EntryEvent event) {
+			if (supportedEventTypes.contains(EventType.INVALIDATED)) {
+				this.processEvent(event);
+			}
+		}
+
+		@Override
+		public void afterDestroy(EntryEvent event) {
+			if (supportedEventTypes.contains(EventType.DESTROYED)) {
+				this.processEvent(event);
+			}
+		}
+
+		private void processEvent(EntryEvent event) {
+			if (payloadExpression != null) {
+				Object evaluationResult = payloadExpression.getValue(event);
+				this.publish(evaluationResult);
+			}
+			else {
+				this.publish(event);
+			}
+		}
+
+		private void publish(Object payload) {
+			sendMessage(MessageBuilder.withPayload(payload).build());
+		}
+	}
+
+}

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

@@ -0,0 +1,205 @@
+/*
+ * 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;
+
+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
+ */
+public class ContinuousQueryMessageProducer extends MessageProducerSupport {
+
+	private final Log logger = LogFactory.getLog(this.getClass());
+
+	/**
+	 * Not sure yet if there's a way to avoid depending on this.
+	 */
+	private volatile Pool pool;
+
+	/**
+	 * Must be provided by the client of this class
+	 */
+	private final Region<?, ?> region;
+
+	/**
+	 * Is the queryString durable (optional)
+	 */
+	private volatile boolean durable = false;
+
+	/**
+	 * the {@link com.gemstone.gemfire.cache.query.CqQuery} instance created and
+	 * registered with the server
+	 */
+	private volatile CqQuery cqQuery;
+
+	/**
+	 * the query to be registered against the cache
+	 */
+	private final String queryString;
+
+	/**
+	 * a reference to a {@link com.gemstone.gemfire.cache.query.QueryService}
+	 * that is obtained through the #regionService instance.
+	 */
+	private volatile QueryService queryService;
+
+	/**
+	 * used when building the queryString itself - optional
+	 */
+	private volatile String queryName;
+
+	/**
+	 * a {@link com.gemstone.gemfire.cache.query.CqAttributesFactory} to generate
+	 * the {@link com.gemstone.gemfire.cache.query.CqAttributes} that in turn
+	 * hold the reference to the listener that we register to in turn funnel
+	 * messages to the clients of this adapter.
+	 */
+	private final CqAttributesFactory cqAttributesFactory = new CqAttributesFactory();
+
+	/**
+	 * the adapter requires a query string to continuously evaluate as well as a
+	 * {@link com.gemstone.gemfire.cache.Region} against which to evaluate the
+	 * query.
+	 * 
+	 * @param region
+	 *            the region against which the query should be evaluated
+	 * @param queryString
+	 *            the query string
+	 */
+	public ContinuousQueryMessageProducer(Region<?, ?> region, Pool pool, String queryString) {
+		this.region = region;
+		Assert.notNull(this.region, "You must provide a reference to a 'Region'");
+		this.pool = pool;
+		Assert.notNull(this.pool, "You must provide a 'pool'");
+		this.queryString = queryString;
+		Assert.hasText(this.queryString, "You must provide a queryString to evaluate against the region");
+	}
+
+	/**
+	 * 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.
+	 */
+	public void setDurable(boolean durable) {
+		this.durable = durable;
+	}
+
+	/**
+	 * Specify the name of the queryString (optional).
+	 */
+	public void setQueryName(String queryName) {
+		this.queryName = queryName;
+	}
+
+	@Override
+	protected void doStart() {
+		try {
+			cqQuery.execute();
+		}
+		catch (Throwable e) {
+			throw new RuntimeException("Failed to start the continuous query", e);
+		}
+	}
+
+	@Override
+	protected void doStop() {
+		try {
+			this.cqQuery.stop();
+		}
+		catch (CqException e) {
+			throw new RuntimeException("Failed to stop the continuous query", e);
+		}
+	}
+
+	/**
+	 * hook to handle registration of the query
+	 */
+	private CqQuery registerContinuousQuery(QueryService queryService,
+			String name, String query, boolean durable, CqListener cqListener) throws Throwable {
+		cqAttributesFactory.addCqListener(cqListener);
+		CqAttributes attrs = cqAttributesFactory.create();
+		CqQuery cqQuery = queryService.newCq(name, query, attrs, durable);
+		return cqQuery;
+	}
+
+	@Override
+	protected void onInit() {
+		try {
+			super.onInit();
+
+			// regionService = this.region.getRegionService();
+			queryService = this.pool.getQueryService();
+			String defaultName = String.format("%s-%s-query",
+					getComponentName() + "", getComponentType() + "");
+			queryName = StringUtils.hasText(queryName) ? queryName : defaultName;
+			this.cqQuery = registerContinuousQuery(queryService, queryName,
+					this.queryString, this.durable,
+					new MessageProducingCqListener());
+		}
+		catch (Throwable e) {
+			throw new RuntimeException("Couldn't properly setup the " + getClass().getName(), e);
+		}
+	}
+
+
+	/**
+	 * Listener that listens for any events being broadcast as a result of the
+	 * evaluation of a continuous query {@link CqQuery}.
+	 */
+	class MessageProducingCqListener implements CqListener {
+
+		public void onEvent(CqEvent cqEvent) {
+			Message<CqEvent> cqEventMessage = MessageBuilder.withPayload(cqEvent).build();
+			sendMessage(cqEventMessage);
+		}
+
+		public void onError(CqEvent cqEvent) {
+			logger.debug("error on " + getClass() + " (a CqListener) ");
+			throw new RuntimeException("error when interacting with region.", cqEvent.getThrowable());
+		}
+
+		public void close() {
+			logger.debug(getClass() + " close() called");
+		}
+	}
+
+}

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/EventType.java

@@ -0,0 +1,35 @@
+/*
+ * 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;
+
+/**
+ * Enumeration of GemFire event types.
+ * 
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public enum EventType {
+
+	CREATED,
+
+	UPDATED,
+
+	DESTROYED,
+
+	INVALIDATED
+
+}

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/outbound/CacheWritingMessageHandler.java

@@ -0,0 +1,65 @@
+/*
+ * 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.outbound;
+
+import java.util.Map;
+
+import org.springframework.data.gemfire.GemfireCallback;
+import org.springframework.data.gemfire.GemfireTemplate;
+import org.springframework.integration.Message;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.util.Assert;
+
+import com.gemstone.gemfire.GemFireCheckedException;
+import com.gemstone.gemfire.GemFireException;
+import com.gemstone.gemfire.cache.Region;
+
+/**
+ * A {@link MessageHandler} implementation that writes to a GemFire Region.
+ * The Message's payload must be an instance of java.util.Map.
+ * 
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public class CacheWritingMessageHandler implements MessageHandler {
+
+	private final GemfireTemplate gemfireTemplate = new GemfireTemplate();
+
+
+	@SuppressWarnings("rawtypes")
+	public CacheWritingMessageHandler(Region region) {
+		Assert.notNull(region, "region must not be null");
+		this.gemfireTemplate.setRegion(region);
+		this.gemfireTemplate.afterPropertiesSet();
+	}
+
+
+	public void handleMessage(Message<?> message) {
+		// TODO: add support for more options to get key/value (SpEL?)
+		Object payload = message.getPayload();
+		Assert.isTrue(payload instanceof Map, "only Map payloads are supported");
+		final Map<?, ?> map = (Map<?, ?>) payload;
+		this.gemfireTemplate.execute(new GemfireCallback<Object>() {
+			@SuppressWarnings({ "rawtypes", "unchecked" })
+			public Object doInGemfire(Region region) throws GemFireCheckedException, GemFireException {
+				region.putAll(map);
+				return null;
+			}
+		});
+	}
+
+}

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageGroupStore.java

@@ -0,0 +1,41 @@
+/*
+ * 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;
+
+import org.springframework.integration.Message;
+
+import com.gemstone.gemfire.cache.Region;
+
+/**
+ * Provides GemFire specific support as a backing key-value based {@link org.springframework.integration.store.MessageGroupStore}.
+ * Currently, this support is limited to explicitly depending on GemFire {@link com.gemstone.gemfire.cache.Region}s, but
+ * might conceptually also support optimized key traversal (using a {@link com.gemstone.gemfire.cache.query.Query}, for example).
+ *
+ * @author Josh Long
+ * @since 2.1
+ * @see {@link org.springframework.integration.gemfire.store.KeyValueMessageGroupStore}
+ */
+public class GemfireMessageGroupStore extends KeyValueMessageGroupStore {
+
+	public GemfireMessageGroupStore(
+			Region<Object, KeyValueMessageGroup> groupIdToMessageGroup,
+			Region<String, Message<?>> marked,
+			Region<String, Message<?>> unmarked ) {
+		super(groupIdToMessageGroup, marked, unmarked);
+	}
+
+}

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/GemfireMessageStore.java

@@ -0,0 +1,57 @@
+/*
+ * 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;
+
+import java.util.UUID;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.store.MessageStore;
+import org.springframework.util.Assert;
+
+import com.gemstone.gemfire.cache.Region;
+
+/**
+ * @author Mark Fisher
+ * @since 2.1
+ */
+public class GemfireMessageStore implements MessageStore {
+
+	private final Region<UUID, Message<?>> region;
+
+	public GemfireMessageStore(Region<UUID, Message<?>> region) {
+		Assert.notNull(region, "region must not be null");
+		this.region = region;
+	}
+
+	public Message<?> getMessage(UUID id) {
+		return this.region.get(id);
+	}
+
+	public <T> Message<T> addMessage(Message<T> message) {
+		this.region.put(message.getHeaders().getId(), message);
+		return message;
+	}
+
+	public Message<?> removeMessage(UUID id) {
+		return this.region.remove(id);
+	}
+
+	public int getMessageCount() {
+		return this.region.size();
+	}
+
+}

spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/store/KeyValueMessageGroup.java

@@ -0,0 +1,339 @@
+/*
+ * 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;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentMap;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.store.MessageGroup;
+
+/**
+ * A {@link org.springframework.integration.store.MessageGroup} that manipulates keys and values to provide persistence.
+ * Responsible for managing one group's messages as a {@link org.springframework.integration.store.MessageGroup}.
+ *
+ * @author Josh Long
+ * @since 2.1
+ */
+public class KeyValueMessageGroup implements MessageGroup, Serializable {
+
+	/**
+	 * this should not be persisted. it's passed in through {@link KeyValueMessageGroupStore}, which has the reference to the {@link java.util.concurrent.ConcurrentMap} instance that should be set here
+	 */
+	private transient Map<String, Message<?>> marked;
+
+	/**
+	 * this should not be persisted. it's passed in through {@link KeyValueMessageGroupStore}, which has the reference to the {@link java.util.concurrent.ConcurrentMap} instance that should be set here
+	 */
+	private transient Map<String, Message<?>> unmarked;
+
+	/**
+	 * the #groupId is the unique ID to associate this aggregation of {@link org.springframework.integration.Message}s
+	 */
+	private Object groupId;
+
+	/**
+	 * passed in through the {@link org.springframework.integration.store.MessageGroupStore}
+	 */
+	private long timestamp;
+
+
+	/**
+	 * default javabean ctor (so that this object plays well as a {@link java.io.Serializable} object)
+	 */
+	public KeyValueMessageGroup() {
+	}
+
+	public KeyValueMessageGroup(Object groupId) {
+		this(groupId, System.currentTimeMillis(), null, null);
+	}
+
+	public KeyValueMessageGroup(Object groupId, long timestamp,
+								ConcurrentMap<String, Message<?>> marked,
+								ConcurrentMap<String, Message<?>> unmarked) {
+		this.groupId = groupId;
+		this.timestamp = timestamp;
+		this.marked = marked;
+		this.unmarked = unmarked;
+	}
+
+	public KeyValueMessageGroup(Object groupId,
+								ConcurrentMap<String, Message<?>> marked,
+								ConcurrentMap<String, Message<?>> unmarked) {
+		this(groupId, System.currentTimeMillis(), marked, unmarked);
+	}
+
+
+	@Override
+	public int hashCode() {
+		return groupId.hashCode();
+	}
+
+	@Override
+	public boolean equals(Object obj) {
+		if (obj instanceof KeyValueMessageGroup) {
+			Object otherGroupId = ((KeyValueMessageGroup) obj).getGroupId();
+			return getGroupId().equals(otherGroupId);
+		}
+		return false;
+	}
+
+	public void setUnmarked(Map<String, Message<?>> unmarked) {
+		this.unmarked = unmarked;
+	}
+
+	public void setMarked( Map<String, Message<?>> marked) {
+		this.marked = marked;
+	}
+
+	/**
+	 * @return the timestamp (milliseconds since epoch) associated with the creation of this group
+	 */
+	public long getTimestamp() {
+		return timestamp;
+	}
+
+	/**
+	 * Query if the message can be added.
+	 */
+	public boolean canAdd(Message<?> message) {
+		return !isMember(message);
+	}
+
+	/**
+	 * Add this {@link org.springframework.integration.Message} to the
+	 * {@link org.springframework.integration.store.MessageGroup}, delegating in this case to the {@link #unmarked} field
+	 *
+	 * @param message the {@link org.springframework.integration.Message} you are adding to the {@link java.util.Map}
+	 */
+	public void add(Message<?> message) {
+		if (isMember(message)) {
+			return;
+		}
+
+		String unmarkedKey = this.unmarkedKey(message);
+		this.unmarked.put(unmarkedKey, (Message<?>) message);
+	}
+
+	/**
+	 * the only reason we differentiate the keys is so that conceptually you could use the <em>same</em> {@link java.util.Map} instance for both <em>marked</em> and <em>unmarked</em> messages.
+	 *
+	 * This method simply differentiates the key, building on {@link #baseKey(org.springframework.integration.Message)}'s return value
+	 *
+	 * @param msg the {@link org.springframework.integration.Message} from which the key should be generated.
+	 * @return a String to be used as a key
+	 */
+	protected String markedKey(Message<?> msg) {
+		return baseKey(msg) + "-m";
+	}
+
+	/**
+	 * the only reason we differentiate the keys is so that conceptually you could use the <em>same</em> {@link java.util.Map} instance for both <em>marked</em> and <em>unmarked</em> messages.
+	 *
+	 * This method simply differentiates the key, building on {@link #baseKey(org.springframework.integration.Message)}'s return value
+	 *
+	 * @param msg the {@link org.springframework.integration.Message} from which the key should be generated.
+	 * @return a String to be used as a key
+	 */
+	protected String unmarkedKey(Message<?> msg) {
+		return baseKey(msg) + "-u";
+	}
+
+	/**
+	 * Removes this {@link org.springframework.integration.Message} from this {@link org.springframework.integration.store.MessageGroup}'s memory
+	 *
+	 * @param message the message to remove
+	 */
+	public void remove(Message<?> message) {
+		if (unmarked.containsValue(message)) {
+			unmarked.remove(unmarkedKey(message));
+		}
+
+		if (marked.containsValue(message)) {
+			marked.remove(markedKey(message));
+		}
+	}
+
+	/**
+	 * the groupKey is based on the groupID and it sits at the beginning of all the keys for this {@link org.springframework.integration.store.MessageGroup}s keys
+	 *
+	 * @return a string based on {@link #getGroupId()}
+	 */
+	protected String groupKey() {
+		return (getGroupId()).toString();
+	}
+
+	protected String baseKey(Message<?> msg) {
+		String groupKey = groupKey();
+		UUID id = msg.getHeaders().getId();
+		Integer sn = msg.getHeaders().getSequenceNumber();
+		Integer ss = msg.getHeaders().getSequenceSize();
+
+		return String.format("%s-%s-%s-%s", groupKey, id.toString(),
+				sn.toString(), ss.toString());
+	}
+
+	public Collection<Message<?>> getUnmarked() {
+		return getMessagesForMessageGroup(this.unmarked);
+	}
+
+	/**
+	 * this method will be used to discover all the messages for a given group in a {@link com.gemstone.gemfire.cache.Region}
+	 *
+	 * @param region the region from which we're hoping to discover these {@link org.springframework.integration.Message}s
+	 * @return a collection of messages
+	 */
+	protected Collection<Message<?>> getMessagesForMessageGroup(
+			Map<String, Message<?>> region) {
+		try {
+			String groupMsgKey = groupKey();
+			Collection<Message<?>> msgs = new ArrayList<Message<?>>();
+
+			for (String k : region.keySet()) {
+				if (k.startsWith(groupMsgKey)) {
+					msgs.add(region.get(k));
+				}
+			}
+
+			return msgs;
+		} catch (Throwable th) {
+			throw new RuntimeException(th);
+		}
+	}
+
+	public Collection<Message<?>> getMarked() {
+		return getMessagesForMessageGroup(this.marked);
+	}
+
+	/**
+	 * @return the key that links these messages together
+	 */
+	public Object getGroupId() {
+		return groupId;
+	}
+
+	/**
+	 * @return true if the group is complete (i.e. no more messages are expected to be added)
+	 */
+	public boolean isComplete() {
+		if (size() == 0) {
+			return true;
+		}
+
+		int sequenceSize = getSequenceSize();
+
+		return (sequenceSize > 0) && (sequenceSize == size());
+	}
+
+	public int getSequenceSize() {
+		if (size() == 0) {
+			return 0;
+		}
+
+		return getOne().getHeaders().getSequenceSize();
+	}
+
+	/**
+	 * Mark the given message in this group. If the message is not part of this group then this call has no effect.
+	 *
+	 * @param messageToMark the message that should be marked
+	 */
+	public void mark(Message<?> messageToMark) {
+		if (this.unmarked.containsValue(messageToMark)) {
+			this.unmarked.remove(baseKey(messageToMark));
+		}
+
+		this.marked.put(baseKey(messageToMark), messageToMark);
+	}
+
+	public void markAll() {
+		for (Message<?> msg : getUnmarked())
+			mark(msg);
+	}
+
+	/**
+	 * @return the total number of messages (marked and unmarked) in this group
+	 */
+	public int size() {
+		return getMarked().size() + getUnmarked().size();
+	}
+
+	/**
+	 * @return a single message from the group
+	 */
+	public Message<?> getOne() {
+		if (!this.unmarked.isEmpty()) {
+			String aKey = this.unmarked.keySet().iterator().next();
+
+			return this.unmarked.get(aKey);
+		}
+
+		return null;
+	}
+
+	/**
+	 * This method determines whether messages have been added to this group that supersede the given message based on
+	 * its sequence id. This can be helpful to avoid ending up with sequences larger than their required sequence size
+	 * or sequences that are missing certain sequence numbers.
+	 *
+	 * @param message the message to test for candidacy
+	 *
+	 * @return whether or not the message is a member of the group
+	 *
+	 */
+	protected boolean isMember(Message<?> message) {
+		if (size() == 0) {
+			return false;
+		}
+
+		Integer messageSequenceNumber = message.getHeaders().getSequenceNumber();
+
+		if ((messageSequenceNumber != null) && (messageSequenceNumber > 0)) {
+			Integer messageSequenceSize = message.getHeaders().getSequenceSize();
+
+			if (!messageSequenceSize.equals(getSequenceSize())) {
+				return true;
+			} else {
+				if (containsSequenceNumber(getUnmarked(), messageSequenceNumber) ||
+						containsSequenceNumber(getUnmarked(),
+								messageSequenceNumber)) {
+					return true;
+				}
+			}
+		}
+
+		return false;
+	}
+
+	protected boolean containsSequenceNumber(Collection<Message<?>> messages,
+											 Integer messageSequenceNumber) {
+		for (Message<?> member : messages) {
+			Integer memberSequenceNumber = member.getHeaders()
+					.getSequenceNumber();
+
+			if (messageSequenceNumber.equals(memberSequenceNumber)) {
+				return true;
+			}
+		}
+
+		return false;
+	}
+}

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

@@ -0,0 +1,155 @@
+/*
+ * 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;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.store.AbstractMessageGroupStore;
+import org.springframework.integration.store.MessageGroup;
+import org.springframework.util.Assert;
+
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * Provides an implementation of {@link org.springframework.integration.store.MessageGroupStore} that delegates to a backend Gemfire instance.
+ * Gemfire holds keys and values. This class provides a strategy to hold objects.
+ *
+ * @author Josh Long
+ * @since 2.1
+ */
+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}
+	 */
+	protected Map<Object, KeyValueMessageGroup> groupIdToMessageGroup;
+
+	/**
+	 * Required {@link com.gemstone.gemfire.cache.Region} to manage the #unmarked data
+	 */
+	protected Map<String, Message<?>> unmarked;
+
+	/**
+	 * Required {@link com.gemstone.gemfire.cache.Region} to manage the #marked data
+	 */
+	protected Map<String, Message<?>> marked;
+
+
+	/**
+	 * Create a KeyValueMessageGroupStore with two backing regions to handle the state management.
+	 *
+	 * @param groupIdToMessageGroup the region to associate
+	 * @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) {
+		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;
+	}
+
+
+	public MessageGroup getMessageGroup(Object groupId) {
+		Assert.notNull(groupId, "'groupId' must not be null");
+		return this.getMessageGroupInternal(groupId);
+	}
+
+	public MessageGroup addMessageToGroup(Object groupId, Message<?> message) {
+		KeyValueMessageGroup group = getMessageGroupInternal(groupId);
+		group.add(message);
+		return group;
+	}
+
+	public MessageGroup markMessageGroup(MessageGroup group) {
+		Object groupId = group.getGroupId();
+		KeyValueMessageGroup internal = getMessageGroupInternal(groupId);
+		internal.markAll();
+		return internal;
+	}
+
+	public void removeMessageGroup(Object groupId) {
+		groupIdToMessageGroup.remove(groupId);
+	}
+
+	public MessageGroup removeMessageFromGroup(Object key, Message<?> messageToRemove) {
+		KeyValueMessageGroup group = getMessageGroupInternal(key);
+		group.remove(messageToRemove);
+		return group;
+	}
+
+	public MessageGroup markMessageFromGroup(Object key, Message<?> messageToMark) {
+		KeyValueMessageGroup group = getMessageGroupInternal(key);
+		group.mark(messageToMark);
+		return group;
+	}
+
+	@Override
+	public Iterator<MessageGroup> iterator() {
+		return new HashSet<MessageGroup>(groupIdToMessageGroup.values()).iterator();
+	}
+
+	protected KeyValueMessageGroup ensureMessageGroupHasReferencesToRegions(KeyValueMessageGroup keyValueMessageGroup) {
+		if (keyValueMessageGroup == null) {
+			return null;
+		}
+		keyValueMessageGroup.setMarked(this.marked);
+		keyValueMessageGroup.setUnmarked(this.unmarked);
+		return keyValueMessageGroup;
+	}
+
+	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)) {
+			groupIdToMessageGroup.put(groupId, new KeyValueMessageGroup(groupId));
+		}
+		return ensureMessageGroupHasReferencesToRegions(groupIdToMessageGroup.get( groupId));
+	}
+
+}