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

Revert IntegrationFlow processing to BFPP

Browse Source 
Add test case for `@MessageEndpoint` to et deal with channel from `IntegrationFlow`
As far as `AbstractMethodAnnotationPostProcessor` tries to resolve `channel` on endpoint creation,
the `IntegrationFlow` processing can't be done in the `BPP`, because `MessagingAnnotationPostProcessor` may parse
annotations for bean before `IntegrationFlowBeanPostProcessor` will do its stuff.
This commit is contained in:
Artem Bilan
2014-02-14 12:08:28 +02:00
parent 1e6d50d33c
commit 0baeb52b3f
4 changed files with 105 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/DslIntegrationConfigurationInitializer.java
View File

@@ -16,13 +16,24 @@
package org.springframework.integration.dsl;
import java.util.Collection;
import java.util.Map;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.GenericBeanDefinition;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.config.IntegrationConfigurationInitializer;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.dsl.config.InstanceBeanDefinition;
import org.springframework.integration.dsl.core.Spec;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
/**
@@ -34,24 +45,87 @@ public class DslIntegrationConfigurationInitializer implements IntegrationConfig
@Override
public void initialize(ConfigurableListableBeanFactory configurableListableBeanFactory) throws BeansException {
Assert.isInstanceOf(BeanDefinitionRegistry.class, configurableListableBeanFactory,
"To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." +
"Consider using 'GenericApplicationContext' implementation.");
this.initializeIntegrationFlows(configurableListableBeanFactory);
this.populateBeansFromSpecs(configurableListableBeanFactory);
configurableListableBeanFactory.addBeanPostProcessor(new IntegrationFlowBeanPostProcessor(configurableListableBeanFactory));
}
private void initializeIntegrationFlows(ConfigurableListableBeanFactory beanFactory) {
Map<String, IntegrationFlow> integrationFlows = beanFactory.getBeansOfType(IntegrationFlow.class, false, false);
BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
for (Map.Entry<String, IntegrationFlow> integrationFlowEntry : integrationFlows.entrySet()) {
String flowName = integrationFlowEntry.getKey();
String flowNamePrefix = flowName + ":";
IntegrationFlow flow = integrationFlowEntry.getValue();
int channelNameIndex = 0;
for (AbstractBeanDefinition beanDefinition : flow.getIntegrationComponents()) {
if (beanDefinition instanceof InstanceBeanDefinition) {
final Object instance = beanDefinition.getSource();
Collection<?> values = beanFactory.getBeansOfType(instance.getClass(), false, false).values();
if (!values.contains(instance)) {
if (instance instanceof AbstractMessageChannel) {
String channelBeanName = ((AbstractMessageChannel) instance).getComponentName();
if (channelBeanName == null) {
channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
}
registry.registerBeanDefinition(channelBeanName, beanDefinition);
}
else if (instance instanceof EndpointSpec) {
EndpointSpec<?, ?> endpointSpec = (EndpointSpec<?, ?>) instance;
MessageHandler messageHandler = endpointSpec.getHandler();
ConsumerEndpointFactoryBean endpoint = endpointSpec.getEndpoint();
String id = endpointSpec.getId();
String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler);
String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null;
BeanComponentDefinition definitionHolder = new BeanComponentDefinition(new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias);
BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry);
String endpointBeanName = id;
if (endpointBeanName == null) {
endpointBeanName = generateInstanceBeanDefinitionName(registry, endpoint);
}
registry.registerBeanDefinition(endpointBeanName, new InstanceBeanDefinition(endpoint));
}
else {
String beanName = generateInstanceBeanDefinitionName(registry, instance);
registry.registerBeanDefinition(beanName, beanDefinition);
}
}
}
else {
BeanDefinitionReaderUtils.registerWithGeneratedName(beanDefinition, registry);
}
}
registry.removeBeanDefinition(flowName);
beanFactory.destroyBean(flowName);
}
}
private void populateBeansFromSpecs(ConfigurableListableBeanFactory beanFactory) {
Assert.isInstanceOf(BeanDefinitionRegistry.class, beanFactory,
"To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." +
"Consider using 'GenericApplicationContext' implementation."
);
Map<String, Spec> specs = beanFactory.getBeansOfType(Spec.class, false, false);
BeanDefinitionRegistry registry = (BeanDefinitionRegistry) beanFactory;
for (Map.Entry<String, Spec> specEntry : specs.entrySet()) {
String id = specEntry.getKey();
Spec<?, ?> spec = specEntry.getValue();
registry.removeBeanDefinition(id);
beanFactory.destroyBean(id);
beanFactory.registerSingleton(id, spec.get());
beanFactory.initializeBean(spec.get(), id);
}
}
@SuppressWarnings("serial")
private static String generateInstanceBeanDefinitionName(BeanDefinitionRegistry registry, final Object instance) {
return BeanDefinitionReaderUtils.generateBeanName(new GenericBeanDefinition() {
@Override
public String getBeanClassName() {
return instance.getClass().getName();
}
}, registry);
}
}

106
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java
View File

@@ -1,106 +0,0 @@
package org.springframework.integration.dsl;
import java.util.Collection;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.parsing.BeanComponentDefinition;
import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.GenericBeanDefinition;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.config.ConsumerEndpointFactoryBean;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.dsl.config.InstanceBeanDefinition;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.Assert;
/**
* @author Artem Bilan
* @since 4.0
*/
public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor {
private final ConfigurableListableBeanFactory beanFactory;
private final BeanDefinitionRegistry registry;
public IntegrationFlowBeanPostProcessor(ConfigurableListableBeanFactory beanFactory) {
Assert.isInstanceOf(BeanDefinitionRegistry.class, beanFactory,
"To use Spring Integration Java DSL the 'beanFactory' has to be an instance of 'BeanDefinitionRegistry'." +
"Consider using 'GenericApplicationContext' implementation."
);
this.beanFactory = beanFactory;
this.registry = (BeanDefinitionRegistry) beanFactory;
}
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof IntegrationFlow) {
String flowNamePrefix = beanName + ":";
int channelNameIndex = 0;
for (AbstractBeanDefinition beanDefinition : ((IntegrationFlow) bean).getIntegrationComponents()) {
if (beanDefinition instanceof InstanceBeanDefinition) {
final Object instance = beanDefinition.getSource();
Collection<?> values = this.beanFactory.getBeansOfType(instance.getClass(), false, false).values();
if (!values.contains(instance)) {
if (instance instanceof AbstractMessageChannel) {
String channelBeanName = ((AbstractMessageChannel) instance).getComponentName();
if (channelBeanName == null) {
channelBeanName = flowNamePrefix + "channel" + BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
}
registry.registerBeanDefinition(channelBeanName, beanDefinition);
}
else if (instance instanceof EndpointSpec) {
EndpointSpec<?, ?> endpointSpec = (EndpointSpec<?, ?>) instance;
MessageHandler messageHandler = endpointSpec.getHandler();
ConsumerEndpointFactoryBean endpoint = endpointSpec.getEndpoint();
String id = endpointSpec.getId();
String handlerBeanName = generateInstanceBeanDefinitionName(registry, messageHandler);
String[] handlerAlias = id != null ? new String[]{id + IntegrationNamespaceUtils.HANDLER_ALIAS_SUFFIX} : null;
BeanComponentDefinition definitionHolder = new BeanComponentDefinition(
new InstanceBeanDefinition(messageHandler), handlerBeanName, handlerAlias);
BeanDefinitionReaderUtils.registerBeanDefinition(definitionHolder, registry);
String endpointBeanName = id;
if (endpointBeanName == null) {
endpointBeanName = generateInstanceBeanDefinitionName(registry, endpoint);
}
registry.registerBeanDefinition(endpointBeanName, new InstanceBeanDefinition(endpoint));
}
else {
String name = generateInstanceBeanDefinitionName(registry, instance);
registry.registerBeanDefinition(name, beanDefinition);
}
}
}
else {
BeanDefinitionReaderUtils.registerWithGeneratedName(beanDefinition, registry);
}
}
}
return bean;
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
return bean;
}
@SuppressWarnings("serial")
private static String generateInstanceBeanDefinitionName(BeanDefinitionRegistry registry, final Object instance) {
return BeanDefinitionReaderUtils.generateBeanName(new GenericBeanDefinition() {
@Override
public String getBeanClassName() {
return instance.getClass().getName();
}
}, registry);
}
}

