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

INT-4252 IntegrationFlow: Allow Custom Bean Names

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

To allow to provide arbitrary bean names for the intermediate components
in the `IntegrationFlow` change `ComponentsRegistration` to return
a `Map<Object, String>` instead of raw `Collection<Object>`

* Use the value from that map in the `IntegrationFlowBeanPostProcessor` as a
fallback option before walking into bean name generation
* Provide `containerId` option for the AMQP DSL components

* Revert `@AfterClass` in the `AmqpTests`
* Expose `id()` for the `JmsDestinationAccessorSpec`
* Register `ListenerContainer` and `JmsTemplate`
as bean via `ComponentRegistration` in the particular JMS Java DSL components
* Fix `Jms` factory to populate the `JmsDefaultListenerContainerSpec`
for the `messageDrivenChannelAdapter()` without class specified

* Refactor AMQP DSL Inbound components to deal with the new ContainerSpec API
* Remove `containerId()` in favor of `id()` in the ContainerSpec
This commit is contained in:
Artem Bilan
2017-05-16 17:33:02 -04:00
committed by Gary Russell
parent 2b99c191f6
commit ea89682368
50 changed files with 396 additions and 267 deletions
Download Patch File Download Diff File Expand all files Collapse all files

140
spring-integration-core/src/main/java/org/springframework/integration/config/dsl/IntegrationFlowBeanPostProcessor.java
View File

