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

INT-3104:Add auto-start attribute to Gemfire Adapters

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3104

Add phase and remove unnecessary auto-starup

INT-3104: Remove reduntant code and format

Polishing and fixing parser tests
This commit is contained in:
David Liu
2014-07-25 12:56:18 +03:00
committed by Artem Bilan
parent eff1c7b7d8
commit aaf88ecc7c
13 changed files with 286 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireCqInboundChannelAdapterParser.java
View File

@@ -51,30 +51,33 @@ public class GemfireCqInboundChannelAdapterParser extends AbstractChannelAdapter
private static final String QUERY_EVENTS_ATTRIBUTE = "query-events";
@Override
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
BeanDefinitionBuilder continuousQueryMesageProducer = BeanDefinitionBuilder.genericBeanDefinition(ContinuousQueryMessageProducer.class);
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element, EXPRESSION_ATTRIBUTE,PAYLOAD_EXPRESSION_PROPERTY);
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element, QUERY_EVENTS_ATTRIBUTE, SUPPORTED_EVENT_TYPES_PROPERTY);
BeanDefinitionBuilder continuousQueryMesageProducer =
BeanDefinitionBuilder.genericBeanDefinition(ContinuousQueryMessageProducer.class);
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element,
EXPRESSION_ATTRIBUTE, PAYLOAD_EXPRESSION_PROPERTY);
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element,
QUERY_EVENTS_ATTRIBUTE, SUPPORTED_EVENT_TYPES_PROPERTY);
if (!element.hasAttribute(QUERY_LISTENER_CONTAINER_ATTRIBUTE)){
parserContext.getReaderContext().error("'" + QUERY_LISTENER_CONTAINER_ATTRIBUTE + "' attribute is required.",element);
if (!element.hasAttribute(QUERY_LISTENER_CONTAINER_ATTRIBUTE)) {
parserContext.getReaderContext()
.error("'" + QUERY_LISTENER_CONTAINER_ATTRIBUTE + "' attribute is required.", element);
}
if (!element.hasAttribute(QUERY_ATTRIBUTE)){
parserContext.getReaderContext().error("'" + QUERY_ATTRIBUTE + "' attribute is required.",element);
if (!element.hasAttribute(QUERY_ATTRIBUTE)) {
parserContext.getReaderContext().error("'" + QUERY_ATTRIBUTE + "' 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.setReferenceIfAttributeDefined(continuousQueryMesageProducer, element,
ERROR_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element, QUERY_NAME_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element, DURABLE_ATTRIBUTE);
return continuousQueryMesageProducer.getBeanDefinition();
}

17
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireInboundChannelAdapterParser.java
View File

@@ -45,19 +45,22 @@ public class GemfireInboundChannelAdapterParser extends AbstractChannelAdapterPa
@Override
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
BeanDefinitionBuilder listeningMessageProducer = BeanDefinitionBuilder.genericBeanDefinition(CacheListeningMessageProducer.class);
IntegrationNamespaceUtils.setValueIfAttributeDefined(listeningMessageProducer, element, EXPRESSION_ATTRIBUTE,PAYLOAD_EXPRESSION_PROPERTY);
IntegrationNamespaceUtils.setValueIfAttributeDefined(listeningMessageProducer, element, CACHE_EVENTS_ATTRIBUTE, SUPPORTED_EVENT_TYPES_PROPERTY);
BeanDefinitionBuilder listeningMessageProducer =
BeanDefinitionBuilder.genericBeanDefinition(CacheListeningMessageProducer.class);
IntegrationNamespaceUtils.setValueIfAttributeDefined(listeningMessageProducer, element,
EXPRESSION_ATTRIBUTE, PAYLOAD_EXPRESSION_PROPERTY);
IntegrationNamespaceUtils.setValueIfAttributeDefined(listeningMessageProducer, element,
CACHE_EVENTS_ATTRIBUTE, SUPPORTED_EVENT_TYPES_PROPERTY);
if (!element.hasAttribute(REGION_ATTRIBUTE)){
parserContext.getReaderContext().error("'region' attribute is required.",element);
if (!element.hasAttribute(REGION_ATTRIBUTE)) {
parserContext.getReaderContext().error("'region' attribute is required.", element);
}
listeningMessageProducer.addConstructorArgReference(element.getAttribute(REGION_ATTRIBUTE));
listeningMessageProducer.addPropertyReference(OUTPUT_CHANNEL_PROPERTY, channelName);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(listeningMessageProducer, element, ERROR_CHANNEL_ATTRIBUTE);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(listeningMessageProducer, element,
ERROR_CHANNEL_ATTRIBUTE);
return listeningMessageProducer.getBeanDefinition();
}

12
spring-integration-gemfire/src/main/java/org/springframework/integration/gemfire/config/xml/GemfireOutboundChannelAdapterParser.java
View File

@@ -38,24 +38,24 @@ public class GemfireOutboundChannelAdapterParser extends AbstractOutboundChannel
private static final String REGION_ATTRIBUTE = "region";
/* (non-Javadoc)
* @see org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser#parseConsumer(org.w3c.dom.Element, org.springframework.beans.factory.xml.ParserContext)
* @see AbstractOutboundChannelAdapterParser#parseConsumer(Element, ParserContext)
*/
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder cacheWritingMessageHandler = BeanDefinitionBuilder.genericBeanDefinition(
CacheWritingMessageHandler.class);
if (!element.hasAttribute(REGION_ATTRIBUTE)){
parserContext.getReaderContext().error("'region' attribute is required.",element);
if (!element.hasAttribute(REGION_ATTRIBUTE)) {
parserContext.getReaderContext().error("'region' attribute is required.", element);
}
cacheWritingMessageHandler.addConstructorArgReference(element.getAttribute(REGION_ATTRIBUTE));
Element cacheEntries = DomUtils.getChildElementByTagName(element,CACHE_ENTRIES_ELEMENT);
Element cacheEntries = DomUtils.getChildElementByTagName(element, CACHE_ENTRIES_ELEMENT);
if (cacheEntries != null) {
Map<?,?> map = parserContext.getDelegate().parseMapElement(cacheEntries,cacheWritingMessageHandler.getBeanDefinition());
Map<?, ?> map = parserContext.getDelegate()
.parseMapElement(cacheEntries, cacheWritingMessageHandler.getBeanDefinition());
cacheWritingMessageHandler.addPropertyValue(CACHE_ENTRIES_PROPERTY, map);
}
return cacheWritingMessageHandler.getBeanDefinition();
}
}

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

@@ -20,6 +20,7 @@ import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;
import com.gemstone.gemfire.cache.query.CqEvent;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -30,8 +31,6 @@ import org.springframework.integration.endpoint.ExpressionMessageProducerSupport
import org.springframework.messaging.Message;
import org.springframework.util.Assert;
import com.gemstone.gemfire.cache.query.CqEvent;
/**
* Responds to a Gemfire continuous query (set using the #query field) that is
* constantly evaluated against a cache
@@ -43,7 +42,9 @@ import com.gemstone.gemfire.cache.query.CqEvent;
* @since 2.1
*
*/
public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSupport implements ContinuousQueryListener {
public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSupport
implements ContinuousQueryListener {
private static Log logger = LogFactory.getLog(ContinuousQueryMessageProducer.class);
private final String query;
@@ -102,7 +103,8 @@ public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSup
queryListenerContainer.addListener(new ContinuousQueryDefinition(this.query, this, this.durable));
}
else {
queryListenerContainer.addListener(new ContinuousQueryDefinition(this.queryName, this.query, this, this.durable));
queryListenerContainer.addListener(new ContinuousQueryDefinition(this.queryName, this.query, this,
this.durable));
}
}
@@ -120,17 +122,18 @@ public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSup
logger.debug(String.format("processing cq event key [%s] event [%s]", event.getQueryOperation()
.toString(), event.getKey()));
}
Message<?> cqEventMessage = this.getMessageBuilderFactory().withPayload(evaluatePayloadExpression(event)).build();
Message<?> cqEventMessage = this.getMessageBuilderFactory().withPayload(evaluatePayloadExpression(event))
.build();
sendMessage(cqEventMessage);
}
}
private boolean isEventSupported(CqEvent event) {
String eventName = event.getQueryOperation().toString() +
(event.getQueryOperation().toString().endsWith("Y")? "ED" : "D");
CqEventType eventType = CqEventType.valueOf(eventName);
return supportedEventTypes.contains(eventType);
String eventName = event.getQueryOperation().toString() +
(event.getQueryOperation().toString().endsWith("Y") ? "ED" : "D");
CqEventType eventType = CqEventType.valueOf(eventName);
return supportedEventTypes.contains(eventType);
}
}

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

