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

INT-2738 Add Missing getComponentType() Methods

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

Add a `getComponentType()` method for all handlers,
message producers, and message sources that implement
`NamedComponent` (see `SPCA.getComponentType()`).

Add rome as an optional dependency - STS complained
because AtomFeedHttpMessageConverter has a dependency
on it.
This commit is contained in:
Gary Russell
2014-04-12 16:20:59 -04:00
committed by Artem Bilan
parent 3839eca5ca
commit a7489909d7
54 changed files with 384 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2011 the original author or authors.
*
* Copyright 2002-2014 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.
@@ -19,62 +19,62 @@ 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.integration.gemfire.inbound.ContinuousQueryMessageProducer;
/**
* @author David Turanski
* @author Dan Oxlade
* @author Gary Russell
* @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 = "cq-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";
@Override
protected AbstractBeanDefinition doParse(Element element, ParserContext parserContext, String channelName) {
BeanDefinitionBuilder continuousQueryMesageProducer = BeanDefinitionBuilder.genericBeanDefinition(GEMFIRE_INBOUND_CONTINUOUS_QUERY_MESSAGE_PRODUCER);
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_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.setValueIfAttributeDefined(continuousQueryMesageProducer, element, QUERY_NAME_ATTRIBUTE);
IntegrationNamespaceUtils.setValueIfAttributeDefined(continuousQueryMesageProducer, element, DURABLE_ATTRIBUTE);
return continuousQueryMesageProducer.getBeanDefinition();
}

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

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2011 the original author or authors.
*
* Copyright 2002-2014 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.
@@ -20,47 +20,44 @@ 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.integration.gemfire.inbound.CacheListeningMessageProducer;
/**
* @author David Turanski
* @author Gary Russell
* @since 2.1
*/
public class GemfireInboundChannelAdapterParser extends AbstractChannelAdapterParser {
private static final String ERROR_CHANNEL_ATTRIBUTE = "error-channel";
private static final String OUTPUT_CHANNEL_PROPERTY = "outputChannel";
private static final String REGION_ATTRIBUTE = "region";
private static final String PAYLOAD_EXPRESSION_PROPERTY = "payloadExpression";
private static final String EXPRESSION_ATTRIBUTE = "expression";
private static final String GEMFIRE_INBOUND_CACHE_LISTENING_MESSAGE_PRODUCER = "org.springframework.integration.gemfire.inbound.CacheListeningMessageProducer";
private static final String SUPPORTED_EVENT_TYPES_PROPERTY = "supportedEventTypes";
private static final String CACHE_EVENTS_ATTRIBUTE = "cache-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 listeningMessageProducer = BeanDefinitionBuilder.genericBeanDefinition(GEMFIRE_INBOUND_CACHE_LISTENING_MESSAGE_PRODUCER);
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);
}
listeningMessageProducer.addConstructorArgReference(element.getAttribute(REGION_ATTRIBUTE));
listeningMessageProducer.addPropertyReference(OUTPUT_CHANNEL_PROPERTY, channelName);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(listeningMessageProducer, element, ERROR_CHANNEL_ATTRIBUTE);
return listeningMessageProducer.getBeanDefinition();
}

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

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2011 the original author or authors.
*
* Copyright 2002-2014 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.
@@ -21,21 +21,21 @@ 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.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.gemfire.outbound.CacheWritingMessageHandler;
import org.springframework.util.xml.DomUtils;
/**
* @author David Turanski
* @author Gary Russell
* @since 2.1
*/
public class GemfireOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
private static final String CACHE_ENTRIES_PROPERTY = "cacheEntries";
private static final String CACHE_ENTRIES_ELEMENT = "cache-entries";
private static final String REGION_ATTRIBUTE = "region";
private static final String GEMFIRE_OUTBOUND_CACHE_WRITING_MESSAGE_HANDLER = "org.springframework.integration.gemfire.outbound.CacheWritingMessageHandler";
/* (non-Javadoc)
* @see org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser#parseConsumer(org.w3c.dom.Element, org.springframework.beans.factory.xml.ParserContext)
@@ -43,19 +43,19 @@ public class GemfireOutboundChannelAdapterParser extends AbstractOutboundChannel
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder cacheWritingMessageHandler = BeanDefinitionBuilder.genericBeanDefinition(
GEMFIRE_OUTBOUND_CACHE_WRITING_MESSAGE_HANDLER);
CacheWritingMessageHandler.class);
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);
if (cacheEntries != null) {
Map<?,?> map = parserContext.getDelegate().parseMapElement(cacheEntries,cacheWritingMessageHandler.getBeanDefinition());
cacheWritingMessageHandler.addPropertyValue(CACHE_ENTRIES_PROPERTY, map);
}
return cacheWritingMessageHandler.getBeanDefinition();
}
}

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2011 the original author or authors.
* Copyright 2002-2014 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.
@@ -68,6 +68,10 @@ public class CacheListeningMessageProducer extends ExpressionMessageProducerSupp
this.supportedEventTypes = new HashSet<EventType>(Arrays.asList(eventTypes));
}
@Override
public String getComponentType() {
return "gemfire:inbound-channel-adapter";
}
@Override
protected void doStart() {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2014 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.
@@ -90,6 +90,11 @@ public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSup
this.supportedEventTypes = new HashSet<CqEventType>(Arrays.asList(eventTypes));
}
@Override
public String getComponentType() {
return "gemfire:cq-inbound-channel-adapter";
}
@Override
protected void onInit() {
super.onInit();
@@ -108,6 +113,7 @@ public class ContinuousQueryMessageProducer extends ExpressionMessageProducerSup
* org.springframework.data.gemfire.listener.QueryListener#onEvent(com.gemstone
* .gemfire.cache.query.CqEvent)
*/
@Override
public void onEvent(CqEvent event) {
if (isEventSupported(event)) {
if (logger.isDebugEnabled()) {

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

@@ -25,9 +25,9 @@ import org.springframework.data.gemfire.GemfireCallback;
import org.springframework.data.gemfire.GemfireTemplate;
import org.springframework.expression.Expression;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.util.Assert;
import com.gemstone.gemfire.GemFireCheckedException;
@@ -37,7 +37,7 @@ 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
* @author David Turanski
* @since 2.1
@@ -51,22 +51,28 @@ public class CacheWritingMessageHandler extends AbstractMessageHandler {
public CacheWritingMessageHandler(Region region) {
Assert.notNull(region, "region must not be null");
this.gemfireTemplate.setRegion(region);
this.gemfireTemplate.afterPropertiesSet();
this.gemfireTemplate.afterPropertiesSet();
}
@Override
public String getComponentType() {
return "gemfire:outbound-channel-adapter";
}
@Override
public void handleMessageInternal(Message<?> message) {
Object payload = message.getPayload();
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");
cacheValues = (Map<?, ?>) payload;
}
final Map<?, ?> map = cacheValues;
this.gemfireTemplate.execute(new GemfireCallback<Object>() {
@Override
@SuppressWarnings({ "rawtypes", "unchecked" })
public Object doInGemfire(Region region) throws GemFireCheckedException, GemFireException {
region.putAll(map);
@@ -89,11 +95,11 @@ public class CacheWritingMessageHandler extends AbstractMessageHandler {
}
public void setCacheEntries(Map<String, String> cacheEntries) {
if (cacheEntryExpressions.size() > 0) {
cacheEntryExpressions.clear();
}
for (Entry<String, String> cacheEntry : cacheEntries.entrySet()) {
this.cacheEntryExpressions.put(new SpelExpressionParser().parseExpression(cacheEntry.getKey()),
new SpelExpressionParser().parseExpression(cacheEntry.getValue()));