@@ -16,10 +16,10 @@
package org.springframework.integration.config.dsl;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import org.springframework.beans.BeansException;
@@ -130,45 +130,44 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
}
}
private Object processStandardIntegrationFlow(StandardIntegrationFlow flow, String beanName) {
String flowNamePrefix = beanName + ".";
private Object processStandardIntegrationFlow(StandardIntegrationFlow flow, String flowBeanName) {
String flowNamePrefix = flowBeanName + ".";
int subFlowNameIndex = 0;
int channelNameIndex = 0;
boolean registerSingleton = flow.isRegisterComponents();
List<Object> integrationComponents = new ArrayList<>(flow.getIntegrationComponents());
for (int i = 0; i < integrationComponents.size(); i++) {
Object component = integrationComponents.get(i);
Map<Object, String> integrationComponents = flow.getIntegrationComponents();
Map<Object, String> targetIntegrationComponents = new LinkedHashMap<>(integrationComponents.size());
for (Map.Entry<Object, String> entry : integrationComponents.entrySet()) {
Object component = entry.getKey();
if (component instanceof ConsumerEndpointSpec) {
ConsumerEndpointSpec<?, ?> endpointSpec = (ConsumerEndpointSpec<?, ?>) component;
MessageHandler messageHandler = endpointSpec.get().getT2();
ConsumerEndpointFactoryBean endpoint = endpointSpec.get().getT1();
String id = endpointSpec.getId();
Collection<?> messageHandlers = this.beanFactory.getBeansOfType(messageHandler.getClass(), false,
false).values();
if (id == null) {
id = generateBeanName(endpoint, entry.getValue());
}
Collection<?> messageHandlers =
this.beanFactory.getBeansOfType(messageHandler.getClass(), false, false)
.values();
if (!messageHandlers.contains(messageHandler)) {
String handlerBeanName = generateBeanName(messageHandler);
String[] handlerAlias = id != null
? new String[] { id + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX }
: null;
String[] handlerAlias = new String[] { id + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX };
registerComponent(messageHandler, handlerBeanName, beanName, registerSingleton);
if (handlerAlias != null) {
for (String alias : handlerAlias) {
this.beanFactory.registerAlias(handlerBeanName, alias);
}
registerComponent(messageHandler, handlerBeanName, flowBeanName, registerSingleton);
for (String alias : handlerAlias) {
this.beanFactory.registerAlias(handlerBeanName, alias);
}
}
String endpointBeanName = id;
if (endpointBeanName == null) {
endpointBeanName = generateBeanName(endpoint);
}
registerComponent(endpoint, endpointBeanName, beanName, registerSingleton);
integrationComponents.set(i, endpoint);
registerComponent(endpoint, id, flowBeanName, registerSingleton);
targetIntegrationComponents.put(endpoint, id);
}
else {
Collection<?> values = this.beanFactory.getBeansOfType(component.getClass(), false, false).values();
@@ -176,17 +175,21 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
if (component instanceof AbstractMessageChannel) {
String channelBeanName = ((AbstractMessageChannel) component).getComponentName();
if (channelBeanName == null) {
channelBeanName = flowNamePrefix + "channel" +
BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
channelBeanName = entry.getValue();
if (channelBeanName == null) {
channelBeanName = flowNamePrefix + "channel" +
BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
}
}
registerComponent(component, channelBeanName, beanName, registerSingleton);
registerComponent(component, channelBeanName, flowBeanName, registerSingleton);
targetIntegrationComponents.put(component, channelBeanName);
}
else if (component instanceof MessageChannelReference) {
String channelBeanName = ((MessageChannelReference) component).getName();
if (!this.beanFactory.containsBean(channelBeanName)) {
DirectChannel directChannel = new DirectChannel();
registerComponent(directChannel, channelBeanName, beanName, registerSingleton);
integrationComponents.set(i, directChannel);
registerComponent(directChannel, channelBeanName, flowBeanName, registerSingleton);
targetIntegrationComponents.put(directChannel, channelBeanName);
}
}
else if (component instanceof FixedSubscriberChannel) {
@@ -196,26 +199,29 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
channelBeanName = flowNamePrefix + "channel" +
BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + channelNameIndex++;
}
registerComponent(component, channelBeanName, beanName, registerSingleton);
registerComponent(component, channelBeanName, flowBeanName, registerSingleton);
targetIntegrationComponents.put(component, channelBeanName);
}
else if (component instanceof SourcePollingChannelAdapterSpec) {
SourcePollingChannelAdapterSpec spec = (SourcePollingChannelAdapterSpec) component;
Collection<Object> componentsToRegister = spec.getComponentsToRegister();
Map<Object, String> componentsToRegister = spec.getComponentsToRegister();
if (!CollectionUtils.isEmpty(componentsToRegister)) {
componentsToRegister.stream()
.filter(o -> !this.beanFactory.getBeansOfType(o.getClass(), false, false)
.values()
.contains(o))
.forEach(o -> registerComponent(o, generateBeanName(o))
);
componentsToRegister.entrySet()
.stream()
.filter(o ->
!this.beanFactory.getBeansOfType(o.getKey().getClass(), false, false)
.values()
.contains(o.getKey()))
.forEach(o ->
registerComponent(o.getKey(), generateBeanName(o.getKey(), o.getValue())));
}
SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.get().getT1();
String id = spec.getId();
if (!StringUtils.hasText(id)) {
id = generateBeanName(pollingChannelAdapterFactoryBean);
id = generateBeanName(pollingChannelAdapterFactoryBean, entry.getValue());
}
registerComponent(pollingChannelAdapterFactoryBean, id, beanName, registerSingleton);
integrationComponents.set(i, pollingChannelAdapterFactoryBean);
registerComponent(pollingChannelAdapterFactoryBean, id, flowBeanName, registerSingleton);
targetIntegrationComponents.put(pollingChannelAdapterFactoryBean, id);
MessageSource<?> messageSource = spec.get().getT2();
if (!this.beanFactory.getBeansOfType(messageSource.getClass(), false, false)
@@ -226,25 +232,37 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
&& ((NamedComponent) messageSource).getComponentName() != null) {
messageSourceId = ((NamedComponent) messageSource).getComponentName();
}
registerComponent(messageSource, messageSourceId, beanName, registerSingleton);
registerComponent(messageSource, messageSourceId, flowBeanName, registerSingleton);
}
}
else if (component instanceof StandardIntegrationFlow) {
String subFlowBeanName = flowNamePrefix + "subFlow" +
BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + subFlowNameIndex++;
registerComponent(component, subFlowBeanName, beanName, registerSingleton);
String subFlowBeanName =
entry.getValue() != null
? entry.getValue()
: flowNamePrefix + "subFlow" +
BeanFactoryUtils.GENERATED_BEAN_NAME_SEPARATOR + subFlowNameIndex++;
registerComponent(component, subFlowBeanName, flowBeanName, registerSingleton);
targetIntegrationComponents.put(component, subFlowBeanName);
}
else if (component instanceof AnnotationGatewayProxyFactoryBean) {
registerComponent(component, flowNamePrefix + "gateway", beanName, registerSingleton);
String gatewayId = entry.getValue() != null
? entry.getValue()
: flowNamePrefix + "gateway";
registerComponent(component, gatewayId, flowBeanName, registerSingleton);
targetIntegrationComponents.put(component, gatewayId);
}
else {
String generateBeanName = generateBeanName(component);
registerComponent(component, generateBeanName, beanName, registerSingleton);
String generateBeanName = generateBeanName(component, entry.getValue());
registerComponent(component, generateBeanName, flowBeanName, registerSingleton);
}
}
else {
targetIntegrationComponents.put(entry.getKey(), entry.getValue());
}
}
}
flow.setIntegrationComponents(integrationComponents);
flow.setIntegrationComponents(targetIntegrationComponents);
return flow;
}
@@ -256,15 +274,21 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
}
private void processIntegrationComponentSpec(IntegrationComponentSpec<?, ?> bean) {
registerComponent(bean.get(), generateBeanName(bean.get()), null, false);
registerComponent(bean.get(), generateBeanName(bean.get(), bean.getId()), null, false);
if (bean instanceof ComponentsRegistration) {
Collection<Object> componentsToRegister = ((ComponentsRegistration) bean).getComponentsToRegister();
Map<Object, String> componentsToRegister = ((ComponentsRegistration) bean).getComponentsToRegister();
if (!CollectionUtils.isEmpty(componentsToRegister)) {
componentsToRegister.stream()
.filter(component -> !this.beanFactory.getBeansOfType(component.getClass(), false, false)
.values()
.contains(component))
.forEach(component -> registerComponent(component, generateBeanName(component)));
componentsToRegister.entrySet()
.stream()
.filter(component ->
!this.beanFactory.getBeansOfType(component.getKey().getClass(), false, false)
.values()
.contains(component.getKey()))
.forEach(component ->
registerComponent(component.getKey(),
generateBeanName(component.getKey(), component.getValue())));
}
}
}
@@ -293,9 +317,17 @@ public class IntegrationFlowBeanPostProcessor implements BeanPostProcessor, Bean
}
private String generateBeanName(Object instance) {
return generateBeanName(instance, null);
}
private String generateBeanName(Object instance, String fallbackId) {
if (instance instanceof NamedComponent && ((NamedComponent) instance).getComponentName() != null) {
return ((NamedComponent) instance).getComponentName();
}
else if (fallbackId != null) {
return fallbackId;
}
String generatedBeanName = instance.getClass().getName();
String id = generatedBeanName;
int counter = -1;

2
spring-integration-core/src/main/java/org/springframework/integration/dsl/AbstractRouterSpec.java
View File

@@ -96,7 +96,7 @@ public class AbstractRouterSpec<S extends AbstractRouterSpec<S, R>, R extends Ab
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
subFlow.configure(flowBuilder);
this.componentsToRegister.add(flowBuilder);
this.componentsToRegister.put(flowBuilder, null);
return defaultOutputChannel(channel);
}

