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 c57cf6d284..fed2e6253f 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 @@ -123,7 +123,7 @@ public class IntegrationFlowBeanPostProcessor String id = endpointSpec.getId(); if (id == null) { - id = generateBeanName(endpoint, entry.getValue()); + id = generateBeanName(endpoint, flowNamePrefix, entry.getValue()); } Collection messageHandlers = @@ -131,13 +131,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); @@ -187,12 +184,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); @@ -238,7 +236,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); } @@ -335,11 +333,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(); } @@ -347,7 +345,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/StandardIntegrationFlowContext.java b/spring-integration-core/src/main/java/org/springframework/integration/dsl/context/StandardIntegrationFlowContext.java index 5f9162a68a..ebf38736ca 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 @@ -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; @@ -46,7 +49,9 @@ import org.springframework.util.Assert; */ public final class StandardIntegrationFlowContext implements IntegrationFlowContext, BeanFactoryAware { - private final Map registry = new HashMap<>(); + private final Map registry = new ConcurrentHashMap<>(); + + private final Lock registerFlowsLock = new ReentrantLock(); private ConfigurableListableBeanFactory beanFactory; @@ -76,17 +81,29 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont private void register(StandardIntegrationFlowRegistrationBuilder 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)); @@ -133,19 +150,26 @@ public final class StandardIntegrationFlowContext implements IntegrationFlowCont * @param flowId the bean name to destroy from */ @Override - 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 e744e486ac..d89a3100d4 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 @@ -19,6 +19,7 @@ package org.springframework.integration.dsl.manualflow; import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.instanceOf; import static org.hamcrest.Matchers.lessThan; +import static org.hamcrest.Matchers.startsWith; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -28,9 +29,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; @@ -41,8 +47,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; @@ -63,6 +72,7 @@ import org.springframework.integration.dsl.context.IntegrationFlowContext.Integr 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; @@ -92,7 +102,7 @@ public class ManualFlowTests { private IntegrationFlowContext integrationFlowContext; @Autowired - private BeanFactory beanFactory; + private ListableBeanFactory beanFactory; @Autowired private SmartLifecycleRoleController roleController; @@ -113,8 +123,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 @@ -138,15 +150,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()) @@ -160,6 +176,7 @@ public class ManualFlowTests { BeanFactoryHandler additionalBean = new BeanFactoryHandler(); IntegrationFlowRegistration flowRegistration = this.integrationFlowContext.registration(myFlow) + .id(flowId) .addBean(additionalBean) .register(); @@ -185,6 +202,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())); @@ -210,7 +229,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.")); } } @@ -231,16 +251,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(); } @@ -283,8 +308,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(); @@ -294,6 +320,8 @@ public class ManualFlowTests { Message receive = resultChannel.receive(1000); assertNotNull(receive); assertEquals("test", receive.getPayload()); + + flowRegistration.destroy(); } @Test @@ -341,8 +369,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(",")) @@ -362,7 +390,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); @@ -370,6 +399,8 @@ public class ManualFlowTests { } assertNull(resultChannel.receive(0)); + + flowRegistration.destroy(); } @Test @@ -380,10 +411,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 @@ -395,12 +427,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 82d70a24af..117bbcf120 100644 --- a/src/reference/asciidoc/dsl.adoc +++ b/src/reference/asciidoc/dsl.adoc @@ -580,6 +580,12 @@ 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. +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 @@ -927,6 +933,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 d1b0655c33..41543aa10f 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -44,3 +44,9 @@ Previously, interceptors were not applied when beans were created after the appl A new `ResultType.BYTES` mode is introduced for the `ObjectToJsonTransformer`. 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.