4
spring-integration-java-dsl/src/main/java/org/springframework/integration/dsl/IntegrationFlowBuilder.java
View File

@@ -28,7 +28,6 @@ import org.springframework.integration.dsl.support.EndpointConfigurer;
import org.springframework.integration.filter.ExpressionEvaluatingSelector;
import org.springframework.integration.filter.MessageFilter;
import org.springframework.integration.filter.MethodInvokingSelector;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.transformer.ExpressionEvaluatingTransformer;
import org.springframework.integration.transformer.GenericTransformer;
import org.springframework.integration.transformer.MessageTransformingHandler;
@@ -117,9 +116,6 @@ public final class IntegrationFlowBuilder {
if (this.currentComponent instanceof MessageProducer) {
((MessageProducer) this.currentComponent).setOutputChannel(outputChannel);
}
if (this.currentComponent instanceof AbstractReplyProducingMessageHandler) {
((AbstractReplyProducingMessageHandler) this.currentComponent).setOutputChannel(outputChannel);
}
else if (this.currentComponent instanceof SourcePollingChannelAdapterFactoryBean) {
((SourcePollingChannelAdapterFactoryBean) this.currentComponent).setOutputChannel(outputChannel);
}

45
spring-integration-java-dsl/src/test/java/org/springframework/integration/dsl/test/IntegrationFlowTests.java
View File

@@ -22,7 +22,6 @@ import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import org.aopalliance.aop.Advice;
@@ -35,8 +34,11 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.Lifecycle;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.MessageDispatchingException;
import org.springframework.integration.annotation.MessageEndpoint;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
@@ -122,16 +124,6 @@ public class IntegrationFlowTests {
@EnableIntegration
public static class ContextConfiguration {
@Bean
public DirectChannelSpec inputChannel() {
return MessageChannels.direct();
}
@Bean
public QueueChannelSpec successChannel() {
return MessageChannels.queue();
}
@Bean
public MessageSource<?> integerMessageSource() {
MethodInvokingMessageSource source = new MethodInvokingMessageSource();
@@ -156,37 +148,50 @@ public class IntegrationFlowTests {
}
@Configuration
@ComponentScan
public static class ContextConfiguration2 {
@Autowired
@Qualifier("inputChannel")
private DirectChannel inputChannel;
@Bean
public QueueChannelSpec successChannel() {
return MessageChannels.queue();
}
@Autowired
@Qualifier("successChannel")
private PollableChannel successChannel;
@Bean
public DirectChannelSpec inputChannel() {
return MessageChannels.direct();
}
@Bean
public Advice expressionAdvice() {
ExpressionEvaluatingRequestHandlerAdvice advice = new ExpressionEvaluatingRequestHandlerAdvice();
advice.setOnSuccessExpression("payload");
advice.setSuccessChannel(this.successChannel);
advice.setSuccessChannel(this.successChannel().get());
return advice;
}
@Bean
public IntegrationFlow flow2() {
return IntegrationFlows.from(this.inputChannel)
return IntegrationFlows.from(this.inputChannel())
.filter(p -> p instanceof String, c -> c.id("filter"))
.<String, Integer>transform(Integer::parseInt)
.transform(new PayloadSerializingTransformer(),
c -> c.autoStartup(false).id("payloadSerializingTransformer"))
.channel(MessageChannels.queue(new SimpleMessageStore(), "fooQueue"))
.transform(new PayloadDeserializingTransformer())
.channel(MessageChannels.executor("executor", Executors.newCachedThreadPool()))
.channel(MessageChannels.publishSubscribe("publishSubscribeChannel"))
.transform((Integer p) -> p * 2, c -> c.advice(this.expressionAdvice()))
.get();
}
}
@MessageEndpoint
public static class AnnotationTestService {
@ServiceActivator(inputChannel = "publishSubscribeChannel")
public void handle(Object payload) {
assertEquals(100, payload);
}
}
}