6
spring-integration-core/src/main/java/org/springframework/integration/dsl/ComponentsRegistration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -16,7 +16,7 @@
package org.springframework.integration.dsl;
import java.util.Collection;
import java.util.Map;
/**
* The marker interface for the {@link IntegrationComponentSpec} implementation,
@@ -33,6 +33,6 @@ import java.util.Collection;
@FunctionalInterface
public interface ComponentsRegistration {
Collection<Object> getComponentsToRegister();
Map<Object, String> getComponentsToRegister();
}

2
spring-integration-core/src/main/java/org/springframework/integration/dsl/ConsumerEndpointSpec.java
View File

@@ -149,7 +149,7 @@ public abstract class ConsumerEndpointSpec<S extends ConsumerEndpointSpec<S, H>,
*/
public S transactional(boolean handleMessageAdvice) {
TransactionInterceptor transactionInterceptor = new TransactionInterceptorBuilder(handleMessageAdvice).build();
this.componentsToRegister.add(transactionInterceptor);
this.componentsToRegister.put(transactionInterceptor, null);
return transactional(transactionInterceptor);
}

12
spring-integration-core/src/main/java/org/springframework/integration/dsl/EndpointSpec.java
View File

@@ -16,8 +16,8 @@
package org.springframework.integration.dsl;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.function.Function;
import org.springframework.beans.factory.BeanNameAware;
@@ -46,7 +46,7 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
extends IntegrationComponentSpec<S, Tuple2<F, H>>
implements ComponentsRegistration {
protected final Collection<Object> componentsToRegister = new ArrayList<>();
protected final Map<Object, String> componentsToRegister = new LinkedHashMap<>();
protected H handler;
@@ -87,9 +87,9 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
* @see PollerSpec
*/
public S poller(PollerSpec pollerMetadataSpec) {
Collection<Object> componentsToRegister = pollerMetadataSpec.getComponentsToRegister();
Map<Object, String> componentsToRegister = pollerMetadataSpec.getComponentsToRegister();
if (componentsToRegister != null) {
this.componentsToRegister.addAll(componentsToRegister);
this.componentsToRegister.putAll(componentsToRegister);
}
return poller(pollerMetadataSpec.get());
}
@@ -116,7 +116,7 @@ public abstract class EndpointSpec<S extends EndpointSpec<S, F, H>, F extends Be
public abstract S autoStartup(boolean autoStartup);
@Override
public Collection<Object> getComponentsToRegister() {
public Map<Object, String> getComponentsToRegister() {
return this.componentsToRegister.isEmpty()
? null
: this.componentsToRegister;

2
spring-integration-core/src/main/java/org/springframework/integration/dsl/EnricherSpec.java
View File

@@ -150,7 +150,7 @@ public class EnricherSpec extends ConsumerEndpointSpec<EnricherSpec, ContentEnri
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(requestChannel);
subFlow.configure(flowBuilder);
this.componentsToRegister.add(flowBuilder.get());
this.componentsToRegister.put(flowBuilder.get(), null);
return requestChannel(requestChannel);
}

2
spring-integration-core/src/main/java/org/springframework/integration/dsl/FilterEndpointSpec.java
View File

@@ -91,7 +91,7 @@ public final class FilterEndpointSpec extends ConsumerEndpointSpec<FilterEndpoin
DirectChannel channel = new DirectChannel();
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
discardFlow.configure(flowBuilder);
this.componentsToRegister.add(flowBuilder.get());
this.componentsToRegister.put(flowBuilder.get(), null);
return discardChannel(channel);
}

2
spring-integration-core/src/main/java/org/springframework/integration/dsl/HeaderEnricherSpec.java
View File

@@ -437,7 +437,7 @@ public class HeaderEnricherSpec extends ConsumerEndpointSpec<HeaderEnricherSpec,
headerEnricher.setShouldSkipNulls(this.shouldSkipNulls);
headerEnricher.setMessageProcessor(this.messageProcessor);
this.componentsToRegister.add(headerEnricher);
this.componentsToRegister.put(headerEnricher, null);
this.handler = new MessageTransformingHandler(headerEnricher);
return super.doGet();

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -65,7 +65,7 @@ public abstract class IntegrationFlowAdapter implements IntegrationFlow, SmartLi
IntegrationFlowDefinition<?> targetFlow = buildFlow();
Assert.state(targetFlow != null, "the 'buildFlow()' must not return null");
flow.integrationComponents.clear();
flow.integrationComponents.addAll(targetFlow.integrationComponents);
flow.integrationComponents.putAll(targetFlow.integrationComponents);
this.targetIntegrationFlow = flow.get();
}