@@ -21,6 +21,10 @@ import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Map.Entry;
import com.gemstone.gemfire.GemFireCheckedException;
import com.gemstone.gemfire.GemFireException;
import com.gemstone.gemfire.cache.Region;
import org.springframework.data.gemfire.GemfireCallback;
import org.springframework.data.gemfire.GemfireTemplate;
import org.springframework.expression.Expression;
@@ -30,10 +34,6 @@ import org.springframework.messaging.Message;
import org.springframework.messaging.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.
@@ -59,13 +59,15 @@ public class CacheWritingMessageHandler extends AbstractMessageHandler {
return "gemfire:outbound-channel-adapter";
}
@SuppressWarnings("unchecked")
@Override
public void handleMessageInternal(Message<?> message) {
Object payload = message.getPayload();
Map<?, ?> cacheValues = (cacheEntryExpressions.size() > 0)?parseCacheEntries(message):null;
Map<?, ?> cacheValues = (cacheEntryExpressions.size() > 0) ? parseCacheEntries(message) : null;
if (cacheValues == null) {
Assert.isTrue(payload instanceof Map, "If cache entry expressions are not configured, then payload must be a Map");
Assert.isTrue(payload instanceof Map,
"If cache entry expressions are not configured, then payload must be a Map");
cacheValues = (Map<?, ?>) payload;
}
@@ -73,7 +75,7 @@ public class CacheWritingMessageHandler extends AbstractMessageHandler {
this.gemfireTemplate.execute(new GemfireCallback<Object>() {
@Override
@SuppressWarnings({ "rawtypes", "unchecked" })
@SuppressWarnings({"rawtypes", "unchecked"})
public Object doInGemfire(Region region) throws GemFireCheckedException, GemFireException {
region.putAll(map);
return null;
@@ -88,7 +90,7 @@ public class CacheWritingMessageHandler extends AbstractMessageHandler {
else {
Map<Object, Object> cacheValues = new HashMap<Object, Object>();
for (Entry<Expression, Expression> expressionEntry : cacheEntryExpressions.entrySet()) {
cacheValues.put(expressionEntry.getKey().getValue(message),expressionEntry.getValue().getValue(message));
cacheValues.put(expressionEntry.getKey().getValue(message), expressionEntry.getValue().getValue(message));
}
return cacheValues;
}

71
spring-integration-gemfire/src/main/resources/org/springframework/integration/gemfire/config/xml/spring-integration-gemfire-4.1.xsd
View File

@@ -1,16 +1,16 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/gemfire"
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
targetNamespace="http://www.springframework.org/schema/integration/gemfire"
elementFormDefault="qualified" attributeFormDefault="unqualified">
xmlns:xsd="http://www.w3.org/2001/XMLSchema" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
xmlns:integration="http://www.springframework.org/schema/integration"
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.xsd" />
<xsd:import namespace="http://www.springframework.org/schema/tool" />
schemaLocation="http://www.springframework.org/schema/beans/spring-beans.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-4.1.xsd" />
schemaLocation="http://www.springframework.org/schema/integration/spring-integration-4.1.xsd"/>
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the core configuration elements for Spring Integration GemFire Support.
@@ -33,14 +33,14 @@
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="com.gemstone.gemfire.cache.Region" />
<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">
use="optional" default="CREATED,UPDATED">
<xsd:annotation>
<xsd:documentation><![CDATA[
Enabled cache entry event types
@@ -48,7 +48,7 @@
<xsd:appinfo>
<tool:annotation kind="value">
<tool:expected-type
type="org.springframework.integration.gemfire.inbound.EventType" />
type="org.springframework.integration.gemfire.inbound.EventType"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -70,20 +70,20 @@
<xsd:complexContent>
<xsd:extension base="InboundChannelAdapterType">
<xsd:attribute name="cq-listener-container" use="required">
<xsd:annotation>
<xsd:annotation>
<xsd:documentation><![CDATA[
Reference to a ContinuousQueryListenerContainer
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="value">
<tool:expected-type
type="org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer" />
type="org.springframework.data.gemfire.listener.ContinuousQueryListenerContainer"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="query-events" type="xsd:string"
use="optional" default="CREATED,UPDATED">
use="optional" default="CREATED,UPDATED">
<xsd:annotation>
<xsd:documentation><![CDATA[
Enabled continuous query event types
@@ -91,24 +91,24 @@
<xsd:appinfo>
<tool:annotation kind="value">
<tool:expected-type
type="org.springframework.integration.gemfire.inbound.CqEventType" />
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[
<xsd:annotation>
<xsd:documentation><![CDATA[
The query string
]]></xsd:documentation>
</xsd:annotation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="query-name" use="optional" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
<xsd:annotation>
<xsd:documentation><![CDATA[
The query name
]]></xsd:documentation>
</xsd:annotation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="durable" use="optional" default="false">
<xsd:annotation>
@@ -121,7 +121,7 @@
</xsd:simpleType>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexContent>
</xsd:complexType>
</xsd:element>
@@ -136,35 +136,25 @@
</xsd:annotation>
<xsd:choice minOccurs="0" maxOccurs="2">
<xsd:element name="cache-entries" type="beans:mapType"
minOccurs="0" maxOccurs="1">
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:documentation>
</xsd:annotation>
</xsd:element>
<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType" minOccurs="0" maxOccurs="1" />
<xsd:element name="request-handler-advice-chain" type="integration:handlerAdviceChainType" minOccurs="0"
maxOccurs="1"/>
</xsd:choice>
<xsd:attribute name="id" type="xsd:string" use="optional" />
<xsd:attribute name="channel" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type
type="org.springframework.messaging.MessageChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<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:expected-type type="com.gemstone.gemfire.cache.Region"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
@@ -184,8 +174,9 @@
<xsd:complexType name="InboundChannelAdapterType">
<xsd:attributeGroup ref="integration:channelAdapterAttributes"/>
<xsd:attribute name="error-channel" type="xsd:string"
use="optional">
use="optional">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -196,7 +187,7 @@
reference to the "nullChannel" here.
]]></xsd:documentation>
<tool:expected-type
type="org.springframework.messaging.MessageChannel" />
type="org.springframework.messaging.MessageChannel"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>