diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java index fed2e6253f..39d24574d1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/IntegrationFlowBeanPostProcessor.java @@ -44,6 +44,7 @@ import org.springframework.integration.config.ConsumerEndpointFactoryBean; import org.springframework.integration.config.IntegrationConfigUtils; import org.springframework.integration.config.SourcePollingChannelAdapterFactoryBean; import org.springframework.integration.core.MessageSource; +import org.springframework.integration.dsl.context.IntegrationFlowContext; import org.springframework.integration.dsl.support.MessageChannelReference; import org.springframework.integration.gateway.AnnotationGatewayProxyFactoryBean; import org.springframework.integration.support.context.NamedComponent; @@ -67,6 +68,8 @@ public class IntegrationFlowBeanPostProcessor private ConfigurableListableBeanFactory beanFactory; + private IntegrationFlowContext flowContext; + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { Assert.isInstanceOf(ConfigurableListableBeanFactory.class, beanFactory, @@ -75,6 +78,8 @@ public class IntegrationFlowBeanPostProcessor ); this.beanFactory = (ConfigurableListableBeanFactory) beanFactory; + this.flowContext = this.beanFactory.getBean(IntegrationFlowContext.class); + Assert.notNull(this.flowContext, "There must be an IntegrationFlowContext in the application context"); } @Override @@ -100,7 +105,7 @@ public class IntegrationFlowBeanPostProcessor throw new BeanCreationNotAllowedException(beanName, "IntegrationFlows can not be scoped beans. " + "Any dependant beans are registered as singletons, meanwhile IntegrationFlow is just a " + "logical container for them. \n" + - "Consider to use [IntegrationFlowContext] for manual registration of IntegrationFlows."); + "Consider using [IntegrationFlowContext] for manual registration of IntegrationFlows."); } } } @@ -108,6 +113,7 @@ public class IntegrationFlowBeanPostProcessor private Object processStandardIntegrationFlow(StandardIntegrationFlow flow, String flowBeanName) { String flowNamePrefix = flowBeanName + "."; + boolean useFlowIdAsPrefix = this.flowContext.isUseIdAsPrefix(flowBeanName); int subFlowNameIndex = 0; int channelNameIndex = 0; @@ -123,7 +129,10 @@ public class IntegrationFlowBeanPostProcessor String id = endpointSpec.getId(); if (id == null) { - id = generateBeanName(endpoint, flowNamePrefix, entry.getValue()); + id = generateBeanName(endpoint, flowNamePrefix, entry.getValue(), useFlowIdAsPrefix); + } + else if (useFlowIdAsPrefix) { + id = flowNamePrefix + id; } Collection messageHandlers = @@ -185,13 +194,19 @@ public class IntegrationFlowBeanPostProcessor .contains(o.getKey())) .forEach(o -> registerComponent(o.getKey(), - generateBeanName(o.getKey(), flowNamePrefix, o.getValue()))); + generateBeanName(o.getKey(), flowNamePrefix, o.getValue(), + useFlowIdAsPrefix))); } SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.get().getT1(); String id = spec.getId(); - if (!StringUtils.hasText(id)) { - id = generateBeanName(pollingChannelAdapterFactoryBean, flowNamePrefix, entry.getValue()); + if (id == null) { + id = generateBeanName(pollingChannelAdapterFactoryBean, flowNamePrefix, entry.getValue(), + useFlowIdAsPrefix); } + else if (useFlowIdAsPrefix) { + id = flowNamePrefix + id; + } + registerComponent(pollingChannelAdapterFactoryBean, id, flowBeanName); targetIntegrationComponents.put(pollingChannelAdapterFactoryBean, id); @@ -236,7 +251,9 @@ public class IntegrationFlowBeanPostProcessor targetIntegrationComponents.put(component, gatewayId); } else { - String generatedBeanName = generateBeanName(component, flowNamePrefix, entry.getValue()); + String generatedBeanName = + generateBeanName(component, flowNamePrefix, entry.getValue(), useFlowIdAsPrefix); + registerComponent(component, generatedBeanName, flowBeanName); targetIntegrationComponents.put(component, generatedBeanName); } @@ -334,15 +351,19 @@ public class IntegrationFlowBeanPostProcessor } private String generateBeanName(Object instance, String prefix) { - return generateBeanName(instance, prefix, null); + return generateBeanName(instance, prefix, null, false); } - private String generateBeanName(Object instance, String prefix, String fallbackId) { + private String generateBeanName(Object instance, String prefix, String fallbackId, boolean useFlowIdAsPrefix) { if (instance instanceof NamedComponent && ((NamedComponent) instance).getComponentName() != null) { - return ((NamedComponent) instance).getComponentName(); + return useFlowIdAsPrefix + ? prefix + ((NamedComponent) instance).getComponentName() + : ((NamedComponent) instance).getComponentName(); } else if (fallbackId != null) { - return fallbackId; + return useFlowIdAsPrefix + ? prefix + fallbackId + : fallbackId; } String generatedBeanName = prefix + instance.getClass().getName(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java index 04ab20d9af..aa25dcefd7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java @@ -95,6 +95,16 @@ public interface IntegrationFlowContext { */ Map getRegistry(); + /** + * Return true to prefix flow bean names with the flow id and a period. + * @param flowId the flow id. + * @return true to use as a prefix. + * @since 5.0.6 + */ + default boolean isUseIdAsPrefix(String flowId) { + return false; + } + /** * @author Gary Russell * @since 5.1 @@ -209,6 +219,19 @@ public interface IntegrationFlowContext { */ IntegrationFlowRegistrationBuilder addBean(String name, Object bean); + /** + * Invoke this method to prefix bean names in the flow with the (required) flow id + * and a period. This is useful if you wish to register the same flow multiple times + * while retaining the ability to reference beans within the flow; adding the unique + * flow id to the bean name makes the name unique. + * @return the current builder instance. + * @see #id(String) + * @since 5.0.6 + */ + default IntegrationFlowRegistrationBuilder useFlowIdAsPrefix() { + return this; + } + /** * Register an {@link IntegrationFlow} and all the dependant and support components * in the application context and return an associated {@link IntegrationFlowRegistration} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java index ebf38736ca..a4ccdc3c61 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java @@ -37,6 +37,7 @@ import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.support.context.NamedComponent; import org.springframework.messaging.MessageChannel; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * Standard implementation of {@link IntegrationFlowContext}. @@ -51,6 +52,8 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont private final Map registry = new ConcurrentHashMap<>(); + private final Map useFlowIdAsPrefix = new ConcurrentHashMap<>(); + private final Lock registerFlowsLock = new ReentrantLock(); private ConfigurableListableBeanFactory beanFactory; @@ -78,6 +81,11 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont return new StandardIntegrationFlowRegistrationBuilder(integrationFlow); } + @Override + public boolean isUseIdAsPrefix(String flowId) { + return Boolean.TRUE.equals(this.useFlowIdAsPrefix.get(flowId)); + } + private void register(StandardIntegrationFlowRegistrationBuilder builder) { IntegrationFlow integrationFlow = builder.integrationFlowRegistration.getIntegrationFlow(); String flowId = builder.integrationFlowRegistration.getId(); @@ -224,6 +232,8 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont private boolean autoStartup = true; + private boolean idAsPrefix; + StandardIntegrationFlowRegistrationBuilder(IntegrationFlow integrationFlow) { this.integrationFlowRegistration = new StandardIntegrationFlowRegistration(integrationFlow); this.integrationFlowRegistration.setBeanFactory(StandardIntegrationFlowContext.this.beanFactory); @@ -282,6 +292,12 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont return this; } + @Override + public IntegrationFlowRegistrationBuilder useFlowIdAsPrefix() { + this.idAsPrefix = true; + return this; + } + /** * Register an {@link IntegrationFlow} and all the dependant and support components * in the application context and return an associated {@link IntegrationFlowRegistration} @@ -290,6 +306,12 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont */ @Override public IntegrationFlowRegistration register() { + String id = this.integrationFlowRegistration.getId(); + Assert.state(!this.idAsPrefix || StringUtils.hasText(id), + "An 'id' must be present to use 'useFlowIdAsPrefix'"); + if (this.idAsPrefix) { + StandardIntegrationFlowContext.this.useFlowIdAsPrefix.put(id, this.idAsPrefix); + } StandardIntegrationFlowContext.this.register(this); return this.integrationFlowRegistration; } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java index d89a3100d4..cbe73b7af5 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dsl/manualflow/ManualFlowTests.java @@ -171,12 +171,13 @@ public class ManualFlowTests { .fixedDelay(10) .maxMessagesPerPoll(1) .receiveTimeout(10))) - .handle(new BeanFactoryHandler()); + .handle(new BeanFactoryHandler(), e -> e.id("anId")); BeanFactoryHandler additionalBean = new BeanFactoryHandler(); IntegrationFlowRegistration flowRegistration = this.integrationFlowContext.registration(myFlow) .id(flowId) + .useFlowIdAsPrefix() .addBean(additionalBean) .register(); @@ -185,6 +186,7 @@ public class ManualFlowTests { BeanFactoryHandler.class); assertSame(additionalBean, bean); assertSame(this.beanFactory, bean.beanFactory); + bean = this.beanFactory.getBean(flowRegistration.getId() + "." + "anId.handler", BeanFactoryHandler.class); MessagingTemplate messagingTemplate = flowRegistration.getMessagingTemplate(); messagingTemplate.setReceiveTimeout(10000); diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 117bbcf120..a30204b220 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -580,7 +580,7 @@ And Lambda flow can't start from `MessageSource` or `MessageProducer`. Starting _version 5.1_, this kind of `IntegrationFlow` are wrapped to the proxy for exposing lifecycle control and provide access to the `inputChannel` of the internally associated `StandardIntegrationFlow`. -Starting with _version 5.0.5_, the generated bean names for the components in an `IntegrationFlow` include the flow bean followed by a dot as a prefix. +Starting with _version 5.0.6_, the generated bean names for the components in an `IntegrationFlow` include the flow bean followed by a dot as a prefix. For example the `ConsumerEndpointFactoryBean` for the `.transform("Hello "::concat)` in the sample above, will end up with te bean name like `lambdaFlow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0`. The `Transformer` implementation bean for that endpoint will have a bean name such as `lambdaFlow.org.springframework.integration.transformer.MethodInvokingTransformer#0`. These generated bean names are prepended with the flow id prefix for purposes such as parsing logs or grouping components together in some analysis tool, as well as to avoid a race condition when we concurrently register integration flows at runtime. @@ -933,10 +933,46 @@ Usually those additional beans are connection factories (AMQP, JMS, (S)FTP, TCP/ Such a dynamically registered `IntegrationFlow` and all its dependant beans can be removed afterwards using `IntegrationFlowRegistration.destroy()` callback. See `IntegrationFlowContext` JavaDocs for more information. -NOTE: Starting with _version 5.0.5_, all generated bean names in an `IntegrationFlow` definition are prepended with flow id as a prefix. +NOTE: Starting with _version 5.0.6_, all generated bean names in an `IntegrationFlow` definition are prepended with flow id as a prefix. It is recommended to always specify an explicit flow id, otherwise a synchronization barrier is initiated in the `IntegrationFlowContext` to generate the bean name for the `IntegrationFlow` and register its beans. We synchronize on these two operations to avoid a race condition when the same generated bean name may be used for different `IntegrationFlow` instances. +Also, starting with _version 5.0.6_, the registration builder API has a new method `useFlowIdAsPrefix()`. +This is useful if you wish to declare multiple instances of the same flow and avoid bean name collisions if components in the flows have the same id. + +For example: + +[source, java] +---- +private void registerFlows() { + IntegrationFlowRegistration flow1 = + this.flowContext.registration(buildFlow(1234)) + .id("tcp1") + .useFlowIdAsPrefix() + .register(); + + IntegrationFlowRegistration flow2 = + this.flowContext.registration(buildFlow(1235)) + .id("tcp2") + .useFlowIdAsPrefix() + .register(); +} + +private IntegrationFlow buildFlow(int port) { + return f -> f + .handle(Tcp.outboundGateway(Tcp.netClient("localhost", port) + .serializer(TcpCodecs.crlf()) + .deserializer(TcpCodecs.lengthHeader1()) + .id("client")) + .remoteTimeout(m -> 5000)) + .transform(Transformers.objectToString()); +} +---- + +In this case, the message handler for the first flow can be referenced with bean name `tcp1.client.handler`. + +NOTE: an `id` is required when using `useFlowIdAsPrefix()`. + [[java-dsl-gateway]] === IntegrationFlow as Gateway