25
spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowDefinition.java
View File

@@ -16,9 +16,8 @@
package org.springframework.integration.dsl;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -120,7 +119,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
private static final Set<MessageProducer> REFERENCED_REPLY_PRODUCERS = new HashSet<>();
protected final Set<Object> integrationComponents = new LinkedHashSet<>();
protected final Map<Object, String> integrationComponents = new LinkedHashMap<>();
protected MessageChannel currentMessageChannel;
@@ -134,13 +133,13 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
B addComponent(Object component) {
this.integrationComponents.add(component);
this.integrationComponents.put(component, null);
return _this();
}
B addComponents(Collection<Object> components) {
B addComponents(Map<Object, String> components) {
if (components != null) {
this.integrationComponents.addAll(components);
this.integrationComponents.putAll(components);
}
return _this();
}
@@ -1974,9 +1973,10 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
BridgeHandler bridgeHandler = new BridgeHandler();
boolean registerSubflowBridge = false;
Collection<Object> componentsToRegister = routerSpec.getComponentsToRegister();
Map<Object, String> componentsToRegister = routerSpec.getComponentsToRegister();
if (!CollectionUtils.isEmpty(componentsToRegister)) {
for (Object component : componentsToRegister) {
for (Map.Entry<Object, String> entry : componentsToRegister.entrySet()) {
Object component = entry.getKey();
if (component instanceof IntegrationFlowDefinition) {
IntegrationFlowDefinition<?> flowBuilder = (IntegrationFlowDefinition<?>) component;
if (flowBuilder.isOutputChannelRequired()) {
@@ -1986,7 +1986,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
addComponent(flowBuilder.get());
}
else {
addComponent(component);
this.integrationComponents.put(component, entry.getValue());
}
}
}
@@ -2621,7 +2621,7 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
private B registerOutputChannelIfCan(MessageChannel outputChannel) {
if (!(outputChannel instanceof FixedSubscriberChannelPrototype)) {
this.integrationComponents.add(outputChannel);
this.integrationComponents.put(outputChannel, null);
if (this.currentComponent != null) {
String channelName = null;
if (outputChannel instanceof MessageChannelReference) {
@@ -2718,7 +2718,10 @@ public abstract class IntegrationFlowDefinition<B extends IntegrationFlowDefinit
}
if (this.implicitChannel) {
Optional<Object> lastComponent = this.integrationComponents.stream().reduce((first, second) -> second);
Optional<Object> lastComponent =
this.integrationComponents.keySet()
.stream()
.reduce((first, second) -> second);
if (lastComponent.get() instanceof WireTapSpec) {
channel(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME);
}

16
spring-integration-core/src/main/java/org/springframework/integration/dsl/PollerSpec.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -16,11 +16,11 @@
package org.springframework.integration.dsl;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import org.aopalliance.aop.Advice;
@@ -49,7 +49,7 @@ public final class PollerSpec extends IntegrationComponentSpec<PollerSpec, Polle
private final List<Advice> adviceChain = new LinkedList<>();
private final Collection<Object> componentsToRegister = new ArrayList<>();
private final Map<Object, String> componentsToRegister = new LinkedHashMap<>();
PollerSpec(Trigger trigger) {
this.target = new PollerMetadata();
@@ -92,7 +92,7 @@ public final class PollerSpec extends IntegrationComponentSpec<PollerSpec, Polle
public PollerSpec errorChannel(MessageChannel errorChannel) {
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
errorHandler.setDefaultErrorChannel(errorChannel);
this.componentsToRegister.add(errorHandler);
this.componentsToRegister.put(errorHandler, null);
return errorHandler(errorHandler);
}
@@ -106,7 +106,7 @@ public final class PollerSpec extends IntegrationComponentSpec<PollerSpec, Polle
public PollerSpec errorChannel(String errorChannelName) {
MessagePublishingErrorHandler errorHandler = new MessagePublishingErrorHandler();
errorHandler.setDefaultErrorChannelName(errorChannelName);
this.componentsToRegister.add(errorHandler);
this.componentsToRegister.put(errorHandler, null);
return errorHandler(errorHandler);
}
@@ -163,7 +163,7 @@ public final class PollerSpec extends IntegrationComponentSpec<PollerSpec, Polle
*/
public PollerSpec transactional() {
TransactionInterceptor transactionInterceptor = new TransactionInterceptorBuilder().build();
this.componentsToRegister.add(transactionInterceptor);
this.componentsToRegister.put(transactionInterceptor, null);
return transactional(transactionInterceptor);
}
@@ -193,7 +193,7 @@ public final class PollerSpec extends IntegrationComponentSpec<PollerSpec, Polle
}
@Override
public Collection<Object> getComponentsToRegister() {
public Map<Object, String> getComponentsToRegister() {
return this.componentsToRegister;
}

17
spring-integration-core/src/main/java/org/springframework/integration/dsl/PublishSubscribeSpec.java
View File

@@ -16,9 +16,8 @@
package org.springframework.integration.dsl;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.Executor;
import org.springframework.integration.dsl.channel.PublishSubscribeChannelSpec;
@@ -30,7 +29,7 @@ import org.springframework.integration.dsl.channel.PublishSubscribeChannelSpec;
*/
public class PublishSubscribeSpec extends PublishSubscribeChannelSpec<PublishSubscribeSpec> {
private final List<Object> subscriberFlows = new ArrayList<>();
private final Map<Object, String> subscriberFlows = new LinkedHashMap<>();
PublishSubscribeSpec() {
super();
@@ -50,15 +49,15 @@ public class PublishSubscribeSpec extends PublishSubscribeChannelSpec<PublishSub
IntegrationFlows.from(this.channel)
.bridge();
flow.configure(flowBuilder);
this.subscriberFlows.add(flowBuilder.get());
this.subscriberFlows.put(flowBuilder.get(), null);
return _this();
}
@Override
public Collection<Object> getComponentsToRegister() {
List<Object> objects = new ArrayList<Object>();
objects.addAll(super.getComponentsToRegister());
objects.addAll(this.subscriberFlows);
public Map<Object, String> getComponentsToRegister() {
Map<Object, String> objects = new LinkedHashMap<>();
objects.putAll(super.getComponentsToRegister());
objects.putAll(this.subscriberFlows);
return objects;
}

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

@@ -16,7 +16,7 @@
package org.springframework.integration.dsl;
import java.util.Set;
import java.util.Map;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
@@ -35,7 +35,7 @@ class PublisherIntegrationFlow<T> extends StandardIntegrationFlow implements Pub
private final Publisher<Message<T>> delegate;
PublisherIntegrationFlow(Set<Object> integrationComponents, Publisher<Message<T>> publisher) {
PublisherIntegrationFlow(Map<Object, String> integrationComponents, Publisher<Message<T>> publisher) {
super(integrationComponents);
this.delegate = publisher;
}

6
spring-integration-core/src/main/java/org/springframework/integration/dsl/RecipientListRouterSpec.java
View File

@@ -76,7 +76,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
public RecipientListRouterSpec recipient(String channelName, Expression expression) {
ExpressionEvaluatingSelector selector = new ExpressionEvaluatingSelector(expression);
this.handler.addRecipient(channelName, selector);
this.componentsToRegister.add(selector);
this.componentsToRegister.put(selector, null);
return _this();
}
@@ -146,7 +146,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
if (expression != null) {
ExpressionEvaluatingSelector selector = new ExpressionEvaluatingSelector(expression);
this.handler.addRecipient(channel, selector);
this.componentsToRegister.add(selector);
this.componentsToRegister.put(selector, null);
}
else {
this.handler.addRecipient(channel);
@@ -247,7 +247,7 @@ public class RecipientListRouterSpec extends AbstractRouterSpec<RecipientListRou
DirectChannel channel = new DirectChannel();
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
subFlow.configure(flowBuilder);
this.componentsToRegister.add(flowBuilder.get());
this.componentsToRegister.put(flowBuilder.get(), null);
return channel;
}

7
spring-integration-core/src/main/java/org/springframework/integration/dsl/RouterSpec.java
View File

@@ -16,7 +16,6 @@
package org.springframework.integration.dsl;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -161,19 +160,19 @@ public final class RouterSpec<K, R extends AbstractMappingMessageRouter>
IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(channel);
subFlow.configure(flowBuilder);
this.componentsToRegister.add(flowBuilder);
this.componentsToRegister.put(flowBuilder, null);
this.mappingProvider.addMapping(key, channel);
return _this();
}
@Override
public Collection<Object> getComponentsToRegister() {
public Map<Object, String> getComponentsToRegister() {
// The 'mappingProvider' must be added to the 'componentsToRegister' in the end to
// let all other components to be registered before the 'RouterMappingProvider.onInit()' logic.
if (!this.mappingProviderRegistered) {
if (!this.mappingProvider.mapping.isEmpty()) {
this.componentsToRegister.add(this.mappingProvider);
this.componentsToRegister.put(this.mappingProvider, null);
}
this.mappingProviderRegistered = true;
}

21
spring-integration-core/src/main/java/org/springframework/integration/dsl/StandardIntegrationFlow.java
View File

@@ -16,10 +16,12 @@
package org.springframework.integration.dsl;
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.Set;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.context.SmartLifecycle;
@@ -63,7 +65,7 @@ import org.springframework.context.SmartLifecycle;
*/
public class StandardIntegrationFlow implements IntegrationFlow, SmartLifecycle {
private final List<Object> integrationComponents;
private final Map<Object, String> integrationComponents;
private final List<SmartLifecycle> lifecycles = new LinkedList<>();
@@ -71,8 +73,8 @@ public class StandardIntegrationFlow implements IntegrationFlow, SmartLifecycle
private boolean running;
StandardIntegrationFlow(Set<Object> integrationComponents) {
this.integrationComponents = new LinkedList<>(integrationComponents);
StandardIntegrationFlow(Map<Object, String> integrationComponents) {
this.integrationComponents = new LinkedHashMap<>(integrationComponents);
}
//TODO Figure out some custom DestinationResolver when we don't register singletons - remove NOSONAR above when done
@@ -84,13 +86,13 @@ public class StandardIntegrationFlow implements IntegrationFlow, SmartLifecycle
return this.registerComponents;
}
public void setIntegrationComponents(List<Object> integrationComponents) {
public void setIntegrationComponents(Map<Object, String> integrationComponents) {
this.integrationComponents.clear();
this.integrationComponents.addAll(integrationComponents);
this.integrationComponents.putAll(integrationComponents);
}
public List<Object> getIntegrationComponents() {
return this.integrationComponents;
public Map<Object, String> getIntegrationComponents() {
return Collections.unmodifiableMap(this.integrationComponents);
}
@Override
@@ -101,7 +103,8 @@ public class StandardIntegrationFlow implements IntegrationFlow, SmartLifecycle
@Override
public void start() {
if (!this.running) {
ListIterator<Object> iterator = this.integrationComponents.listIterator(this.integrationComponents.size());
List<Object> components = new LinkedList<>(this.integrationComponents.keySet());
ListIterator<Object> iterator = components.listIterator(this.integrationComponents.size());
this.lifecycles.clear();
while (iterator.hasPrevious()) {
Object component = iterator.previous();

9
spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/MessageChannelSpec.java
View File

@@ -18,9 +18,10 @@ package org.springframework.integration.dsl.channel;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import org.springframework.integration.channel.AbstractMessageChannel;
import org.springframework.integration.channel.interceptor.WireTap;
@@ -44,7 +45,7 @@ public abstract class MessageChannelSpec<S extends MessageChannelSpec<S, C>, C e
extends IntegrationComponentSpec<S, C>
implements ComponentsRegistration {
private final List<Object> componentsToRegister = new ArrayList<>();
private final Map<Object, String> componentsToRegister = new LinkedHashMap<>();
private final List<Class<?>> datatypes = new ArrayList<>();
@@ -108,7 +109,7 @@ public abstract class MessageChannelSpec<S extends MessageChannelSpec<S, C>, C e
*/
public S wireTap(WireTapSpec wireTapSpec) {
WireTap interceptor = wireTapSpec.get();
this.componentsToRegister.add(interceptor);
this.componentsToRegister.put(interceptor, null);
return interceptor(interceptor);
}
@@ -118,7 +119,7 @@ public abstract class MessageChannelSpec<S extends MessageChannelSpec<S, C>, C e
}
@Override
public Collection<Object> getComponentsToRegister() {
public Map<Object, String> getComponentsToRegister() {
return this.componentsToRegister;
}

6
spring-integration-core/src/main/java/org/springframework/integration/dsl/channel/WireTapSpec.java
View File

@@ -16,8 +16,8 @@
package org.springframework.integration.dsl.channel;
import java.util.Collection;
import java.util.Collections;
import java.util.Map;
import org.springframework.expression.Expression;
import org.springframework.integration.channel.interceptor.WireTap;
@@ -101,9 +101,9 @@ public class WireTapSpec extends IntegrationComponentSpec<WireTapSpec, WireTap>
}
@Override
public Collection<Object> getComponentsToRegister() {
public Map<Object, String> getComponentsToRegister() {
if (this.selector != null) {
return Collections.singleton(this.selector);
return Collections.singletonMap(this.selector, null);
}
else {
return null;

5
spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowRegistration.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2016 the original author or authors.
* Copyright 2016-2017 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.
@@ -30,6 +30,7 @@ import org.springframework.messaging.MessageChannel;
* and provide an API for some useful {@link IntegrationFlow} options and its lifecycle.
*
* @author Artem Bilan
*
* @since 5.0
*
* @see IntegrationFlowContext
@@ -80,7 +81,7 @@ public class IntegrationFlowRegistration {
if (this.inputChannel == null) {
if (this.integrationFlow instanceof StandardIntegrationFlow) {
StandardIntegrationFlow integrationFlow = (StandardIntegrationFlow) this.integrationFlow;
Object next = integrationFlow.getIntegrationComponents().iterator().next();
Object next = integrationFlow.getIntegrationComponents().keySet().iterator().next();
if (next instanceof MessageChannel) {
this.inputChannel = (MessageChannel) next;
}