From 48df4f211c8993ce0d26fa68217b1848ac9491f6 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 21 May 2018 16:37:45 -0400 Subject: [PATCH] INT-4473: Support prefix bean names with flow id JIRA: https://jira.spring.io/browse/INT-4473 Previously, dynamic registration of integration flows with components configured with the same `id` would fail with duplicate bean names. Add `useFlowIdAsPrefix()` to the registration builder to enable the option. Then, in the BPP, check the flag before naming the beans. **cherry-pick to 5.0.x** * Polishing according PR comments * Widen `flowNamePrefix` responsibility in the `IntegrationFlowBeanPostProcessor` since we may have many other components in the dynamic flow with the same id, not only consumer endpoints # Conflicts: # spring-integration-core/src/main/java/org/springframework/integration/config/dsl/IntegrationFlowBeanPostProcessor.java # spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowContext.java # spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java # src/reference/asciidoc/dsl.adoc --- .../dsl/IntegrationFlowBeanPostProcessor.java | 41 ++++++++++++++----- .../dsl/context/IntegrationFlowContext.java | 38 +++++++++++++++++ .../dsl/manualflow/ManualFlowTests.java | 4 +- src/reference/asciidoc/dsl.adoc | 40 +++++++++++++++++- 4 files changed, 110 insertions(+), 13 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/dsl/IntegrationFlowBeanPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/dsl/IntegrationFlowBeanPostProcessor.java index e0f3aa9a2e..c3f792ef64 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/dsl/IntegrationFlowBeanPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/dsl/IntegrationFlowBeanPostProcessor.java @@ -49,6 +49,7 @@ import org.springframework.integration.dsl.IntegrationFlowBuilder; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.SourcePollingChannelAdapterSpec; import org.springframework.integration.dsl.StandardIntegrationFlow; +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; @@ -72,6 +73,8 @@ public class IntegrationFlowBeanPostProcessor private ConfigurableListableBeanFactory beanFactory; + private IntegrationFlowContext flowContext; + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { Assert.isInstanceOf(ConfigurableListableBeanFactory.class, beanFactory, @@ -80,6 +83,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 @@ -105,7 +110,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."); } } } @@ -113,6 +118,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; @@ -128,7 +134,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 = @@ -190,13 +199,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); @@ -241,7 +256,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); } @@ -306,15 +323,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 3102e12f2e..a95ecef97b 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 @@ -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; /** * A public API for dynamic (manual) registration of {@link IntegrationFlow}s, @@ -70,6 +71,8 @@ public final class IntegrationFlowContext implements BeanFactoryAware { private final Map registry = new ConcurrentHashMap<>(); + private final Map useFlowIdAsPrefix = new ConcurrentHashMap<>(); + private final Lock registerFlowsLock = new ReentrantLock(); private ConfigurableListableBeanFactory beanFactory; @@ -96,6 +99,15 @@ public final class IntegrationFlowContext implements BeanFactoryAware { return new IntegrationFlowRegistrationBuilder(integrationFlow); } + /** + * 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 + */ + public boolean isUseIdAsPrefix(String flowId) { + return Boolean.TRUE.equals(this.useFlowIdAsPrefix.get(flowId)); + } private void register(IntegrationFlowRegistrationBuilder builder) { IntegrationFlow integrationFlow = builder.integrationFlowRegistration.getIntegrationFlow(); @@ -228,8 +240,12 @@ public final class IntegrationFlowContext implements BeanFactoryAware { } /** + * @author Gary Russell + * @since 5.1 + * * A Builder pattern implementation for the options to register {@link IntegrationFlow} * in the application context. + */ public final class IntegrationFlowRegistrationBuilder { @@ -239,6 +255,8 @@ public final class IntegrationFlowContext implements BeanFactoryAware { private boolean autoStartup = true; + private boolean idAsPrefix; + IntegrationFlowRegistrationBuilder(IntegrationFlow integrationFlow) { this.integrationFlowRegistration = new IntegrationFlowRegistration(integrationFlow); this.integrationFlowRegistration.setBeanFactory(IntegrationFlowContext.this.beanFactory); @@ -293,6 +311,20 @@ public final class IntegrationFlowContext implements BeanFactoryAware { return this; } + /** + * 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 + */ + 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} @@ -300,6 +332,12 @@ public final class IntegrationFlowContext implements BeanFactoryAware { * @return the {@link IntegrationFlowRegistration} instance. */ 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) { + IntegrationFlowContext.this.useFlowIdAsPrefix.put(id, this.idAsPrefix); + } IntegrationFlowContext.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 b7e447ef8e..cced2f9fd1 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 @@ -170,12 +170,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(); @@ -184,6 +185,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 8bed5fedb1..3d38645454 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -578,7 +578,7 @@ The result of this definition is the same bunch of Integration components wired Only limitation is here, that this flow is started with named direct channel - `lambdaFlow.input`. And Lambda flow can't start from `MessageSource` or `MessageProducer`. -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. @@ -895,10 +895,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