diff --git a/docs/src/reference/docbook/gemfire.xml b/docs/src/reference/docbook/gemfire.xml new file mode 100644 index 0000000000..f451e5719f --- /dev/null +++ b/docs/src/reference/docbook/gemfire.xml @@ -0,0 +1,129 @@ + + + GemFire Support + + Spring Integration provides support for VMWare vFabric GemFire + +
+ Introduction + + VMWare vFabric GemFire (GemFire) is a distributed data management platform providing a key-value data grid along with advanced distributed system features such as event processing, continuous querying, and + remote function execution. This guide assumes + some familiarity with GemFire + and its API. + + + Spring integration provides support for GemFire by providing inbound adapters for entry and continuous query events, + and an outbound adapter to write entries to the cache. Spring integration leverages the + Spring Gemfire project, providing a thin + wrapper over its components. + + + To configure the 'int-gfe' namespace, include the following elements within the headers of your XML configuration file: + + + +
+
+ Inbound Channel Adapter + + The inbound-channel-adapter produces messages on a channel triggered by a GemFire EntryEvent. GemFire + generates events whenever an entry is CREATED, UPDATED, DESTROYED, or INVALIDATED in the associated region. The inbound channel adapter allows you to filter on a subset of these + events. For example, you may want to only produce messages in response to an entry being CREATED. In addition, the inbound channel adapter can evaluate a SpEL + expression if, for example, you want your message payload to contain an event property such as the new entry value. + + + + + ]]> + + + In the above configuration, we are creating a GemFire Cache and Region using Spring GemFire's 'gfe' namespace. + The inbound-channel-adapter requires a reference to the GemFire region for which the adapter will be listening for events. Optional attributes include cache-events + which can contain a comma separated list of event types for which a message will be produced on the input channel. By default all event types are enabled. + Note that this adapter conforms to Spring integration conventions. + If no channel attribute is provided, the channel will be created from the id attribute. This adapter also supports an error-channel. + If expression is not provided the message payload will be a GemFire EntryEvent + +
+
+ Continuous Query Inbound Channel Adapter + + The cq-inbound-channel-adapter produces messages a channel triggered by a GemFire continuous query or CqEvent event. Spring GemFire introduced + continuous query support in release 1.1, including a QueryListenerContainer which provides a nice abstraction over the GemFire native API. This adapter requires a + reference to a QueryListenerContainer, and creates a listener for a given query and executes the query. The continuous query acts as an event source that will fire whenever its + result set changes state. + + GemFire queries are written in OQL and are scoped to the entire cache (not just one region). Additionally, continuous queries require a remote (i.e., running in a separate process or remote host) + cache server. Please consult the GemFire documentation for more information on + implementing continuous queries. + + + + + + + + + + + + + + + + + + ]]> + + + In the above configuration, we are creating a GemFire client cache + (recall a cache server is required for this implementation and its address is configured as a sub-element of the pool), a client region and a QueryListenerContainer + using Spring GemFire. The continuous query inbound channel adapter requires a query-listener-container attribute which contains a reference to the QueryListenerContainer. Optionally, + it accepts an expression attribute which uses SpEL to transform the CqEvent or extract an individual property as needed. The cq-inbound-channel-adapter also supports a + query-events attribute, containing a comma separated list of event types for which a message will be produced on the input channel (all events are enabled by default), query-name which provides an optional query name, and + expression which works as described in the above section. + Note that this adapter conforms to Spring integration conventions. + If no channel attribute is provided, the channel will be created from the id attribute. This adapter also supports an error-channel + +
+ +
+ Outbound Channel Adapter + + The outbound-channel-adapter writes cache entries mapped from the message payload. In its simplest form, it expects a + payload of type java.util.Map and puts the map entries into its configured region. + + + ]]> + + + + Given the above configuration, an exception will be thrown if the payload is not a Map. Additionally, the outbound channel adapter can be configured to create a + map of cache entries using SpEL of course. + + + + + + + + + ]]> + + In the above configuration, the inner element cache-entries is semantically equivalent to Spring 'map' element. The adapter interprets the key and + value attributes as SpEL expressions with the message as the evaluation context. Note that this contain + arbitrary cache entries (not only those derived from the message) and that literal values must be enclosed in single quotes. In the above example, if the message sent to + cacheChannel has a String payload with a value "Hello", two entries [HELLO:hello, foo:bar] will be written (created or updated) in the cache region. + This adapter also supports the order attribute which may be useful if it is bound to a PublishSubscribeChannel. + +
+ diff --git a/docs/src/reference/docbook/index.xml b/docs/src/reference/docbook/index.xml index d727641856..2b678aa1c0 100644 --- a/docs/src/reference/docbook/index.xml +++ b/docs/src/reference/docbook/index.xml @@ -135,7 +135,8 @@ - + + Appendices diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java index 8d7296e8c0..3cbf6c97e2 100644 --- a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CacheListeningMessageProducer.java @@ -22,10 +22,6 @@ 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; @@ -47,7 +43,7 @@ import com.gemstone.gemfire.cache.util.CacheListenerAdapter; * @since 2.1 */ @SuppressWarnings({"rawtypes", "unchecked"}) -public class CacheListeningMessageProducer extends MessageProducerSupport { +public class CacheListeningMessageProducer extends SpelMessageProducerSupport { private final Log logger = LogFactory.getLog(this.getClass()); @@ -58,9 +54,6 @@ public class CacheListeningMessageProducer extends MessageProducerSupport { private volatile Set supportedEventTypes = new HashSet(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"); @@ -74,14 +67,6 @@ public class CacheListeningMessageProducer extends MessageProducerSupport { this.supportedEventTypes = new HashSet(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() { @@ -105,8 +90,7 @@ public class CacheListeningMessageProducer extends MessageProducerSupport { } } - - + private class MessageProducingCacheListener extends CacheListenerAdapter { @Override @@ -137,14 +121,9 @@ public class CacheListeningMessageProducer extends MessageProducerSupport { } } - private void processEvent(EntryEvent event) { - if (payloadExpression != null) { - Object evaluationResult = payloadExpression.getValue(event); - this.publish(evaluationResult); - } - else { - this.publish(event); - } + private void processEvent(EntryEvent event) { + this.publish(evaluationResult(event)); + } private void publish(Object payload) { 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 a16b38f26b..b5727ea112 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,15 +16,19 @@ 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.data.gemfire.listener.CqQueryDefinition; import org.springframework.data.gemfire.listener.QueryListener; import org.springframework.data.gemfire.listener.QueryListenerContainer; import org.springframework.integration.Message; -import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.support.MessageBuilder; import org.springframework.util.Assert; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; + import com.gemstone.gemfire.cache.query.CqEvent; /** @@ -38,17 +42,24 @@ import com.gemstone.gemfire.cache.query.CqEvent; * @since 2.1 * */ -public class ContinuousQueryMessageProducer extends MessageProducerSupport implements QueryListener { +public class ContinuousQueryMessageProducer extends SpelMessageProducerSupport implements QueryListener { private static Log logger = LogFactory.getLog(ContinuousQueryMessageProducer.class); - + private final String query; + private final QueryListenerContainer queryListenerContainer; + private volatile String queryName; + private boolean durable; - + + private volatile Set supportedEventTypes = new HashSet(Arrays.asList(CqEventType.CREATED, + CqEventType.UPDATED)); + /** * - * @param queryListenerContainer a {@link org.springframework.data.gemfire.listener.QueryListenerContainer} + * @param queryListenerContainer a + * {@link org.springframework.data.gemfire.listener.QueryListenerContainer} * @param query the query string */ public ContinuousQueryMessageProducer(QueryListenerContainer queryListenerContainer, String query) { @@ -57,7 +68,7 @@ public class ContinuousQueryMessageProducer extends MessageProducerSupport imple this.queryListenerContainer = queryListenerContainer; this.query = query; } - + /** * * @param queryName optional query name @@ -65,7 +76,7 @@ public class ContinuousQueryMessageProducer extends MessageProducerSupport imple public void setQueryName(String queryName) { this.queryName = queryName; } - + /** * * @param durable true if the query is a durable subscription @@ -74,16 +85,22 @@ public class ContinuousQueryMessageProducer extends MessageProducerSupport imple this.durable = durable; } + public void setSupportedEventTypes(CqEventType... eventTypes) { + Assert.notEmpty(eventTypes, "eventTypes must not be empty"); + this.supportedEventTypes = new HashSet(Arrays.asList(eventTypes)); + } + @Override protected void onInit() { super.onInit(); - if (queryName == null){ + if (queryName == null) { queryListenerContainer.addListener(new CqQueryDefinition(this.query, this, this.durable)); - } else { + } + else { queryListenerContainer.addListener(new CqQueryDefinition(this.queryName, this.query, this, this.durable)); } } - + /* * (non-Javadoc) * @@ -92,11 +109,24 @@ public class ContinuousQueryMessageProducer extends MessageProducerSupport imple * .gemfire.cache.query.CqEvent) */ public void onEvent(CqEvent event) { - if (logger.isDebugEnabled()){ - logger.debug(String.format("processing cq event key [%s] event [%s]",event.getBaseOperation().toString(),event.getKey())); + if (isEventSupported(event)) { + if (logger.isDebugEnabled()) { + logger.debug(String.format("processing cq event key [%s] event [%s]", event.getBaseOperation() + .toString(), event.getKey())); + } + Message cqEventMessage = MessageBuilder.withPayload(evaluationResult(event)).build(); + sendMessage(cqEventMessage); } - Message cqEventMessage = MessageBuilder.withPayload(event).build(); - sendMessage(cqEventMessage); + } + + /** + * @param event + * @return + */ + private boolean isEventSupported(CqEvent event) { + String eventName = event.getBaseOperation().toString()+"D"; + CqEventType eventType = CqEventType.valueOf(eventName); + return supportedEventTypes.contains(eventType); } } \ No newline at end of file diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CqEventType.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CqEventType.java new file mode 100644 index 0000000000..1401879371 --- /dev/null +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/CqEventType.java @@ -0,0 +1,30 @@ +/* + * 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 Continuous Query Event Types + * @author David Turanski + * @since 2.1 + */ +public enum CqEventType { + CREATED, + + UPDATED, + + DESTROYED, + + REGION_CLEARED, + + REGION_INVALIDATED +} diff --git a/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/SpelMessageProducerSupport.java b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/SpelMessageProducerSupport.java new file mode 100644 index 0000000000..0b662d7dbd --- /dev/null +++ b/spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/inbound/SpelMessageProducerSupport.java @@ -0,0 +1,54 @@ +/* + * 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.springframework.expression.Expression; +import org.springframework.expression.spel.standard.SpelExpressionParser; +import org.springframework.integration.endpoint.MessageProducerSupport; + +/** + * @author David Turanski + * @since 2.1 + * + */ +abstract class SpelMessageProducerSupport extends MessageProducerSupport { + + private volatile Expression payloadExpression; + + private final SpelExpressionParser parser = new SpelExpressionParser(); + + + @Override + protected void onInit(){ + super.onInit(); + } + + public void setPayloadExpression(String payloadExpression) { + if (payloadExpression == null) { + this.payloadExpression = null; + } + else { + this.payloadExpression = this.parser.parseExpression(payloadExpression); + } + } + + protected Object evaluationResult(Object payload){ + Object evaluationResult = payload; + if (payloadExpression != null) { + evaluationResult = payloadExpression.getValue(payload); + } + return evaluationResult; + } + + +} diff --git a/spring-integration-gemfire/src/main/resources/META-INF/spring.schemas b/spring-integration-gemfire/src/main/resources/META-INF/spring.schemas index fc4b7ef1d9..b542bdfd8c 100644 --- a/spring-integration-gemfire/src/main/resources/META-INF/spring.schemas +++ b/spring-integration-gemfire/src/main/resources/META-INF/spring.schemas @@ -1,2 +1,2 @@ -http\://www.springframework.org/schema/integration/scripting/spring-integration-gemfire.xsd=org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-2.1.xsd -http\://www.springframework.org/schema/integration/scripting/spring-integration-gemfire-2.1.xsd=org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-2.1.xsd +http\://www.springframework.org/schema/integration/gemfire/spring-integration-gemfire.xsd=org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-2.1.xsd +http\://www.springframework.org/schema/integration/gemfire/spring-integration-gemfire-2.1.xsd=org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-2.1.xsd diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml index d31c297649..8439b77a53 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml @@ -2,9 +2,9 @@ diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java index 5201d86214..003bac50be 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests.java @@ -23,7 +23,6 @@ import org.springframework.integration.MessagingException; import org.springframework.integration.core.MessageHandler; import org.springframework.integration.core.SubscribableChannel; import org.springframework.integration.message.ErrorMessage; -import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests-context.xml index 0b5c54e67a..60665d582b 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests-context.xml @@ -3,9 +3,9 @@ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:gfe="http://www.springframework.org/schema/gemfire" xmlns:int="http://www.springframework.org/schema/integration" - xmlns:int-gfe="http://www.springframework.org/schema/integration/gemfire" xmlns:util="http://www.springframework.org/schema/util" - xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire http://www.springframework.org/schema/integration/scripting/spring-integration-gemfire.xsd + xmlns:int-gfe="http://www.springframework.org/schema/integration/gemfire" + xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire http://www.springframework.org/schema/integration/gemfire/spring-integration-gemfire.xsd http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire-1.1.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd @@ -23,14 +23,26 @@ - + - + - + + + + + + + + + + + + + diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests.java b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests.java index 7e6640fa1e..c87de95792 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests.java +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests.java @@ -12,6 +12,7 @@ */ package org.springframework.integration.gemfire.inbound.cq; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertTrue; @@ -19,6 +20,7 @@ import java.io.IOException; import java.io.OutputStream; import org.junit.AfterClass; +import org.junit.Before; import org.junit.BeforeClass; import org.junit.Test; import org.junit.runner.RunWith; @@ -41,6 +43,8 @@ import com.gemstone.gemfire.internal.cache.LocalRegion; @ContextConfiguration public class ContinuousQueryMessageProducerTests { + static ConfigurableApplicationContext staticCtx; + @Autowired LocalRegion region; @@ -48,7 +52,10 @@ public class ContinuousQueryMessageProducerTests { ConfigurableApplicationContext applicationContext; @Autowired - PollableChannel outputChannel; + PollableChannel outputChannel1; + + @Autowired + PollableChannel outputChannel2; static OutputStream os; @BeforeClass @@ -56,30 +63,36 @@ public class ContinuousQueryMessageProducerTests { os = ForkUtil.cacheServer(); } - + + @Before + public void setUp() { + staticCtx = applicationContext; + } @Test - public void test() throws InterruptedException { + public void testCqEvent() throws InterruptedException { region.put("one",1); - Message msg = outputChannel.receive(1000); + Message msg = outputChannel1.receive(1000); assertNotNull(msg); assertTrue(msg.getPayload() instanceof CqEvent); - /* - * Avoid shutdown errors - */ - applicationContext.close(); + } + + @Test + public void testPayloadExpression() throws InterruptedException { + region.put("one",1); + Message msg = outputChannel2.receive(1000); + assertNotNull(msg); + assertEquals(1,msg.getPayload()); + + } @AfterClass public static void cleanUp() { - - try { - Thread.sleep(3000); - } - catch (InterruptedException e) { - // TODO Auto-generated catch block - e.printStackTrace(); - } + /* + * Avoid shutdown errors + */ + staticCtx.close(); sendSignal(); } diff --git a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml index 1d4f990a42..e1723add50 100644 --- a/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml +++ b/spring-integration-gemfire/src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml @@ -4,7 +4,7 @@ xmlns:int-gfe="http://www.springframework.org/schema/integration/gemfire" xmlns:int="http://www.springframework.org/schema/integration" xmlns:gfe="http://www.springframework.org/schema/gemfire" - xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire http://www.springframework.org/schema/integration/scripting/spring-integration-gemfire.xsd + xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire http://www.springframework.org/schema/integration/gemfire/spring-integration-gemfire.xsd http://www.springframework.org/schema/gemfire http://www.springframework.org/schema/gemfire/spring-gemfire-1.1.xsd http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">