From 1d93702b236155ea3c93b1e7b6eaffc4173f7a77 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Mon, 7 Mar 2016 13:01:34 -0500 Subject: [PATCH] INT-3961: Add Channel Late Binding for `@ICA` JIRA: https://jira.spring.io/browse/INT-3961 The `@InboundChannelAdapter` may be initialized by its `MessagingAnnotationPostProcessor` a bit earlier than the provided channel is created and registered in the context - via some other more late `BPP`, e.g. `IntegrationFlowBeanPostProcessor` from Java DSL. * Add channel late-binding logic to the `InboundChannelAdapterAnnotationPostProcessor` and `SourcePollingChannelAdapter` by the provided `outputChannelName` option. * Add the `channel` alias for the `value` annotation attribute in the `@InboundChannelAdapter` * Add the `@InboundChannelAdapter` configuration into the `EnableIntegrationTests` without the channel declaration to be sure that the fix provide the expected results without `DestinationResolutionException` * Mention both changes in the `configuration.adoc` and `whats-new.adoc` --- .../annotation/InboundChannelAdapter.java | 15 ++++++- ...ChannelAdapterAnnotationPostProcessor.java | 7 +--- .../endpoint/SourcePollingChannelAdapter.java | 27 +++++++++++-- .../configuration/EnableIntegrationTests.java | 40 ++++++++++++++++--- src/reference/asciidoc/configuration.adoc | 8 ++++ src/reference/asciidoc/whats-new.adoc | 7 ++++ 6 files changed, 88 insertions(+), 16 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/annotation/InboundChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/annotation/InboundChannelAdapter.java index f63aae326d..bc4bfe1802 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/annotation/InboundChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/annotation/InboundChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors. + * Copyright 2014-2016 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. @@ -23,6 +23,8 @@ import java.lang.annotation.Retention; import java.lang.annotation.RetentionPolicy; import java.lang.annotation.Target; +import org.springframework.core.annotation.AliasFor; + /** * Indicates that a method is capable of producing a {@link org.springframework.messaging.Message} * or {@link org.springframework.messaging.Message} {@code payload}. @@ -53,9 +55,18 @@ import java.lang.annotation.Target; public @interface InboundChannelAdapter { /** + * Alias for the {@link #channel()} attribute. * @return the 'channel' bean name to send the {@link org.springframework.messaging.Message}. */ - String value(); + @AliasFor("channel") + String value() default ""; + + /** + * @return the 'channel' bean name to send the {@link org.springframework.messaging.Message}. + * @since 4.2.6 + */ + @AliasFor("value") + String channel() default ""; /* {@code SmartLifecycle} options. diff --git a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java index 39b12f41f1..8c6255b24f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/config/annotation/InboundChannelAdapterAnnotationPostProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors. + * Copyright 2014-2016 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. @@ -31,7 +31,6 @@ import org.springframework.integration.core.MessageSource; import org.springframework.integration.endpoint.MethodInvokingMessageSource; import org.springframework.integration.endpoint.SourcePollingChannelAdapter; import org.springframework.integration.util.MessagingAnnotationUtils; -import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; import org.springframework.util.Assert; @@ -72,10 +71,8 @@ public class InboundChannelAdapterAnnotationPostProcessor extends return null; } - MessageChannel channel = this.channelResolver.resolveDestination(channelName); - SourcePollingChannelAdapter adapter = new SourcePollingChannelAdapter(); - adapter.setOutputChannel(channel); + adapter.setOutputChannelName(channelName); adapter.setSource(messageSource); configurePollingEndpoint(adapter, annotations); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java index 8cc4e6d06b..03d2db67fb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/endpoint/SourcePollingChannelAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2015 the original author or authors. + * Copyright 2002-2016 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. @@ -53,6 +53,8 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint private volatile MessageChannel outputChannel; + private volatile String outputChannelName; + private volatile boolean shouldTrack; private final MessagingTemplate messagingTemplate = new MessagingTemplate(); @@ -75,6 +77,11 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint this.outputChannel = outputChannel; } + public void setOutputChannelName(String outputChannelName) { + Assert.hasText(outputChannelName, "'outputChannelName' must not be empty"); + this.outputChannelName = outputChannelName; + } + /** * Specify the maximum time to wait for a Message to be sent to the * output channel. @@ -145,20 +152,34 @@ public class SourcePollingChannelAdapter extends AbstractPollingEndpoint @Override protected void onInit() { Assert.notNull(this.source, "source must not be null"); - Assert.notNull(this.outputChannel, "outputChannel must not be null"); + Assert.state((this.outputChannelName == null && this.outputChannel != null) + || (this.outputChannelName != null && this.outputChannel == null), + "One and only one of 'outputChannelName' or 'outputChannel' is required."); super.onInit(); if (this.getBeanFactory() != null) { this.messagingTemplate.setBeanFactory(this.getBeanFactory()); } } + public MessageChannel getOutputChannel() { + if (this.outputChannelName != null) { + synchronized (this) { + if (this.outputChannelName != null) { + this.outputChannel = getChannelResolver().resolveDestination(this.outputChannelName); + this.outputChannelName = null; + } + } + } + return this.outputChannel; + } + @Override protected void handleMessage(Message message) { if (this.shouldTrack) { message = MessageHistory.write(message, this, this.getMessageBuilderFactory()); } try { - this.messagingTemplate.send(this.outputChannel, message); + this.messagingTemplate.send(getOutputChannel(), message); } catch (Exception e) { if (e instanceof MessagingException) { diff --git a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java index 6a1681a26b..5d84e774e8 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/configuration/EnableIntegrationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014-2015 the original author or authors. + * Copyright 2014-2016 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. @@ -45,7 +45,6 @@ import java.util.concurrent.atomic.AtomicReference; import org.aopalliance.intercept.MethodInterceptor; import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; @@ -57,6 +56,7 @@ import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.config.AbstractFactoryBean; +import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.context.ApplicationContext; import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; @@ -92,6 +92,7 @@ import org.springframework.integration.config.EnablePublisher; import org.springframework.integration.config.ExpressionControlBusFactoryBean; import org.springframework.integration.config.GlobalChannelInterceptor; import org.springframework.integration.config.IntegrationConverter; +import org.springframework.integration.core.MessageSource; import org.springframework.integration.core.MessagingTemplate; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.MethodInvokingMessageSource; @@ -264,12 +265,16 @@ public class EnableIntegrationTests { @Autowired private MessageChannel controlBusChannel; + @Autowired + private CountDownLatch inputReceiveLatch; + @Autowired @Qualifier("enableIntegrationTests.ContextConfiguration2.sendAsyncHandler.serviceActivator") private AbstractEndpoint sendAsyncHandler; @Autowired - private CountDownLatch inputReceiveLatch; + @Qualifier("enableIntegrationTests.ChildConfiguration.autoCreatedChannelMessageSource.inboundChannelAdapter") + private Lifecycle autoCreatedChannelMessageSourceAdapter; @Test public void testAnnotatedServiceActivator() throws Exception { @@ -645,6 +650,21 @@ public class EnableIntegrationTests { assertEquals(2, lifecycles.get("foo").size()); } + @Test + public void testSourcePollingChannelAdapterOutputChannelLateBinding() { + QueueChannel testChannel = new QueueChannel(); + ConfigurableListableBeanFactory beanFactory = + (ConfigurableListableBeanFactory) this.context.getAutowireCapableBeanFactory(); + beanFactory.registerSingleton("lateBindingChannel", testChannel); + beanFactory.initializeBean(testChannel, "lateBindingChannel"); + + this.autoCreatedChannelMessageSourceAdapter.start(); + Message receive = testChannel.receive(10000); + assertNotNull(receive); + assertEquals("bar", receive.getPayload()); + this.autoCreatedChannelMessageSourceAdapter.stop(); + } + @Configuration @ComponentScan @IntegrationComponentScan @@ -854,8 +874,6 @@ public class EnableIntegrationTests { @GlobalChannelInterceptor public static class TestChannelInterceptor extends ChannelInterceptorAdapter { - private final Log logger = LogFactory.getLog(TestChannelInterceptor.class); - private final AtomicInteger invoked = new AtomicInteger(); @Override @@ -1050,6 +1068,14 @@ public class EnableIntegrationTests { return new WireTap(new NullChannel()); } + //Before INT-3961 it fails with the DestinationResolutionException + @InboundChannelAdapter(channel = "lateBindingChannel", autoStartup = "false", + poller = @Poller(fixedDelay = "100")) + @Bean + public MessageSource autoCreatedChannelMessageSource() { + return () -> new GenericMessage<>("bar"); + } + } public interface AnnotationTestService { @@ -1307,7 +1333,7 @@ public class EnableIntegrationTests { @Target({ElementType.TYPE, ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) @MessagingGateway(defaultRequestChannel = "gatewayChannel", reactorEnvironment = "reactorEnv", - defaultRequestTimeout="${default.request.timeout:12300}", defaultReplyTimeout="#{13400}", + defaultRequestTimeout = "${default.request.timeout:12300}", defaultReplyTimeout = "#{13400}", defaultHeaders = @GatewayHeader(name = "foo", value = "FOO")) public @interface TestMessagingGateway { @@ -1440,6 +1466,7 @@ public class EnableIntegrationTests { adviceChain = {"annAdvice"}, poller = @Poller(fixedDelay = "1000")) public @interface MyServiceActivatorNoLocalAtts { + } @Target(ElementType.METHOD) @@ -1521,6 +1548,7 @@ public class EnableIntegrationTests { @Retention(RetentionPolicy.RUNTIME) @BridgeTo(autoStartup = "false") public @interface MyBridgeTo { + } // Error because the annotation is on a class; it must be on an interface diff --git a/src/reference/asciidoc/configuration.adoc b/src/reference/asciidoc/configuration.adoc index cc8d9c8cd2..105bc97fc8 100644 --- a/src/reference/asciidoc/configuration.adoc +++ b/src/reference/asciidoc/configuration.adoc @@ -438,6 +438,14 @@ public String foo() { } ---- +Starting with _version 4.3_ the `channel` alias for the `value` annotation attribute has been introduced for better +source code readability. +Also the target `MessageChannel` bean is resolved in the `SourcePollingChannelAdapter` by the provided name +(`outputChannelName` options) on the first `receive()` call, not during +initialization phase. +It allows the 'late binding' logic, when the target `MessageChannel` bean from the consumer perspective +is created and registered a bit later than the `@InboundChannelAdapter` parsing phase. + The first example requires that the default poller has been declared elsewhere in the application context. *@MessagingGateway* diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index 5dce113faf..71289557cc 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -141,3 +141,10 @@ See <>, <>, and <> for more information. + +==== @InboundChannelAdapter + +The `@InboundChannelAdapter` has now an alias `channel` attribute for regular `value`. +In addition the target `SourcePollingChannelAdapter` components can now resolve the target `outputChannel` bean +from its provided name (`outputChannelName` options) in late-binding manner. +See <> for more information.