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

INTEXT-115 Update to SI 4.0, SF 4.0

Browse Source 
JIRA: http://jira.spring.io/browse/INTEXT-115
This commit is contained in:
Gary Russell
2014-10-16 10:17:50 -04:00
parent 54245721d1
commit f2bedb8f6a
14 changed files with 109 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
pom.xml
View File

@@ -4,11 +4,11 @@
<modelVersion>4.0.0</modelVersion>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-flow</artifactId>
<version>1.0.0.BUILD-SNAPSHOT</version>
<version>4.0.0.BUILD-SNAPSHOT</version>
<name>Spring Integration Flow Support</name>
<properties>
<spring.version>3.2.2.RELEASE</spring.version>
<spring.integration.version>3.0.0.BUILD-SNAPSHOT</spring.integration.version>
<spring.version>4.0.7.RELEASE</spring.version>
<spring.integration.version>4.0.4.RELEASE</spring.integration.version>
<project.build.sourceEncoding>UTF8</project.build.sourceEncoding>
</properties>
<licenses>

39
src/main/java/org/springframework/integration/flow/Flow.java
View File

@@ -9,6 +9,7 @@ import java.util.Set;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
@@ -17,13 +18,13 @@ import org.springframework.beans.factory.support.BeanDefinitionValidationExcepti
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.flow.config.FlowUtils;
import org.springframework.integration.flow.interceptor.FlowInterceptor;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.support.channel.ChannelResolver;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.core.DestinationResolver;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -36,16 +37,16 @@ import org.springframework.util.StringUtils;
* bean definitions used by the flow. In addition, beans defined in the parent
* application context may be referenced or overridden in the flow application
* context.
*
*
* By convention the flow configuration resource locations are
* classpath:META-INF/spring/integration/flows/[flow-id]/*.xml
*
*
* The flow-id defaults to the bean name if not set
*
*
* @author David Turanski
*
*
*/
public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, ApplicationContextAware {
public class Flow implements InitializingBean, BeanNameAware, DestinationResolver, ApplicationContextAware {
private static Log logger = LogFactory.getLog(Flow.class);
@@ -65,7 +66,7 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
private volatile String flowId;
private volatile ChannelResolver flowChannelResolver;
private volatile BeanFactoryChannelResolver flowChannelResolver;
private volatile SubscribableChannel flowOutputChannel;
@@ -79,7 +80,7 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
}
/**
*
*
* @param flowProperties properties for this flow instance
* @param configLocations Spring configuration resource locations containing
* bean definitions included in the flow application context
@@ -90,7 +91,7 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
}
/**
*
*
* @param configLocations Spring configuration resource locations containing
* bean definitions included in the flow application context
*/
@@ -98,6 +99,7 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
this.configLocations = configLocations;
}
@Override
public void afterPropertiesSet() {
if (this.flowId == null) {
@@ -153,6 +155,7 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
return this.flowConfiguration;
}
@Override
public void setBeanName(String name) {
this.beanName = name;
@@ -170,7 +173,7 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
}
/**
*
*
* @param referencedBeanLocations Additional resource locations containing
* referenced bean definitions
*/
@@ -179,7 +182,7 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
}
/**
*
*
* @param flowProperties properties referenced in the flow definition
* property placeholders
*/
@@ -192,7 +195,7 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
}
/**
*
*
* @param help if true write the flow documentation to stdout The default
* document location is
* "classpath:META-INF/spring/integration/flows/[flow-id]/flow.doc"
@@ -219,8 +222,9 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
this.flowOutputChannel = flowOutputChannel;
}
public MessageChannel resolveChannelName(String channelName) {
return flowChannelResolver.resolveChannelName(channelName);
@Override
public MessageChannel resolveDestination(String channelName) {
return flowChannelResolver.resolveDestination(channelName);
}
private void addReferencedProperties() {
@@ -266,7 +270,7 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
for (PortConfiguration targetPortConfiguration : this.getFlowConfiguration().getPortConfigurations()) {
for (String outputPort : targetPortConfiguration.getOutputPortNames()) {
String targetOutputChannelName = (String) targetPortConfiguration.getOutputChannel(outputPort);
SubscribableChannel inputChannel = (SubscribableChannel) resolveChannelName(targetOutputChannelName);
SubscribableChannel inputChannel = (SubscribableChannel) resolveDestination(targetOutputChannelName);
((AbstractMessageChannel) inputChannel).addInterceptor(new FlowInterceptor(outputPort));
@@ -277,6 +281,7 @@ public class Flow implements InitializingBean, BeanNameAware, ChannelResolver, A
}
}
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}

22
src/main/java/org/springframework/integration/flow/config/FlowMessageHandlerFactoryBean.java
View File

@@ -1,12 +1,12 @@
/*
* 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
@@ -17,18 +17,19 @@ package org.springframework.integration.flow.config;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.config.AbstractSimpleMessageHandlerFactoryBean;
import org.springframework.integration.flow.Flow;
import org.springframework.integration.flow.PortConfiguration;
import org.springframework.integration.flow.handler.FlowMessageHandler;
import org.springframework.messaging.MessageChannel;
import org.springframework.util.Assert;
/**
* Creates an instance of {@link FlowMessageHandler}
* @author David Turanski
*
*
*/
public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerFactoryBean
implements InitializingBean {
@@ -51,7 +52,7 @@ public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerF
@Override
protected FlowMessageHandler createHandler() {
MessageChannel flowInputChannel = flow.resolveChannelName((String) flowConfiguration.getInputChannel());
MessageChannel flowInputChannel = flow.resolveDestination((String) flowConfiguration.getInputChannel());
FlowMessageHandler flowMessageHandler = new FlowMessageHandler(flowInputChannel, flow.getFlowOutputChannel(),
timeout);
@@ -62,7 +63,7 @@ public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerF
}
/**
*
*
* @param flow the flow handled by the FlowMessageHandler
*/
public void setFlow(Flow flow) {
@@ -70,7 +71,7 @@ public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerF
}
/**
*
*
* @param inputPortName the flow input port associated with the handler. If
* not set and the flow defines only one input port, that will be used by
* default.
@@ -80,7 +81,7 @@ public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerF
}
/**
*
*
* @param timeout send timeout for the handler
*/
public void setTimeout(long timeout) {
@@ -88,13 +89,14 @@ public class FlowMessageHandlerFactoryBean extends AbstractSimpleMessageHandlerF
}
/**
*
*
* @param errorChannel
*/
public void setErrorChannel(MessageChannel errorChannel) {
this.errorChannel = errorChannel;
}
@Override
public void afterPropertiesSet() throws Exception {
this.flowConfiguration = null;
if (this.inputPortName == null) {

4
src/main/java/org/springframework/integration/flow/config/FlowUtils.java
View File

@@ -26,8 +26,8 @@ import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.util.ResourceUtils;

54
src/main/java/org/springframework/integration/flow/handler/FlowMessageHandler.java
View File

@@ -1,12 +1,12 @@
/*
* 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
@@ -20,38 +20,40 @@ import java.util.UUID;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.flow.FlowConstants;
import org.springframework.integration.flow.config.FlowUtils;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.ErrorMessage;
/**
* A MessageHandler for Handling Flow input and output. Sends messages on its
* input channel to the flow input channel and replies with the flow output (if
* there is one) to its output channel.
*
*
* Internally creates a subscriber to a PublishSubscribeChannel automatically
* created for the flow. Since all FlowMessageHandler instances subscribe to
* this channel, a unique flow conversationId is used to correlate flow input
* and output messages
*
*
* The output message contains a FLOW_OUTPUT_PORT_HEADER identifying which flow
* output port produced the message
* @see FlowUtils
*
*
* Error handling is done in a standard way. If the flow includes an error
* channel which is bound to an output port, the handler will send the
* ErrorMessage to the outputChannel. If the flow throws an exception, the
* handler will create an ErrorMessage and send it to the outputChannel
*
*
* @author David Turanski
*
*
*/
public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
@@ -66,7 +68,7 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
private final long timeout;
/**
*
*
* @param flowInputChannel the Flow input channel
* @param flowOutputChannel a PublishSubscribeChannel internally created and
* bridged to all flow output channels
@@ -91,16 +93,16 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
ResponseMessageHandler responseMessageHandler = new ResponseMessageHandler(conversationId);
flowOutputChannel.subscribe(responseMessageHandler);
flowInputChannel.send(message, timeout);
flowInputChannel.send(message, timeout);
flowOutputChannel.unsubscribe(responseMessageHandler);
return responseMessageHandler.getResponse();
}
catch (MessagingException me) {
log.error(me.getMessage(), me);
if (conversationId.equals(me.getFailedMessage().getHeaders().getCorrelationId())) {
if (conversationId.equals(new IntegrationMessageHeaderAccessor(me.getFailedMessage()).getCorrelationId())) {
if (errorChannel != null) {
errorChannel.send(new ErrorMessage(me, Collections.singletonMap(
FlowConstants.FLOW_OUTPUT_PORT_HEADER,
@@ -129,18 +131,20 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
/*
* (non-Javadoc)
*
*
* @see
* org.springframework.integration.core.MessageHandler#handleMessage
* (org.springframework.integration.Message)
* org.springframework.messaging.MessageHandler#handleMessage
* (org.springframework.messaging.Message)
*/
@Override
public void handleMessage(Message<?> message) throws MessagingException {
Object correlationId = new IntegrationMessageHeaderAccessor(message).getCorrelationId();
if (log.isDebugEnabled()) {
log.debug("handling flow response message with conversation Id "
+ message.getHeaders().getCorrelationId() + ". Target conversation Id = " + this.conversationId
+ " match = " + conversationId.equals(message.getHeaders().getCorrelationId()));
+ correlationId + ". Target conversation Id = " + this.conversationId
+ " match = " + conversationId.equals(correlationId));
}
if (conversationId.equals(message.getHeaders().getCorrelationId())) {
if (conversationId.equals(correlationId)) {
this.response = MessageBuilder.fromMessage(message).popSequenceDetails().build();
if (log.isDebugEnabled()) {
log.debug("set flow response message " + this.response);
@@ -154,7 +158,7 @@ public class FlowMessageHandler extends AbstractReplyProducingMessageHandler {
*/
if (message instanceof ErrorMessage) {
MessagingException me = (MessagingException) message.getPayload();
if (conversationId.equals(me.getFailedMessage().getHeaders().getCorrelationId())) {
if (conversationId.equals(new IntegrationMessageHeaderAccessor(me.getFailedMessage()).getCorrelationId())) {
this.response = message;
}
}

6
src/main/java/org/springframework/integration/flow/interceptor/FlowInterceptor.java
View File

@@ -20,9 +20,9 @@ import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.interceptor.ChannelInterceptorAdapter;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.ChannelInterceptorAdapter;
import org.springframework.integration.flow.FlowConstants;
import org.springframework.integration.flow.config.FlowUtils;
import org.springframework.integration.support.MessageBuilder;

8
src/test/java/org/springframework/integration/flow/FlowUtilsTests.java
View File

@@ -4,14 +4,14 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import org.junit.Test;
import org.springframework.integration.Message;
import org.springframework.messaging.Message;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.integration.flow.config.FlowUtils;
import org.springframework.integration.message.GenericMessage;
import org.springframework.messaging.support.GenericMessage;
/**
*
* @author David Turanski

4
src/test/java/org/springframework/integration/flow/Transaction/TransactionalServiceActivator.java
View File

@@ -12,7 +12,7 @@
*/
package org.springframework.integration.flow.Transaction;
import org.springframework.integration.Message;
import org.springframework.messaging.Message;
import org.springframework.integration.gateway.RequestReplyExchanger;
import org.springframework.transaction.annotation.Transactional;
@@ -28,7 +28,7 @@ public class TransactionalServiceActivator implements RequestReplyExchanger {
this.gateway = gateway;
}
/* (non-Javadoc)
* @see org.springframework.integration.gateway.RequestReplyExchanger#exchange(org.springframework.integration.Message)
* @see org.springframework.integration.gateway.RequestReplyExchanger#exchange(org.springframework.messaging.Message)
*/
public Message<?> exchange(Message<?> request) {
return gateway.exchange(request);

2
src/test/java/org/springframework/integration/flow/config/xml/FlowClientNamespaceTests.java
View File

@@ -25,7 +25,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.Message;
import org.springframework.messaging.Message;
import org.springframework.integration.flow.Flow;
import org.springframework.integration.test.support.AbstractRequestResponseScenarioTests;
import org.springframework.integration.test.support.MessageValidator;

18
src/test/java/org/springframework/integration/flow/config/xml/FlowWithErrorTests.java
View File

@@ -21,13 +21,13 @@ import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
/**
*
@@ -93,8 +93,8 @@ public class FlowWithErrorTests {
* (non-Javadoc)
*
* @see
* org.springframework.integration.core.MessageHandler#handleMessage
* (org.springframework.integration.Message)
* org.springframework.messaging.MessageHandler#handleMessage
* (org.springframework.messaging.Message)
*/
public void handleMessage(Message<?> message) throws MessagingException {
this.gotResponse = true;

12
src/test/java/org/springframework/integration/flow/config/xml/FlowWithOptionalResponseTests.java
View File

@@ -6,12 +6,12 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

2
src/test/java/org/springframework/integration/flow/config/xml/FlowWithReferencesTests.java
View File

@@ -18,7 +18,7 @@ package org.springframework.integration.flow.config.xml;
import static org.junit.Assert.assertEquals;
import org.junit.runner.RunWith;
import org.springframework.integration.Message;
import org.springframework.messaging.Message;
import org.springframework.integration.test.support.MessageValidator;
import org.springframework.integration.test.support.RequestResponseScenario;
import org.springframework.integration.test.support.SingleRequestResponseScenarioTests;

12
src/test/java/org/springframework/integration/flow/config/xml/NestedFlowTests.java
View File

@@ -7,12 +7,12 @@ import java.util.concurrent.atomic.AtomicInteger;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;

18
src/test/java/org/springframework/integration/flow/config/xml/TransactionalFlowTests.java
View File

@@ -20,15 +20,15 @@ import static org.junit.Assert.fail;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.integration.flow.FlowConstants;
import org.springframework.integration.flow.Transaction.StubTransactionManager;
import org.springframework.integration.message.ErrorMessage;
import org.springframework.integration.message.GenericMessage;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.messaging.support.GenericMessage;
/**
* @author David Turanski
@@ -119,8 +119,8 @@ public class TransactionalFlowTests {
* (non-Javadoc)
*
* @see
* org.springframework.integration.core.MessageHandler#handleMessage
* (org.springframework.integration.Message)
* org.springframework.messaging.MessageHandler#handleMessage
* (org.springframework.messaging.Message)
*/
public void handleMessage(Message<?> message) throws MessagingException {