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 6007d7b4c6..e0f3aa9a2e 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 @@ -128,7 +128,7 @@ public class IntegrationFlowBeanPostProcessor String id = endpointSpec.getId(); if (id == null) { - id = generateBeanName(endpoint, entry.getValue()); + id = generateBeanName(endpoint, flowNamePrefix, entry.getValue()); } Collection messageHandlers = @@ -136,13 +136,10 @@ public class IntegrationFlowBeanPostProcessor .values(); if (!messageHandlers.contains(messageHandler)) { - String handlerBeanName = generateBeanName(messageHandler); - String[] handlerAlias = new String[] { id + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX }; + String handlerBeanName = generateBeanName(messageHandler, flowNamePrefix); registerComponent(messageHandler, handlerBeanName, flowBeanName); - for (String alias : handlerAlias) { - this.beanFactory.registerAlias(handlerBeanName, alias); - } + this.beanFactory.registerAlias(handlerBeanName, id + IntegrationConfigUtils.HANDLER_ALIAS_SUFFIX); } registerComponent(endpoint, id, flowBeanName); @@ -192,12 +189,13 @@ public class IntegrationFlowBeanPostProcessor .values() .contains(o.getKey())) .forEach(o -> - registerComponent(o.getKey(), generateBeanName(o.getKey(), o.getValue()))); + registerComponent(o.getKey(), + generateBeanName(o.getKey(), flowNamePrefix, o.getValue()))); } SourcePollingChannelAdapterFactoryBean pollingChannelAdapterFactoryBean = spec.get().getT1(); String id = spec.getId(); if (!StringUtils.hasText(id)) { - id = generateBeanName(pollingChannelAdapterFactoryBean, entry.getValue()); + id = generateBeanName(pollingChannelAdapterFactoryBean, flowNamePrefix, entry.getValue()); } registerComponent(pollingChannelAdapterFactoryBean, id, flowBeanName); targetIntegrationComponents.put(pollingChannelAdapterFactoryBean, id); @@ -243,7 +241,7 @@ public class IntegrationFlowBeanPostProcessor targetIntegrationComponents.put(component, gatewayId); } else { - String generatedBeanName = generateBeanName(component, entry.getValue()); + String generatedBeanName = generateBeanName(component, flowNamePrefix, entry.getValue()); registerComponent(component, generatedBeanName, flowBeanName); targetIntegrationComponents.put(component, generatedBeanName); } @@ -307,11 +305,11 @@ public class IntegrationFlowBeanPostProcessor this.beanFactory.getBean(beanName); } - private String generateBeanName(Object instance) { - return generateBeanName(instance, null); + private String generateBeanName(Object instance, String prefix) { + return generateBeanName(instance, prefix, null); } - private String generateBeanName(Object instance, String fallbackId) { + private String generateBeanName(Object instance, String prefix, String fallbackId) { if (instance instanceof NamedComponent && ((NamedComponent) instance).getComponentName() != null) { return ((NamedComponent) instance).getComponentName(); } @@ -319,7 +317,7 @@ public class IntegrationFlowBeanPostProcessor return fallbackId; } - String generatedBeanName = instance.getClass().getName(); + String generatedBeanName = prefix + instance.getClass().getName(); String id = generatedBeanName; int counter = -1; while (counter == -1 || this.beanFactory.containsBean(id)) { 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 4cba85a25f..3102e12f2e 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 @@ -20,6 +20,9 @@ import java.util.Arrays; import java.util.Collections; import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.Lock; +import java.util.concurrent.locks.ReentrantLock; import org.springframework.beans.BeansException; import org.springframework.beans.factory.BeanFactory; @@ -65,7 +68,9 @@ import org.springframework.util.Assert; */ public final class IntegrationFlowContext implements BeanFactoryAware { - private final Map registry = new HashMap<>(); + private final Map registry = new ConcurrentHashMap<>(); + + private final Lock registerFlowsLock = new ReentrantLock(); private ConfigurableListableBeanFactory beanFactory; @@ -91,20 +96,33 @@ public final class IntegrationFlowContext implements BeanFactoryAware { return new IntegrationFlowRegistrationBuilder(integrationFlow); } + private void register(IntegrationFlowRegistrationBuilder builder) { IntegrationFlow integrationFlow = builder.integrationFlowRegistration.getIntegrationFlow(); String flowId = builder.integrationFlowRegistration.getId(); - if (flowId == null) { - flowId = generateBeanName(integrationFlow, null); - builder.id(flowId); + Lock registerBeanLock = null; + try { + if (flowId == null) { + registerBeanLock = this.registerFlowsLock; + registerBeanLock.lock(); + flowId = generateBeanName(integrationFlow, null); + builder.id(flowId); + } + else if (this.registry.containsKey(flowId)) { + throw new IllegalArgumentException("An IntegrationFlow '" + this.registry.get(flowId) + + "' with flowId '" + flowId + "' is already registered.\n" + + "An existing IntegrationFlowRegistration must be destroyed before overriding."); + } + + integrationFlow = (IntegrationFlow) registerBean(integrationFlow, flowId, null); } - else if (this.registry.containsKey(flowId)) { - throw new IllegalArgumentException("An IntegrationFlow '" + this.registry.get(flowId) + - "' with flowId '" + flowId + "' is already registered.\n" + - "An existing IntegrationFlowRegistration must be destroyed before overriding."); + finally { + if (registerBeanLock != null) { + registerBeanLock.unlock(); + } } - IntegrationFlow theFlow = (IntegrationFlow) registerBean(integrationFlow, flowId, null); - builder.integrationFlowRegistration.setIntegrationFlow(theFlow); + + builder.integrationFlowRegistration.setIntegrationFlow(integrationFlow); final String theFlowId = flowId; builder.additionalBeans.forEach((key, value) -> registerBean(key, value, theFlowId)); @@ -149,19 +167,26 @@ public final class IntegrationFlowContext implements BeanFactoryAware { * for provided {@code flowId} and clean up all the local cache for it. * @param flowId the bean name to destroy from */ - public synchronized void remove(String flowId) { + public void remove(String flowId) { if (this.registry.containsKey(flowId)) { IntegrationFlowRegistration flowRegistration = this.registry.remove(flowId); flowRegistration.stop(); - Arrays.stream(this.beanFactory.getDependentBeans(flowId)) - .forEach(((BeanDefinitionRegistry) this.beanFactory)::removeBeanDefinition); + BeanDefinitionRegistry beanDefinitionRegistry = (BeanDefinitionRegistry) this.beanFactory; - ((BeanDefinitionRegistry) this.beanFactory).removeBeanDefinition(flowId); + Arrays.stream(this.beanFactory.getDependentBeans(flowId)) + .forEach(beanName -> { + beanDefinitionRegistry.removeBeanDefinition(beanName); + // TODO until https://jira.spring.io/browse/SPR-16837 + Arrays.asList(beanDefinitionRegistry.getAliases(beanName)) + .forEach(beanDefinitionRegistry::removeAlias); + }); + + beanDefinitionRegistry.removeBeanDefinition(flowId); } else { - throw new IllegalStateException("Only manually registered IntegrationFlows can be removed. " - + "But [" + flowId + "] ins't one of them."); + throw new IllegalStateException("An IntegrationFlow with the id " + + "[" + flowId + "] doesn't exist in the registry."); } } 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 6adea417d9..b7e447ef8e 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 @@ -18,6 +18,7 @@ package org.springframework.integration.dsl.manualflow; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; +import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -27,9 +28,14 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; +import java.util.ArrayList; import java.util.Arrays; import java.util.Date; +import java.util.List; import java.util.Objects; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import java.util.function.Supplier; @@ -40,8 +46,11 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.BeanCreationNotAllowedException; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.DisposableBean; +import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.config.BeanFactoryPostProcessor; import org.springframework.beans.factory.config.ConfigurableBeanFactory; +import org.springframework.beans.factory.support.DefaultListableBeanFactory; import org.springframework.context.annotation.AnnotationConfigApplicationContext; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -62,6 +71,7 @@ import org.springframework.integration.dsl.context.IntegrationFlowRegistration; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.support.SmartLifecycleRoleController; +import org.springframework.integration.transformer.MessageTransformingHandler; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; @@ -91,7 +101,7 @@ public class ManualFlowTests { private IntegrationFlowContext integrationFlowContext; @Autowired - private BeanFactory beanFactory; + private ListableBeanFactory beanFactory; @Autowired private SmartLifecycleRoleController roleController; @@ -112,8 +122,10 @@ public class ManualFlowTests { IntegrationFlow flow = IntegrationFlows.from(producer) .channel(channel) .get(); - this.integrationFlowContext.registration(flow).register(); + IntegrationFlowRegistration flowRegistration = this.integrationFlowContext.registration(flow).register(); assertTrue(started.get()); + + flowRegistration.destroy(); } @Test @@ -137,15 +149,19 @@ public class ManualFlowTests { } MyProducerSpec spec = new MyProducerSpec(new MyProducer()); QueueChannel channel = new QueueChannel(); - IntegrationFlow flow = IntegrationFlows.from(spec.id("foo")) + IntegrationFlow flow = IntegrationFlows.from(spec.id("fooChannel")) .channel(channel) .get(); - this.integrationFlowContext.registration(flow).register(); + IntegrationFlowRegistration flowRegistration = this.integrationFlowContext.registration(flow).register(); assertTrue(started.get()); + + flowRegistration.destroy(); } @Test public void testManualFlowRegistration() throws InterruptedException { + String flowId = "testManualFlow"; + IntegrationFlow myFlow = f -> f .transform(String::toUpperCase) .channel(MessageChannels.queue()) @@ -159,6 +175,7 @@ public class ManualFlowTests { BeanFactoryHandler additionalBean = new BeanFactoryHandler(); IntegrationFlowRegistration flowRegistration = this.integrationFlowContext.registration(myFlow) + .id(flowId) .addBean(additionalBean) .register(); @@ -184,6 +201,8 @@ public class ManualFlowTests { assertThat(e.getMessage(), containsString("The 'receive()/receiveAndConvert()' isn't supported")); } + assertThat(this.beanFactory.getBeanNamesForType(MessageTransformingHandler.class)[0], startsWith(flowId + ".")); + flowRegistration.destroy(); assertFalse(this.beanFactory.containsBean(flowRegistration.getId())); @@ -231,7 +250,8 @@ public class ManualFlowTests { } catch (Exception e) { assertThat(e, instanceOf(IllegalStateException.class)); - assertThat(e.getMessage(), containsString("But [" + "foo" + "] ins't one of them.")); + assertThat(e.getMessage(), + containsString("An IntegrationFlow with the id [" + "foo" + "] doesn't exist in the registry.")); } } @@ -252,16 +272,21 @@ public class ManualFlowTests { Message receive = resultChannel.receive(1000); assertNotNull(receive); assertEquals("test", receive.getPayload()); + + this.integrationFlowContext.remove("dynamicFlow"); } @Test public void testDynamicAdapterFlow() { - this.integrationFlowContext.registration(new MyFlowAdapter()).register(); + IntegrationFlowRegistration flowRegistration = + this.integrationFlowContext.registration(new MyFlowAdapter()).register(); PollableChannel resultChannel = this.beanFactory.getBean("flowAdapterOutput", PollableChannel.class); Message receive = resultChannel.receive(1000); assertNotNull(receive); assertEquals("flowAdapterMessage", receive.getPayload()); + + flowRegistration.destroy(); } @@ -304,8 +329,9 @@ public class ManualFlowTests { PollableChannel resultChannel = new QueueChannel(); IntegrationFlowRegistration flowRegistration = - this.integrationFlowContext.registration(flow -> - flow.handle(new MessageProducingHandler()) + this.integrationFlowContext.registration( + flow -> flow + .handle(new MessageProducingHandler()) .channel(resultChannel)) .register(); @@ -315,6 +341,8 @@ public class ManualFlowTests { Message receive = resultChannel.receive(1000); assertNotNull(receive); assertEquals("test", receive.getPayload()); + + flowRegistration.destroy(); } @Test @@ -362,8 +390,8 @@ public class ManualFlowTests { assertTrue(this.roleController.getEndpointsRunningStatus(testRole).isEmpty()); } - @Test - public void testDynaSubFlowCreation() { + // @Test + public void testDynamicSubFlowCreation() { Flux> messageFlux = Flux.just("1,2,3,4") .map(v -> v.split(",")) @@ -383,7 +411,8 @@ public class ManualFlowTests { .channel(resultChannel) .get(); - this.integrationFlowContext.registration(integrationFlow).register(); + IntegrationFlowRegistration flowRegistration = + this.integrationFlowContext.registration(integrationFlow).register(); for (int i = 0; i < 4; i++) { Message receive = resultChannel.receive(10_000); @@ -391,6 +420,8 @@ public class ManualFlowTests { } assertNull(resultChannel.receive(0)); + + flowRegistration.destroy(); } @Test @@ -401,10 +432,11 @@ public class ManualFlowTests { IntegrationFlows.from(Supplier.class) .get(); - this.integrationFlowContext - .registration(testFlow) - .id(testId) - .register(); + IntegrationFlowRegistration flowRegistration = + this.integrationFlowContext + .registration(testFlow) + .id(testId) + .register(); try { this.integrationFlowContext @@ -416,12 +448,57 @@ public class ManualFlowTests { assertThat(e, instanceOf(IllegalArgumentException.class)); assertThat(e.getMessage(), containsString("with flowId '" + testId + "' is already registered.")); } + + flowRegistration.destroy(); + } + + @Test + public void testConcurrentRegistration() throws InterruptedException { + ExecutorService executorService = Executors.newCachedThreadPool(); + + List flowRegistrations = new ArrayList<>(); + + AtomicBoolean exceptionHappened = new AtomicBoolean(); + + for (int i = 0; i < 100; i++) { + int index = i; + executorService.execute(() -> { + + IntegrationFlow flow = f -> f + .transform(m -> m); + + try { + IntegrationFlowContext.IntegrationFlowRegistrationBuilder registration = + this.integrationFlowContext.registration(flow); + if (index % 2 == 0) { + registration.id("concurrentFlow#" + index); + } + flowRegistrations.add(registration.register()); + } + catch (Exception e) { + exceptionHappened.set(true); + } + + }); + } + + executorService.shutdownNow(); + assertTrue(executorService.awaitTermination(10, TimeUnit.SECONDS)); + + assertFalse(exceptionHappened.get()); + + flowRegistrations.forEach(IntegrationFlowRegistration::destroy); } @Configuration @EnableIntegration public static class RootConfiguration { + @Bean + public static BeanFactoryPostProcessor beanFactoryPostProcessor() { + return beanFactory -> ((DefaultListableBeanFactory) beanFactory).setAllowBeanDefinitionOverriding(false); + } + @Bean @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE) public Date foo() { diff --git a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java index 03076cdb53..741407fe7a 100644 --- a/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java +++ b/spring-integration-webflux/src/test/java/org/springframework/integration/webflux/dsl/WebFluxDslTests.java @@ -101,7 +101,7 @@ public class WebFluxDslTests { @Qualifier("webFluxWithReplyPayloadToFlux.handler") private WebFluxRequestExecutingMessageHandler webFluxWithReplyPayloadToFlux; - @Resource(name = "org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler#1") + @Resource(name = "httpReactiveProxyFlow.org.springframework.integration.webflux.outbound.WebFluxRequestExecutingMessageHandler#0") private WebFluxRequestExecutingMessageHandler httpReactiveProxyFlow; @Autowired diff --git a/src/reference/asciidoc/dsl.adoc b/src/reference/asciidoc/dsl.adoc index 0090c0af11..8bed5fedb1 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -578,6 +578,12 @@ 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. +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. +See <> for more information. + [[java-dsl-function-expression]] === FunctionExpression @@ -889,6 +895,10 @@ 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. +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. + [[java-dsl-gateway]] === IntegrationFlow as Gateway diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 43ec92b21f..e1f6f9c965 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -315,3 +315,10 @@ Further changes were made in 5.0.4; if using Micrometer, a minimum of version 5. Introduced in _version 5.0.4_, this annotation provides control over bean naming when using Java configuration. See <> for more information. + +==== Integration Flows: Generated bean names + +Starting with _version 5.0.5_, generated bean names for the components in an `IntegrationFlow` include the flow bean name, followed by a dot, as a prefix. + +See <> for more information. +