INT-2075 - Added namespace support for cq-inbound-channel-adapter. Also some general cleanup and additional tests
This commit is contained in:
David Turanski
@@ -16,9 +16,9 @@
|
||||
</para>
|
||||
<para>
|
||||
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
|
||||
<ulink url="http://www.springsource.org/spring-gemfire">Spring Gemfire</ulink> project, providing a thin
|
||||
wrapper over its components.
|
||||
an outbound adapter to write entries to the cache, and <classname>MessageStore</classname> and <classname>MessageGroupStore</classname> implementations.
|
||||
Spring integration leverages the
|
||||
<ulink url="http://www.springsource.org/spring-gemfire">Spring Gemfire</ulink> project, providing a thin wrapper over its components.
|
||||
</para>
|
||||
<para>
|
||||
To configure the 'int-gfe' namespace, include the following elements within the headers of your XML configuration file:
|
||||
@@ -45,7 +45,7 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire
|
||||
|
||||
In the above configuration, we are creating a GemFire <classname>Cache</classname> and <classname>Region</classname> 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 <code>cache-events</code>
|
||||
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.
|
||||
which can contain a comma separated list of event types for which a message will be produced on the input channel. By default CREATED and UPDATED are enabled.
|
||||
Note that this adapter conforms to Spring integration conventions.
|
||||
If no <code>channel</code> attribute is provided, the channel will be created from the <code>id</code> attribute. This adapter also supports an <code>error-channel</code>.
|
||||
If <code>expression</code> is not provided the message payload will be a GemFire <classname>EntryEvent</classname>
|
||||
@@ -88,9 +88,10 @@ xsi:schemaLocation="http://www.springframework.org/schema/integration/gemfire
|
||||
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 <classname>QueryListenerContainer</classname>
|
||||
using Spring GemFire. The continuous query inbound channel adapter requires a <code>query-listener-container</code> attribute which contains a reference to the <classname>QueryListenerContainer</classname>. Optionally,
|
||||
it accepts an <code>expression</code> attribute which uses SpEL to transform the <code>CqEvent</code> or extract an individual property as needed. The cq-inbound-channel-adapter also supports a
|
||||
<code>query-events</code> 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), <code>query-name</code> which provides an optional query name, and
|
||||
<code>expression</code> which works as described in the above section.
|
||||
it accepts an <code>expression</code> attribute which uses SpEL to transform the <code>CqEvent</code> or extract an individual property as needed. The cq-inbound-channel-adapter provides a
|
||||
<code>query-events</code> attribute, containing a comma separated list of event types for which a message will be produced on the input channel. Available event types are CREATED, UPDATED, DESTROYED,
|
||||
REGION_DESTROYED, REGION_INVALIDATED. CREATED and UPDATED are enabled by default. Additional optional attributes include, <code>query-name</code> which provides an optional query name, and
|
||||
<code>expression</code> which works as described in the above section, and <code>durable</code> - a boolean value indicating if the query is durable (false by default).
|
||||
Note that this adapter conforms to Spring integration conventions.
|
||||
If no <code>channel</code> attribute is provided, the channel will be created from the <code>id</code> attribute. This adapter also supports an <code>error-channel</code>
|
||||
</para>
|
||||
|
||||
@@ -9,7 +9,6 @@
|
||||
<configs>
|
||||
<config>src/test/java/org/springframework/integration/gemfire/inbound/GemfireInboundChannelAdapterTests-context.xml</config>
|
||||
<config>src/test/java/org/springframework/integration/gemfire/outbound/GemfireOutboundChannelAdapterTests-context.xml</config>
|
||||
<config>src/test/java/org/springframework/integration/gemfire/inbound/cq/ContinuousQueryMessageProducerTests-context.xml</config>
|
||||
</configs>
|
||||
<configSets>
|
||||
</configSets>
|
||||
|
||||
@@ -0,0 +1,82 @@
|
||||
/*
|
||||
* 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.config.xml;
|
||||
|
||||
import org.springframework.beans.factory.support.AbstractBeanDefinition;
|
||||
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
|
||||
import org.springframework.beans.factory.xml.ParserContext;
|
||||
import org.springframework.integration.config.xml.AbstractChannelAdapterParser;
|
||||
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
|
||||
import org.springframework.util.StringUtils;
|
||||
import org.w3c.dom.Element;
|
||||
|
||||
/**
|
||||
* @author David Turanski
|
||||
* @since 2.1
|
||||
*
|
||||
*/
|
||||
public class GemfireCqInboundChannelAdapterParser extends AbstractChannelAdapterParser {
|
||||
|
||||
|
||||
private static final String ERROR_CHANNEL_ATTRIBUTE = "error-channel";
|
||||
|
||||
private static final String OUTPUT_CHANNEL_PROPERTY = "outputChannel";
|
||||
|
||||
private static final String QUERY_LISTENER_CONTAINER_ATTRIBUTE = "query-listener-container";
|
||||
|
||||
private static final String DURABLE_ATTRIBUTE = "durable";
|
||||
|
||||
private static final String QUERY_NAME_ATTRIBUTE = "query-name";
|
||||
|
||||
private static final String QUERY_ATTRIBUTE = "query";
|
||||
|
||||
private static final String PAYLOAD_EXPRESSION_PROPERTY = "payloadExpression";
|
||||
|
||||
private static final String EXPRESSION_ATTRIBUTE = "expression";
|
||||
|
||||
private static final String GEMFIRE_INBOUND_CONTINUOUS_QUERY_MESSAGE_PRODUCER = "org.springframework.integration.gemfire.inbound.ContinuousQueryMessageProducer";
|
||||
|
||||
private static final String SUPPORTED_EVENT_TYPES_PROPERTY = "supportedEventTypes";
|
||||
|
||||
private static final String QUERY_EVENTS_ATTRIBUTE = "query-events";
|
||||
|
||||
/* (non-Javadoc)
|
||||
* @see org.springframework.integration.config.xml.AbstractChannelAdapterParser#doParse(org.w3c.dom.Element, org.springframework.beans.factory.xml.ParserContext, java.lang.String)
|
||||
*/
|
||||
@Override
|
||||
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
|
||||
BeanDefinitionBuilder continuousQueryMesageProducer = BeanDefinitionBuilder.genericBeanDefinition(GEMFIRE_INBOUND_CONTINUOUS_QUERY_MESSAGE_PRODUCER);
|
||||
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element, EXPRESSION_ATTRIBUTE,PAYLOAD_EXPRESSION_PROPERTY);
|
||||
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element, QUERY_EVENTS_ATTRIBUTE, SUPPORTED_EVENT_TYPES_PROPERTY);
|
||||
|
||||
if (!StringUtils.hasText(QUERY_LISTENER_CONTAINER_ATTRIBUTE)){
|
||||
parserContext.getReaderContext().error("'query-listener-container' attribute is required.",element);
|
||||
}
|
||||
|
||||
if (!StringUtils.hasText(QUERY_ATTRIBUTE)){
|
||||
parserContext.getReaderContext().error("'query' attribute is required.",element);
|
||||
}
|
||||
|
||||
continuousQueryMesageProducer.addConstructorArgReference(element.getAttribute(QUERY_LISTENER_CONTAINER_ATTRIBUTE));
|
||||
continuousQueryMesageProducer.addConstructorArgValue(element.getAttribute(QUERY_ATTRIBUTE));
|
||||
|
||||
continuousQueryMesageProducer.addPropertyReference(OUTPUT_CHANNEL_PROPERTY, channelName);
|
||||
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(continuousQueryMesageProducer, element, ERROR_CHANNEL_ATTRIBUTE);
|
||||
|
||||
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element, QUERY_NAME_ATTRIBUTE);
|
||||
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element, DURABLE_ATTRIBUTE);
|
||||
|
||||
return continuousQueryMesageProducer.getBeanDefinition();
|
||||
}
|
||||
|
||||
}
|
||||
@@ -16,7 +16,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
|
||||
|
||||
/**
|
||||
* @author David Turanski
|
||||
*
|
||||
* @since 2.1
|
||||
*/
|
||||
public class GemfireIntegrationNamespaceHandler extends AbstractIntegrationNamespaceHandler{
|
||||
|
||||
@@ -25,8 +25,8 @@ public class GemfireIntegrationNamespaceHandler extends AbstractIntegrationNames
|
||||
*/
|
||||
public void init() {
|
||||
registerBeanDefinitionParser("inbound-channel-adapter", new GemfireInboundChannelAdapterParser());
|
||||
registerBeanDefinitionParser("cq-inbound-channel-adapter", new GemfireCqInboundChannelAdapterParser());
|
||||
registerBeanDefinitionParser("outbound-channel-adapter", new GemfireOutboundChannelAdapterParser());
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -111,7 +111,7 @@ public class ContinuousQueryMessageProducer extends SpelMessageProducerSupport i
|
||||
public void onEvent(CqEvent event) {
|
||||
if (isEventSupported(event)) {
|
||||
if (logger.isDebugEnabled()) {
|
||||
logger.debug(String.format("processing cq event key [%s] event [%s]", event.getBaseOperation()
|
||||
logger.debug(String.format("processing cq event key [%s] event [%s]", event.getQueryOperation()
|
||||
.toString(), event.getKey()));
|
||||
}
|
||||
Message<?> cqEventMessage = MessageBuilder.withPayload(evaluationResult(event)).build();
|
||||
@@ -124,7 +124,9 @@ public class ContinuousQueryMessageProducer extends SpelMessageProducerSupport i
|
||||
* @return
|
||||
*/
|
||||
private boolean isEventSupported(CqEvent event) {
|
||||
String eventName = event.getBaseOperation().toString()+"D";
|
||||
|
||||
String eventName = event.getQueryOperation().toString() +
|
||||
(event.getQueryOperation().toString().endsWith("Y")? "ED" : "D");
|
||||
CqEventType eventType = CqEventType.valueOf(eventName);
|
||||
return supportedEventTypes.contains(eventType);
|
||||
}
|
||||
|
||||
@@ -6,14 +6,14 @@
|
||||
targetNamespace="http://www.springframework.org/schema/integration/gemfire"
|
||||
elementFormDefault="qualified" attributeFormDefault="unqualified">
|
||||
|
||||
<xsd:import namespace="http://www.springframework.org/schema/beans"
|
||||
schemaLocation="http://www.springframework.org/schema/beans/spring-beans-3.1.xsd"/>
|
||||
<xsd:import namespace="http://www.springframework.org/schema/beans"
|
||||
schemaLocation="http://www.springframework.org/schema/beans/spring-beans-3.1.xsd" />
|
||||
<xsd:import namespace="http://www.springframework.org/schema/tool" />
|
||||
<xsd:import namespace="http://www.springframework.org/schema/integration"
|
||||
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-2.1.xsd" />
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
Defines the core configuration elements for Spring Integration Gemfire Support.
|
||||
Defines the core configuration elements for Spring Integration GemFire Support.
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
|
||||
@@ -21,88 +21,131 @@
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Configures an inbound Channel Adapter backed by a
|
||||
Gemfire CacheListener
|
||||
GemFire CacheListener
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
|
||||
<xsd:complexType>
|
||||
<xsd:complexContent>
|
||||
<xsd:extension base="InboundChannelAdapterType">
|
||||
|
||||
<xsd:attribute name="region" type="xsd:string" use="required">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type type="com.gemstone.gemfire.cache.Region" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
<xsd:attribute name="cache-events" type="xsd:string"
|
||||
use="optional" default="CREATED,UPDATED">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
Enabled cache entry event types
|
||||
]]></xsd:documentation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="value">
|
||||
<tool:expected-type
|
||||
type="org.springframework.integration.gemfire.inbound.EventType" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
</xsd:extension>
|
||||
</xsd:complexContent>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:element name="cq-inbound-channel-adapter">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Configures an inbound Channel Adapter backed by a
|
||||
Spring Gemfire QueryListener
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:complexType>
|
||||
<xsd:attribute name="id" type="xsd:string" use="optional" />
|
||||
<xsd:attribute name="channel" type="xsd:string" use="optional">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type
|
||||
type="org.springframework.integration.core.MessageChannel" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
<xsd:attribute name="error-channel" type="xsd:string"
|
||||
use="optional">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type
|
||||
type="org.springframework.integration.core.MessageChannel" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
<xsd:attribute name="region" type="xsd:string" use="required">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type type="com.gemstone.gemfire.cache.Region" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
<xsd:attribute name="expression" type="xsd:string" use="optional">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
Expression to be evaluated to produce a value for the payload.
|
||||
<xsd:complexContent>
|
||||
<xsd:extension base="InboundChannelAdapterType">
|
||||
<xsd:attribute name="query-listener-container" use="required">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
Reference to a QueryListenerContainer
|
||||
]]></xsd:documentation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="value">
|
||||
<tool:expected-type
|
||||
type="org.springframework.data.gemfire.listener.QueryListenerContainer" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="query-events" type="xsd:string"
|
||||
use="optional" default="CREATED,UPDATED">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
Enabled continuous query event types
|
||||
]]></xsd:documentation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="value">
|
||||
<tool:expected-type
|
||||
type="org.springframework.integration.gemfire.inbound.CqEventType" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="query" use="required" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
The query string
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="cache-events" type="xsd:string"
|
||||
use="optional" default="CREATED,UPDATED">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
Enabled cache event types
|
||||
</xsd:attribute>
|
||||
<xsd:attribute name="query-name" use="optional" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
The query name
|
||||
]]></xsd:documentation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="value">
|
||||
<tool:expected-type
|
||||
type="org.springframework.integration.gemfire.inbound.EventType" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
</xsd:attribute>
|
||||
|
||||
<xsd:attribute name="durable" use="optional" type="xsd:boolean" default="false">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
Indicates if the query is a durable subscription
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
</xsd:extension>
|
||||
</xsd:complexContent>
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
|
||||
<xsd:element name="outbound-channel-adapter">
|
||||
<xsd:complexType>
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
Configures an outbound Channel Adapter that writes Message payloads to a
|
||||
Configures an outbound Channel Adapter that
|
||||
writes Message payloads to a
|
||||
File.
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
<xsd:sequence>
|
||||
<xsd:element name="cache-entries" type="beans:mapType" minOccurs="0" maxOccurs="1">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
A map of SpEL expressions used to create cache entries. If not provided, payload must be a Map
|
||||
<xsd:element name="cache-entries" type="beans:mapType"
|
||||
minOccurs="0" maxOccurs="1">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation>
|
||||
A map of SpEL expressions used to create cache entries. If not
|
||||
provided, payload must be a Map
|
||||
</xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:annotation>
|
||||
</xsd:element>
|
||||
</xsd:sequence>
|
||||
<xsd:attribute name="id" type="xsd:string" use="optional" />
|
||||
|
||||
|
||||
|
||||
<xsd:attribute name="channel" type="xsd:string">
|
||||
<xsd:annotation>
|
||||
@@ -114,7 +157,7 @@
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
|
||||
<xsd:attribute name="region" type="xsd:string" use="required">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
@@ -124,7 +167,7 @@
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
|
||||
<xsd:attribute name="order" type="xsd:string" use="optional">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
@@ -133,8 +176,42 @@
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
|
||||
|
||||
</xsd:complexType>
|
||||
</xsd:element>
|
||||
|
||||
<xsd:complexType name="InboundChannelAdapterType">
|
||||
<xsd:attribute name="id" type="xsd:string" use="optional" />
|
||||
<xsd:attribute name="channel" type="xsd:string" use="optional">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type
|
||||
type="org.springframework.integration.core.MessageChannel" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
<xsd:attribute name="error-channel" type="xsd:string"
|
||||
use="optional">
|
||||
<xsd:annotation>
|
||||
<xsd:appinfo>
|
||||
<tool:annotation kind="ref">
|
||||
<tool:expected-type
|
||||
type="org.springframework.integration.core.MessageChannel" />
|
||||
</tool:annotation>
|
||||
</xsd:appinfo>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
<xsd:attribute name="expression" type="xsd:string" use="optional">
|
||||
<xsd:annotation>
|
||||
<xsd:documentation><![CDATA[
|
||||
Expression to be evaluated to produce a value for the payload.
|
||||
]]></xsd:documentation>
|
||||
</xsd:annotation>
|
||||
</xsd:attribute>
|
||||
|
||||
</xsd:complexType>
|
||||
</xsd:schema>
|
||||
@@ -0,0 +1,153 @@
|
||||
/*
|
||||
* 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 static org.junit.Assert.assertEquals;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.springframework.data.gemfire.listener.QueryListenerContainer;
|
||||
import org.springframework.integration.Message;
|
||||
import org.springframework.integration.MessagingException;
|
||||
import org.springframework.integration.channel.DirectChannel;
|
||||
import org.springframework.integration.core.MessageHandler;
|
||||
|
||||
import com.gemstone.gemfire.cache.Operation;
|
||||
import com.gemstone.gemfire.cache.query.CqEvent;
|
||||
import com.gemstone.gemfire.cache.query.CqQuery;
|
||||
import com.gemstone.gemfire.cache.query.internal.CqQueryImpl;
|
||||
|
||||
/**
|
||||
* @author David Turanski
|
||||
* @since 2.1
|
||||
*/
|
||||
public class ContinuousQueryMessageProducerTests {
|
||||
QueryListenerContainer queryListenerContainer;
|
||||
|
||||
ContinuousQueryMessageProducer cqMessageProducer;
|
||||
|
||||
CqMessageHandler handler;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
queryListenerContainer = mock(QueryListenerContainer.class);
|
||||
cqMessageProducer = new ContinuousQueryMessageProducer(queryListenerContainer, "");
|
||||
DirectChannel outputChannel = new DirectChannel();
|
||||
cqMessageProducer.setOutputChannel(outputChannel);
|
||||
handler = new CqMessageHandler();
|
||||
outputChannel.subscribe(handler);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessageProduced() {
|
||||
CqEvent cqEvent = event(Operation.CREATE, "hello");
|
||||
|
||||
cqMessageProducer.onEvent(cqEvent);
|
||||
|
||||
assertEquals(1, handler.count);
|
||||
assertEquals(cqEvent, handler.payload);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessageNotProducedForUnsupportedEventType() {
|
||||
CqEvent cqEvent = event(Operation.DESTROY, "hello");
|
||||
|
||||
cqMessageProducer.onEvent(cqEvent);
|
||||
|
||||
assertEquals(0, handler.count);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testMessageProducedForAddedEventType() {
|
||||
|
||||
CqEvent cqEvent = event(Operation.DESTROY, null);
|
||||
|
||||
cqMessageProducer.setSupportedEventTypes(CqEventType.DESTROYED);
|
||||
cqMessageProducer.onEvent(cqEvent);
|
||||
|
||||
assertEquals(1, handler.count);
|
||||
assertEquals(cqEvent, handler.payload);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPayloadExpression() {
|
||||
CqEvent cqEvent = event(Operation.CREATE, "hello");
|
||||
cqMessageProducer.setPayloadExpression("newValue.toUpperCase() + ', WORLD'");
|
||||
cqMessageProducer.onEvent(cqEvent);
|
||||
assertEquals(1, handler.count);
|
||||
assertEquals("HELLO, WORLD", handler.payload);
|
||||
}
|
||||
|
||||
CqEvent event(final Operation operation, final Object value) {
|
||||
|
||||
CqEvent event = new CqEvent() {
|
||||
|
||||
final CqQuery cq = new CqQueryImpl();
|
||||
|
||||
final byte[] ba = new byte[0];
|
||||
|
||||
final Object key = new Object();
|
||||
|
||||
final Exception ex = new Exception();
|
||||
|
||||
public Operation getBaseOperation() {
|
||||
return operation;
|
||||
}
|
||||
|
||||
public CqQuery getCq() {
|
||||
return cq;
|
||||
}
|
||||
|
||||
public byte[] getDeltaValue() {
|
||||
return ba;
|
||||
}
|
||||
|
||||
public Object getKey() {
|
||||
return key;
|
||||
}
|
||||
|
||||
public Object getNewValue() {
|
||||
return value;
|
||||
}
|
||||
|
||||
public Operation getQueryOperation() {
|
||||
return operation;
|
||||
}
|
||||
|
||||
public Throwable getThrowable() {
|
||||
return ex;
|
||||
}
|
||||
};
|
||||
|
||||
return event;
|
||||
}
|
||||
|
||||
private static class CqMessageHandler implements MessageHandler {
|
||||
public int count;
|
||||
|
||||
public Object payload;
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
*
|
||||
* @see
|
||||
* org.springframework.integration.core.MessageHandler#handleMessage
|
||||
* (org.springframework.integration.Message)
|
||||
*/
|
||||
public void handleMessage(Message<?> message) throws MessagingException {
|
||||
count++;
|
||||
payload = message.getPayload();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -23,24 +23,17 @@
|
||||
<property name="cache" ref="client-cache"/>
|
||||
</bean>
|
||||
|
||||
<bean id="cqMessageProducer" class="org.springframework.integration.gemfire.inbound.ContinuousQueryMessageProducer">
|
||||
<constructor-arg ref="queryListenerContainer"/>
|
||||
<constructor-arg value="select * from /test"/>
|
||||
<property name="outputChannel" ref="outputChannel1"/>
|
||||
<property name="durable" value="true"/>
|
||||
</bean>
|
||||
<int-gfe:cq-inbound-channel-adapter query-listener-container="queryListenerContainer"
|
||||
query="select * from /test" channel="outputChannel1" durable="true"/>
|
||||
|
||||
<int:channel id="outputChannel1">
|
||||
<int:queue/>
|
||||
</int:channel>
|
||||
|
||||
<bean id="spelCqMessageProducer" class="org.springframework.integration.gemfire.inbound.ContinuousQueryMessageProducer">
|
||||
<constructor-arg ref="queryListenerContainer"/>
|
||||
<constructor-arg value="select * from /test"/>
|
||||
<property name="outputChannel" ref="outputChannel2"/>
|
||||
<property name="payloadExpression" value="newValue"/>
|
||||
</bean>
|
||||
<int-gfe:cq-inbound-channel-adapter query-listener-container="queryListenerContainer"
|
||||
query="select * from /test" channel="outputChannel2" expression="newValue" query-events="CREATED"/>
|
||||
|
||||
|
||||
<int:channel id="outputChannel2">
|
||||
<int:queue/>
|
||||
</int:channel>
|
||||
@@ -10,7 +10,7 @@
|
||||
* 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;
|
||||
package org.springframework.integration.gemfire.inbound;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
@@ -37,12 +37,11 @@ import com.gemstone.gemfire.internal.cache.LocalRegion;
|
||||
|
||||
/**
|
||||
* @author David Turanski
|
||||
* @since 2.1
|
||||
*
|
||||
*/
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@ContextConfiguration
|
||||
public class ContinuousQueryMessageProducerTests {
|
||||
|
||||
public class CqInboundChannelAdapterTests {
|
||||
static ConfigurableApplicationContext staticCtx;
|
||||
|
||||
@Autowired
|
||||
@@ -82,9 +81,7 @@ public class ContinuousQueryMessageProducerTests {
|
||||
region.put("one",1);
|
||||
Message<?> msg = outputChannel2.receive(1000);
|
||||
assertNotNull(msg);
|
||||
assertEquals(1,msg.getPayload());
|
||||
|
||||
|
||||
assertEquals(1,msg.getPayload());
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
@@ -14,7 +14,7 @@
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.gemfire.store.messagegroupstore;
|
||||
package org.springframework.integration.gemfire.store;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Arrays;
|
||||
@@ -9,9 +9,9 @@
|
||||
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">
|
||||
|
||||
<bean class="org.springframework.integration.gemfire.store.messagegroupstore.GemfireMessageGroupStoreTestConfiguration"/>
|
||||
<bean class="org.springframework.integration.gemfire.store.GemfireMessageGroupStoreTestConfiguration"/>
|
||||
|
||||
<context:property-placeholder location="org/springframework/integration/gemfire/store/messagegroupstore/common.properties"/>
|
||||
<context:property-placeholder location="org/springframework/integration/gemfire/store/common.properties"/>
|
||||
|
||||
<int:channel id="i"/>
|
||||
|
||||
@@ -21,6 +21,6 @@
|
||||
|
||||
<int:service-activator input-channel="o" ref="messageGroupStoreActivator" />
|
||||
|
||||
<util:properties id="props" location="org/springframework/integration/gemfire/store/messagegroupstore/gfe-cache.properties"/>
|
||||
<util:properties id="props" location="org/springframework/integration/gemfire/store/gfe-cache.properties"/>
|
||||
|
||||
</beans>
|
||||
@@ -14,7 +14,7 @@
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package org.springframework.integration.gemfire.store.messagegroupstore;
|
||||
package org.springframework.integration.gemfire.store;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
@@ -44,7 +44,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
*/
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@ContextConfiguration
|
||||
public class GemfireMessageGroupStoreTest {
|
||||
public class GemfireMessageGroupStoreTests {
|
||||
|
||||
@Autowired
|
||||
private GemfireMessageGroupStoreTestConfiguration.FakeMessageConsumer consumer;
|
||||
@@ -12,7 +12,7 @@
|
||||
|
||||
<context:property-placeholder location="org/springframework/integration/gemfire/inbound/cq/common.properties"/>
|
||||
|
||||
<context:component-scan base-package="org.springframework.integration.gemfire.store.messagegroupstore"/>
|
||||
<context:component-scan base-package="org.springframework.integration.gemfire.store"/>
|
||||
|
||||
<int:channel id="i"/>
|
||||
|
||||
@@ -12,11 +12,11 @@
|
||||
|
||||
<!-- Loggers -->
|
||||
<logger name="org.springframework">
|
||||
<level value="debug" />
|
||||
<level value="warn" />
|
||||
</logger>
|
||||
|
||||
<logger name="org.springframework.integration">
|
||||
<level value="debug" />
|
||||
<level value="warn" />
|
||||
</logger>
|
||||
|
||||
<!-- Root Logger -->
|
||||
|
||||
Reference in New Issue
Block a user