Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Defer Messaging annotations process (#2769)

Browse Source 
* Defer Messaging annotations process

The `AbstractMethodAnnotationPostProcessor` and its implementations
have a `beanFactory.getBean()` call for the `@Bean` methods with
Messaging annotations.
This is done, actually, from the
`MessagingAnnotationPostProcessor.postProcessAfterInitialization()`
which might be still too early in some scenarios, like Spring Cloud Feign
with its child application contexts being initialized from the
`FeignClientFactoryBean`, causing a `BeanCurrentlyInCreationException`

See https://stackoverflow.com/questions/54887963/beancurrentlyincreationexception-when-using-spring-integration-with-spring-cloud

* Implement a `SmartInitializingSingleton` for the `MessagingAnnotationPostProcessor`
and gather `Runnable` wrappers for newly introduced `postProcessMethodAndRegisterEndpointIfAny()`
to be called later in the `afterSingletonsInstantiated()` when context is
still in the initialization phase.
All runtime-registered beans are going to be processed normally from the
regular `postProcessAfterInitialization()`

**Cherry-pick to 5.1.x**

* * Fix unused imports in the `MessagingAnnotationsWithBeanAnnotationTests`

* * Fix `IntegrationEndpointsInitializer` in the testing framework to handle
all the possible `AbstractEndpoint` beans registration.
See its JavaDocs for more info
* Fix `AbstractCorrelatingMessageHandlerParser` and
`AbstractConsumerEndpointParser` to use bean names for `outputChannel`
and `discardChannel` instead of bean references.
Since `MessagingAnnotationPostProcessor` now registers endpoints and
beans for channels much later, than parsers, we can't rely on bean
references any more there.
* Fixes for failing tests which expected `outputChannel/discardChannel`
bean references, when it is already just their names for late binding.
* Apply some code style polishing for the affected classes.
* Add `@Nullable` for `MessageSelector` parameter in the `QueueChannel.purge()`
This commit is contained in:
Artem Bilan
2019-03-01 10:26:05 -05:00
committed by Gary Russell
parent 4365eaef6d
commit 3657d05596
27 changed files with 345 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/inbound/SftpStreamingMessageSourceTests.java
View File

@@ -21,6 +21,8 @@ import static org.assertj.core.api.Assertions.assertThat;
import java.io.InputStream;
import java.util.Arrays;
import java.util.Comparator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -38,8 +40,10 @@ import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.filters.AcceptAllFileListFilter;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.metadata.SimpleMetadataStore;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.integration.sftp.SftpTestSupport;
import org.springframework.integration.sftp.filters.SftpPersistentAcceptOnceFileListFilter;
import org.springframework.integration.sftp.session.SftpFileInfo;
import org.springframework.integration.sftp.session.SftpRemoteFileTemplate;
import org.springframework.integration.transformer.StreamTransformer;
@@ -76,6 +80,9 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
@Autowired
private ApplicationContext context;
@Autowired
private ConcurrentMap<String, String> metadataMap;
@SuppressWarnings("unchecked")
@Test
public void testAllContents() {
@@ -106,7 +113,9 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
this.adapter.stop();
this.source.setFileInfoJson(false);
this.data.purge(null);
this.metadataMap.clear();
this.adapter.start();
assertThat(this.data.receive(10000)).isNotNull();
received = (Message<byte[]>) this.data.receive(10000);
assertThat(received).isNotNull();
assertThat(received.getHeaders().get(FileHeaders.REMOTE_FILE_INFO)).isInstanceOf(SftpFileInfo.class);
@@ -179,12 +188,20 @@ public class SftpStreamingMessageSourceTests extends SftpTestSupport {
return pollerMetadata;
}
@Bean
public ConcurrentMap<String, String> metadataMap() {
return new ConcurrentHashMap<>();
}
@Bean
@InboundChannelAdapter(channel = "stream", autoStartup = "false")
public MessageSource<InputStream> sftpMessageSource() {
SftpStreamingMessageSource messageSource = new SftpStreamingMessageSource(template(),
Comparator.comparing(LsEntry::getFilename));
messageSource.setFilter(new AcceptAllFileListFilter<>());
messageSource.setFilter(
new SftpPersistentAcceptOnceFileListFilter(
new SimpleMetadataStore(metadataMap()), "testStreaming"));
messageSource.setRemoteDirectory("sftpSource/");
return messageSource;
}