From 910f1163320654f405e6137fc33943248664d1bc Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Fri, 15 Jul 2022 20:36:21 -0400 Subject: [PATCH] GH-276 Apply ComponentCustomizer in other modules (#284) * GH-276 Apply ComponentCustomizer in other modules Fixes https://github.com/spring-cloud/stream-applications/issues/276 Some changes are straightforward as just a `ComponentCustomizer` injection and its optional usage for the target component customization. Other changes are more drastic since it is better to use `proxyBeanMethods = false`, so some direct bean method calls had to be reworked to the bean method argument injections * Fix some deprecations from Project Reactor * Remove `UpdatingMongoDbMessageSource` since `MongoDbMessageSource` supports now an `update` option in Spring Integration NOTE: The `ZeroMqSupplierConfiguration` has its own customizer already like `Consumer`, so it can be revised to a more broader solution, but in the future version. Some modules just does not make sense to modify for more advance customization since all their configuration is covered with plain configuration properties. See `splitter-function` or `wavefront-consumer` * * Use `ObjectProvider` for `HttpRequestFunctionConfiguration` instead of our own `ComponentCustomizer`. Although this has to be revised in favor of fully auto-configured `WebClient.Builder`. Even that `maximumBufferSize` is included into codecs configuration properties. * The `MongoDbMessageSource` has now an `update` option in Spring Integration therefore remove redundant already `UpdatingMongoDbMessageSource` and populate `this.properties.getUpdateExpression()` directly ot the `MongoDbMessageSource` bean * Modify supplier configurations to avoid manual starts for endpoints on reactive `Publisher` subscriptions. Spring Integration provides now a `toReactivePublisher(true)` operator to have all the endpoints stopped on start, but started automatically when subscription happens to the provided `Publisher` * Update Copyright to the current year for all the affected classes * Use `@Nullable` instead of `ObjectProvider` in the `AggregatorFunctionConfiguration` dependency injection for consistency * Fix `LogConsumerConfiguration` to not have a `log()` in the end - it is not terminal in Spring Integration anymore. Use recommended `nullChannel()` terminating operator for use-cases like this. * * Remove `beanName` from `ComponentCustomizer` contract Co-authored-by: Chris Bono --- .../fn/common/config/ComponentCustomizer.java | 2 +- .../file/FileConsumerConfiguration.java | 29 +++-- .../ftp/FtpConsumerConfiguration.java | 17 ++- .../geode/GeodeConsumerConfiguration.java | 17 ++- .../log/LogConsumerConfiguration.java | 7 +- .../mongo/MongoDbConsumerConfiguration.java | 18 ++- .../mqtt/MqttConsumerConfiguration.java | 18 ++- .../rabbit/RabbitConsumerConfiguration.java | 20 ++- .../rsocket/RsocketConsumerConfiguration.java | 17 +-- .../sftp/SftpConsumerConfiguration.java | 14 ++- .../AggregatorFunctionConfiguration.java | 34 +++-- .../HttpRequestFunctionConfiguration.java | 7 +- .../file/FileSupplierConfiguration.java | 2 +- .../file/AbstractFileSupplierTests.java | 4 +- .../ftp/FtpSupplierConfiguration.java | 41 ++++-- .../jdbc/JdbcSupplierConfiguration.java | 24 ++-- .../jms/JmsSupplierConfiguration.java | 33 +++-- .../mail/MailSupplierConfiguration.java | 67 +++++----- .../mongo/MongodbSupplierConfiguration.java | 39 +++--- .../mongo/UpdatingMongoDbMessageSource.java | 118 ------------------ .../mqtt/MqttSupplierConfiguration.java | 39 +++--- .../rabbit/RabbitSupplierConfiguration.java | 55 +++++--- .../s3/AwsS3SupplierConfiguration.java | 48 +++---- 23 files changed, 348 insertions(+), 322 deletions(-) delete mode 100644 supplier/mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/UpdatingMongoDbMessageSource.java diff --git a/common/config-common/src/main/java/org/springframework/cloud/fn/common/config/ComponentCustomizer.java b/common/config-common/src/main/java/org/springframework/cloud/fn/common/config/ComponentCustomizer.java index 1b17c343..3af872ae 100644 --- a/common/config-common/src/main/java/org/springframework/cloud/fn/common/config/ComponentCustomizer.java +++ b/common/config-common/src/main/java/org/springframework/cloud/fn/common/config/ComponentCustomizer.java @@ -29,6 +29,6 @@ package org.springframework.cloud.fn.common.config; @FunctionalInterface public interface ComponentCustomizer { - void customize(T component, String beanName); + void customize(T component); } diff --git a/consumer/file-consumer/src/main/java/org/springframework/cloud/fn/consumer/file/FileConsumerConfiguration.java b/consumer/file-consumer/src/main/java/org/springframework/cloud/fn/consumer/file/FileConsumerConfiguration.java index d00f57f5..784418f3 100644 --- a/consumer/file-consumer/src/main/java/org/springframework/cloud/fn/consumer/file/FileConsumerConfiguration.java +++ b/consumer/file-consumer/src/main/java/org/springframework/cloud/fn/consumer/file/FileConsumerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2022 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. @@ -19,6 +19,7 @@ package org.springframework.cloud.fn.consumer.file; import java.util.function.Consumer; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.expression.ExpressionParser; @@ -26,6 +27,7 @@ import org.springframework.expression.spel.standard.SpelExpressionParser; import org.springframework.integration.file.DefaultFileNameGenerator; import org.springframework.integration.file.FileNameGenerator; import org.springframework.integration.file.FileWritingMessageHandler; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -33,7 +35,7 @@ import org.springframework.messaging.Message; * @author Artem Bilan * @author Soby Chacko */ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(FileConsumerProperties.class) public class FileConsumerConfiguration { @@ -46,21 +48,29 @@ public class FileConsumerConfiguration { } @Bean - public Consumer> fileConsumer() { - return fileWritingMessageHandler()::handleMessage; + public Consumer> fileConsumer(FileWritingMessageHandler fileWritingMessageHandler) { + return fileWritingMessageHandler::handleMessage; } @Bean - public FileWritingMessageHandler fileWritingMessageHandler() { - FileWritingMessageHandler handler = (properties.getDirectoryExpression() != null) - ? new FileWritingMessageHandler(EXPRESSION_PARSER.parseExpression(properties.getDirectoryExpression())) - : new FileWritingMessageHandler(properties.getDirectory()); + public FileWritingMessageHandler fileWritingMessageHandler(FileNameGenerator fileNameGenerator, + @Nullable ComponentCustomizer fileWritingMessageHandlerCustomizer) { + + FileWritingMessageHandler handler = + this.properties.getDirectoryExpression() != null + ? new FileWritingMessageHandler( + EXPRESSION_PARSER.parseExpression(this.properties.getDirectoryExpression())) + : new FileWritingMessageHandler(this.properties.getDirectory()); handler.setAutoCreateDirectory(true); handler.setAppendNewLine(!properties.isBinary()); handler.setCharset(properties.getCharset()); handler.setExpectReply(false); handler.setFileExistsMode(properties.getMode()); - handler.setFileNameGenerator(fileNameGenerator()); + handler.setFileNameGenerator(fileNameGenerator); + + if (fileWritingMessageHandlerCustomizer != null) { + fileWritingMessageHandlerCustomizer.customize(handler); + } return handler; } @@ -70,4 +80,5 @@ public class FileConsumerConfiguration { fileNameGenerator.setExpression(properties.getNameExpression()); return fileNameGenerator; } + } diff --git a/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerConfiguration.java b/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerConfiguration.java index 3a3bb67c..ece2d0a3 100644 --- a/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerConfiguration.java +++ b/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2022 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. @@ -22,6 +22,7 @@ import org.apache.commons.net.ftp.FTPFile; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.cloud.fn.common.ftp.FtpSessionFactoryConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; @@ -35,9 +36,10 @@ import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.ftp.dsl.Ftp; import org.springframework.integration.ftp.dsl.FtpMessageHandlerSpec; import org.springframework.integration.ftp.session.FtpRemoteFileTemplate; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(FtpConsumerProperties.class) @Import(FtpSessionFactoryConfiguration.class) public class FtpConsumerConfiguration { @@ -48,7 +50,8 @@ public class FtpConsumerConfiguration { FtpConsumerProperties ftpConsumerProperties; @Bean - public IntegrationFlow ftpInboundFlow(FtpConsumerProperties properties, SessionFactory ftpSessionFactory) { + public IntegrationFlow ftpInboundFlow(FtpConsumerProperties properties, SessionFactory ftpSessionFactory, + @Nullable ComponentCustomizer ftpMessageHandlerSpecCustomizer) { IntegrationFlowBuilder integrationFlowBuilder = IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("ftpConsumer")); @@ -60,8 +63,14 @@ public class FtpConsumerConfiguration { .autoCreateDirectory(properties.isAutoCreateDir()) .temporaryFileSuffix(properties.getTmpFileSuffix()); if (properties.getFilenameExpression() != null) { - handlerSpec.fileNameExpression(EXPRESSION_PARSER.parseExpression(properties.getFilenameExpression()).getExpressionString()); + handlerSpec.fileNameExpression( + EXPRESSION_PARSER.parseExpression(properties.getFilenameExpression()).getExpressionString()); } + + if (ftpMessageHandlerSpecCustomizer != null) { + ftpMessageHandlerSpecCustomizer.customize(handlerSpec); + } + return integrationFlowBuilder .handle(handlerSpec) .get(); diff --git a/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerConfiguration.java b/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerConfiguration.java index b4a0e1e4..474c0c7a 100644 --- a/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerConfiguration.java +++ b/consumer/geode-consumer/src/main/java/org/springframework/cloud/fn/consumer/geode/GeodeConsumerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-2022 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,11 +23,13 @@ import java.util.function.Function; import org.apache.geode.cache.Region; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.cloud.fn.common.geode.GeodeClientRegionConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.integration.gemfire.outbound.CacheWritingMessageHandler; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -35,7 +37,7 @@ import org.springframework.messaging.Message; * Message, using a SpEL expression for a key, and the payload for the value. * @author David Turanski */ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(GeodeConsumerProperties.class) @Import(GeodeClientRegionConfiguration.class) public class GeodeConsumerConfiguration { @@ -52,10 +54,15 @@ public class GeodeConsumerConfiguration { } @Bean - CacheWritingMessageHandler cacheWriter(Region region, GeodeConsumerProperties properties) { + CacheWritingMessageHandler cacheWriter(Region region, GeodeConsumerProperties properties, + @Nullable ComponentCustomizer cacheWritingMessageHandlerCustomizer) { + CacheWritingMessageHandler messageHandler = new CacheWritingMessageHandler(region); - messageHandler.setCacheEntries( - Collections.singletonMap(properties.getKeyExpression(), "payload")); + messageHandler.setCacheEntries(Collections.singletonMap(properties.getKeyExpression(), "payload")); + if (cacheWritingMessageHandlerCustomizer != null) { + cacheWritingMessageHandlerCustomizer.customize(messageHandler); + } return messageHandler; } + } diff --git a/consumer/log-consumer/src/main/java/org/springframework/cloud/fn/consumer/log/LogConsumerConfiguration.java b/consumer/log-consumer/src/main/java/org/springframework/cloud/fn/consumer/log/LogConsumerConfiguration.java index 96be80ab..3ae4913b 100644 --- a/consumer/log-consumer/src/main/java/org/springframework/cloud/fn/consumer/log/LogConsumerConfiguration.java +++ b/consumer/log-consumer/src/main/java/org/springframework/cloud/fn/consumer/log/LogConsumerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-2022 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. @@ -35,16 +35,15 @@ import org.springframework.messaging.Message; * * @author Artem Bilan */ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(LogConsumerProperties.class) public class LogConsumerConfiguration { @Bean IntegrationFlow logConsumerFlow(LogConsumerProperties logSinkProperties) { return IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("logConsumer")) - .handle((payload, headers) -> payload) .log(logSinkProperties.getLevel(), logSinkProperties.getName(), logSinkProperties.getExpression()) - .get(); + .nullChannel(); } private interface MessageConsumer extends Consumer> { diff --git a/consumer/mongodb-consumer/src/main/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerConfiguration.java b/consumer/mongodb-consumer/src/main/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerConfiguration.java index ff270ace..e50d3528 100644 --- a/consumer/mongodb-consumer/src/main/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerConfiguration.java +++ b/consumer/mongodb-consumer/src/main/java/org/springframework/cloud/fn/consumer/mongo/MongoDbConsumerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2020 the original author or authors. + * Copyright 2017-2022 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. @@ -22,12 +22,14 @@ import java.util.function.Function; import reactor.core.publisher.Mono; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.mongodb.core.ReactiveMongoTemplate; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.mongodb.outbound.ReactiveMongoDbStoringMessageHandler; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.ReactiveMessageHandler; @@ -39,7 +41,7 @@ import org.springframework.messaging.ReactiveMessageHandler; * @author David Turanski * */ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties({ MongoDbConsumerProperties.class }) public class MongoDbConsumerConfiguration { @@ -58,12 +60,16 @@ public class MongoDbConsumerConfiguration { } @Bean - public Function, Mono> mongodbConsumerFunction(ReactiveMessageHandler mongoConsumerMessageHandler) { + public Function, Mono> mongodbConsumerFunction( + ReactiveMessageHandler mongoConsumerMessageHandler) { + return mongoConsumerMessageHandler::handleMessage; } @Bean - public ReactiveMessageHandler mongoConsumerMessageHandler() { + public ReactiveMessageHandler mongoConsumerMessageHandler( + @Nullable ComponentCustomizer mongoDbMessageHandlerCustomizer) { + ReactiveMongoDbStoringMessageHandler mongoDbMessageHandler = new ReactiveMongoDbStoringMessageHandler( this.mongoTemplate); Expression collectionExpression = this.properties.getCollectionExpression(); @@ -71,6 +77,10 @@ public class MongoDbConsumerConfiguration { collectionExpression = new LiteralExpression(this.properties.getCollection()); } mongoDbMessageHandler.setCollectionNameExpression(collectionExpression); + if (mongoDbMessageHandlerCustomizer != null) { + mongoDbMessageHandlerCustomizer.customize(mongoDbMessageHandler); + } return mongoDbMessageHandler; } + } diff --git a/consumer/mqtt-consumer/src/main/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerConfiguration.java b/consumer/mqtt-consumer/src/main/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerConfiguration.java index 85a69726..cd8d8a0d 100644 --- a/consumer/mqtt-consumer/src/main/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerConfiguration.java +++ b/consumer/mqtt-consumer/src/main/java/org/springframework/cloud/fn/consumer/mqtt/MqttConsumerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2020 the original author or authors. + * Copyright 2017-2022 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. @@ -21,6 +21,7 @@ import java.util.function.Consumer; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.cloud.fn.common.mqtt.MqttConfiguration; import org.springframework.cloud.fn.common.mqtt.MqttProperties; import org.springframework.context.annotation.Bean; @@ -29,6 +30,7 @@ import org.springframework.context.annotation.Import; import org.springframework.integration.mqtt.core.MqttPahoClientFactory; import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; @@ -38,7 +40,7 @@ import org.springframework.messaging.MessageHandler; * @author Janne Valkealahti * */ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties({ MqttProperties.class, MqttConsumerProperties.class }) @Import(MqttConfiguration.class) public class MqttConsumerConfiguration { @@ -53,16 +55,21 @@ public class MqttConsumerConfiguration { private BeanFactory beanFactory; @Bean - public Consumer> mqttConsumer() { - return mqttOutbound()::handleMessage; + public Consumer> mqttConsumer(MessageHandler mqttOutbound) { + return mqttOutbound::handleMessage; } @Bean - public MessageHandler mqttOutbound() { + public MessageHandler mqttOutbound( + @Nullable ComponentCustomizer mqttMessageHandlerCustomizer) { + MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(properties.getClientId(), mqttClientFactory); messageHandler.setAsync(properties.isAsync()); messageHandler.setDefaultTopic(properties.getTopic()); messageHandler.setConverter(pahoMessageConverter()); + if (mqttMessageHandlerCustomizer != null) { + mqttMessageHandlerCustomizer.customize(messageHandler); + } return messageHandler; } @@ -72,4 +79,5 @@ public class MqttConsumerConfiguration { converter.setBeanFactory(beanFactory); return converter; } + } diff --git a/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java b/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java index 06b15247..ad92019c 100644 --- a/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java +++ b/consumer/rabbit-consumer/src/main/java/org/springframework/cloud/fn/consumer/rabbit/RabbitConsumerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2021 the original author or authors. + * Copyright 2019-2022 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. @@ -39,12 +39,14 @@ import org.springframework.boot.autoconfigure.amqp.RabbitConnectionFactoryBeanCo import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.io.ResourceLoader; import org.springframework.expression.Expression; import org.springframework.integration.amqp.dsl.Amqp; import org.springframework.integration.amqp.dsl.AmqpOutboundChannelAdapterSpec; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; @@ -92,7 +94,8 @@ public class RabbitConsumerConfiguration implements DisposableBean { } @Bean - public MessageHandler amqpChannelAdapter(ConnectionFactory rabbitConnectionFactory) + public AmqpOutboundChannelAdapterSpec amqpChannelAdapter(ConnectionFactory rabbitConnectionFactory, + @Nullable ComponentCustomizer amqpOutboundChannelAdapterSpecCustomizer) throws Exception { AmqpOutboundChannelAdapterSpec handler = Amqp @@ -119,7 +122,12 @@ public class RabbitConsumerConfiguration implements DisposableBean { else { handler.routingKey(this.properties.getRoutingKey()); } - return handler.get(); + + if (amqpOutboundChannelAdapterSpecCustomizer != null) { + amqpOutboundChannelAdapterSpecCustomizer.customize(handler); + } + + return handler; } @Bean @@ -162,7 +170,8 @@ public class RabbitConsumerConfiguration implements DisposableBean { * [UPGRADE_CONSIDERATION] this should stay somewhat in sync w/ the functionality provided by its original source. */ RabbitConnectionFactoryBean connectionFactoryBean = new RabbitConnectionFactoryBean(); - RabbitConnectionFactoryBeanConfigurer connectionFactoryBeanConfigurer = new RabbitConnectionFactoryBeanConfigurer(resourceLoader, properties); + RabbitConnectionFactoryBeanConfigurer connectionFactoryBeanConfigurer = + new RabbitConnectionFactoryBeanConfigurer(resourceLoader, properties); connectionFactoryBeanConfigurer.setCredentialsProvider(credentialsProvider.getIfUnique()); connectionFactoryBeanConfigurer.setCredentialsRefreshService(credentialsRefreshService.getIfUnique()); connectionFactoryBeanConfigurer.configure(connectionFactoryBean); @@ -173,7 +182,8 @@ public class RabbitConsumerConfiguration implements DisposableBean { .forEach((customizer) -> customizer.customize(connectionFactory)); CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory); - CachingConnectionFactoryConfigurer cachingConnectionFactoryConfigurer = new CachingConnectionFactoryConfigurer(properties); + CachingConnectionFactoryConfigurer cachingConnectionFactoryConfigurer = + new CachingConnectionFactoryConfigurer(properties); cachingConnectionFactoryConfigurer.setConnectionNameStrategy(cf -> "rabbit.sink.own.connection"); cachingConnectionFactoryConfigurer.configure(cachingConnectionFactory); cachingConnectionFactory.afterPropertiesSet(); diff --git a/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerConfiguration.java b/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerConfiguration.java index 727dd5e6..029c21ef 100644 --- a/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerConfiguration.java +++ b/consumer/rsocket-consumer/src/main/java/org/springframework/cloud/fn/consumer/rsocket/RsocketConsumerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-2022 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. @@ -34,17 +34,18 @@ public class RsocketConsumerConfiguration { @Bean public Function>, Mono> rsocketConsumer(RSocketRequester.Builder builder, RsocketConsumerProperties rsocketConsumerProperties) { - final Mono rSocketRequester = - rsocketConsumerProperties.getUri() != null ? builder.connectWebSocket(rsocketConsumerProperties.getUri()).cache() : - builder.connectTcp(rsocketConsumerProperties.getHost(), - rsocketConsumerProperties.getPort()).cache(); + RSocketRequester rSocketRequester = + rsocketConsumerProperties.getUri() != null + ? builder.websocket(rsocketConsumerProperties.getUri()) + : builder.tcp(rsocketConsumerProperties.getHost(), rsocketConsumerProperties.getPort()); + + String route = rsocketConsumerProperties.getRoute(); return input -> input.flatMap(message -> - rSocketRequester - .flatMap(requester -> requester.route(rsocketConsumerProperties.getRoute()) + rSocketRequester.route(route) .data(message.getPayload()) - .send())) + .send()) .ignoreElements(); } diff --git a/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java b/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java index 0b908813..520419d4 100644 --- a/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java +++ b/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2021 the original author or authors. + * Copyright 2015-2022 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. @@ -21,6 +21,7 @@ import java.util.function.Consumer; import com.jcraft.jsch.ChannelSftp; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; @@ -31,16 +32,18 @@ import org.springframework.integration.file.remote.session.SessionFactory; import org.springframework.integration.sftp.dsl.Sftp; import org.springframework.integration.sftp.dsl.SftpMessageHandlerSpec; import org.springframework.integration.sftp.session.SftpRemoteFileTemplate; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(SftpConsumerProperties.class) @Import(SftpConsumerSessionFactoryConfiguration.class) public class SftpConsumerConfiguration { @Bean public IntegrationFlow ftpOutboundFlow(SftpConsumerProperties properties, - SessionFactory ftpSessionFactory) { + SessionFactory ftpSessionFactory, + @Nullable ComponentCustomizer sftpMessageHandlerSpecCustomizer) { IntegrationFlowBuilder integrationFlowBuilder = IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("sftpConsumer")); @@ -56,6 +59,11 @@ public class SftpConsumerConfiguration { if (properties.getFilenameExpression() != null) { handlerSpec.fileNameExpression(properties.getFilenameExpression()); } + + if (sftpMessageHandlerSpecCustomizer != null) { + sftpMessageHandlerSpecCustomizer.customize(handlerSpec); + } + return integrationFlowBuilder .handle(handlerSpec) .get(); diff --git a/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java b/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java index 5711987a..ed2f7ec3 100644 --- a/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java +++ b/function/aggregator-function/src/main/java/org/springframework/cloud/fn/aggregator/AggregatorFunctionConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2020 the original author or authors. + * Copyright 2020-2022 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. @@ -22,12 +22,12 @@ import reactor.core.publisher.Flux; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.BeanFactoryAware; -import org.springframework.beans.factory.ObjectProvider; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; @@ -42,6 +42,7 @@ import org.springframework.integration.annotation.ServiceActivator; import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.config.AggregatorFactoryBean; import org.springframework.integration.store.MessageGroupStore; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -79,21 +80,26 @@ public class AggregatorFunctionConfiguration { @Bean @ServiceActivator(inputChannel = "inputChannel") public AggregatorFactoryBean aggregator( - ObjectProvider correlationStrategy, - ObjectProvider releaseStrategy, - ObjectProvider messageGroupProcessor, - ObjectProvider messageStore, - @Qualifier("outputChannel") MessageChannel outputChannel) { + @Nullable CorrelationStrategy correlationStrategy, + @Nullable ReleaseStrategy releaseStrategy, + @Nullable MessageGroupProcessor messageGroupProcessor, + @Nullable MessageGroupStore messageStore, + @Qualifier("outputChannel") MessageChannel outputChannel, + @Nullable ComponentCustomizer aggregatorCustomizer) { AggregatorFactoryBean aggregator = new AggregatorFactoryBean(); aggregator.setExpireGroupsUponCompletion(true); aggregator.setSendPartialResultOnExpiry(true); aggregator.setGroupTimeoutExpression(this.properties.getGroupTimeout()); - aggregator.setCorrelationStrategy(correlationStrategy.getIfAvailable()); - aggregator.setReleaseStrategy(releaseStrategy.getIfAvailable()); + if (correlationStrategy != null) { + aggregator.setCorrelationStrategy(correlationStrategy); + } + if (releaseStrategy != null) { + aggregator.setReleaseStrategy(releaseStrategy); + } - MessageGroupProcessor groupProcessor = messageGroupProcessor.getIfAvailable(); + MessageGroupProcessor groupProcessor = messageGroupProcessor; if (groupProcessor == null) { groupProcessor = new DefaultAggregatingMessageGroupProcessor(); @@ -101,9 +107,15 @@ public class AggregatorFunctionConfiguration { } aggregator.setProcessorBean(groupProcessor); - aggregator.setMessageStore(messageStore.getIfAvailable()); + if (messageStore != null) { + aggregator.setMessageStore(messageStore); + } aggregator.setOutputChannel(outputChannel); + if (aggregatorCustomizer != null) { + aggregatorCustomizer.customize(aggregator); + } + return aggregator; } diff --git a/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java b/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java index 48fa392a..7f074e44 100644 --- a/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java +++ b/function/http-request-function/src/main/java/org/springframework/cloud/fn/http/request/HttpRequestFunctionConfiguration.java @@ -23,6 +23,7 @@ import java.util.function.Function; import reactor.core.publisher.Flux; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.web.reactive.function.client.WebClientCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.http.HttpHeaders; @@ -33,8 +34,6 @@ import org.springframework.web.reactive.function.client.WebClient; import org.springframework.web.util.DefaultUriBuilderFactory; import org.springframework.web.util.UriBuilderFactory; - - /** * Configuration for a {@link Function} that makes HTTP requests to a resource and for * each request, returns a {@link ResponseEntity}. @@ -43,7 +42,7 @@ import org.springframework.web.util.UriBuilderFactory; * @author Sunny Hemdev * **/ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(HttpRequestFunctionProperties.class) public class HttpRequestFunctionConfiguration { @@ -57,6 +56,7 @@ public class HttpRequestFunctionConfiguration { * returns a {@code Flux>}. */ public static class HttpRequestFunction implements Function>, Flux> { + private final WebClient webClient; private final UriBuilderFactory uriBuilderFactory = new DefaultUriBuilderFactory(); @@ -109,4 +109,5 @@ public class HttpRequestFunctionConfiguration { } } + } diff --git a/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java b/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java index 401fe320..36768728 100644 --- a/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java +++ b/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java @@ -101,7 +101,7 @@ public class FileSupplierConfiguration { Files.inboundAdapter(this.fileSupplierProperties.getDirectory()) .filter(fileListFilter); if (fileInboundChannelAdapterSpecCustomizer != null) { - fileInboundChannelAdapterSpecCustomizer.customize(adapterSpec, "fileMessageSource"); + fileInboundChannelAdapterSpecCustomizer.customize(adapterSpec); } return adapterSpec; } diff --git a/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/AbstractFileSupplierTests.java b/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/AbstractFileSupplierTests.java index e18235e2..aa3c1e5b 100644 --- a/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/AbstractFileSupplierTests.java +++ b/supplier/file-supplier/src/test/java/org/springframework/cloud/fn/supplier/file/AbstractFileSupplierTests.java @@ -65,12 +65,12 @@ public class AbstractFileSupplierTests { @Bean ComponentCustomizer fileInboundChannelAdapterSpecCustomizer() { - return (adapterSpec, beanName) -> adapterSpec.watchEvents(FileReadingMessageSource.WatchEventType.DELETE); + return (adapterSpec) -> adapterSpec.watchEvents(FileReadingMessageSource.WatchEventType.DELETE); } @Bean ComponentCustomizer fakeCustomizer() { - return (date, beanName) -> { + return (date) -> { throw new RuntimeException("Must not happen"); }; } diff --git a/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierConfiguration.java b/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierConfiguration.java index 8f47696e..c2a5ae8f 100644 --- a/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierConfiguration.java +++ b/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2020 the original author or authors. + * Copyright 2015-2022 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. @@ -17,6 +17,7 @@ package org.springframework.cloud.fn.supplier.ftp; import java.util.function.Supplier; +import java.util.regex.Pattern; import org.apache.commons.net.ftp.FTPFile; import org.reactivestreams.Publisher; @@ -24,10 +25,12 @@ import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import reactor.core.scheduler.Schedulers; +import org.springframework.beans.factory.BeanInitializationException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.cloud.fn.common.file.FileConsumerProperties; import org.springframework.cloud.fn.common.file.FileReadingMode; import org.springframework.cloud.fn.common.file.FileUtils; @@ -47,6 +50,7 @@ import org.springframework.integration.ftp.filters.FtpSimplePatternFileListFilte import org.springframework.integration.ftp.inbound.FtpInboundFileSynchronizingMessageSource; import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.util.IntegrationReactiveUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.util.StringUtils; @@ -57,7 +61,7 @@ import org.springframework.util.StringUtils; * @author Artem Bilan * @author Christian Tzolov */ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties({ FtpSupplierProperties.class, FileConsumerProperties.class }) @Import(FtpSessionFactoryConfiguration.class) public class FtpSupplierConfiguration { @@ -66,7 +70,7 @@ public class FtpSupplierConfiguration { private final FileConsumerProperties fileConsumerProperties; - private ConcurrentMetadataStore metadataStore; + private final ConcurrentMetadataStore metadataStore; SessionFactory ftpSessionFactory; @@ -79,6 +83,7 @@ public class FtpSupplierConfiguration { FileConsumerProperties fileConsumerProperties, ConcurrentMetadataStore metadataStore, SessionFactory ftpSessionFactory) { + this.ftpSupplierProperties = ftpSupplierProperties; this.fileConsumerProperties = fileConsumerProperties; this.metadataStore = metadataStore; @@ -86,7 +91,9 @@ public class FtpSupplierConfiguration { } @Bean - public FtpInboundChannelAdapterSpec ftpMessageSource() { + public FtpInboundChannelAdapterSpec ftpMessageSource( + @Nullable ComponentCustomizer ftpInboundChannelAdapterSpecCustomizer) { + FtpInboundChannelAdapterSpec messageSourceBuilder = Ftp.inboundAdapter(ftpSessionFactory) .preserveTimestamp(this.ftpSupplierProperties.isPreserveTimestamp()) .remoteDirectory(this.ftpSupplierProperties.getRemoteDir()) @@ -98,16 +105,21 @@ public class FtpSupplierConfiguration { ChainFileListFilter chainFileListFilter = new ChainFileListFilter<>(); - if (StringUtils.hasText(this.ftpSupplierProperties.getFilenamePattern())) { - chainFileListFilter.addFilter(new FtpSimplePatternFileListFilter(this.ftpSupplierProperties.getFilenamePattern())); + String filenamePattern = this.ftpSupplierProperties.getFilenamePattern(); + Pattern filenameRegex = this.ftpSupplierProperties.getFilenameRegex(); + if (StringUtils.hasText(filenamePattern)) { + chainFileListFilter.addFilter(new FtpSimplePatternFileListFilter(filenamePattern)); } - else if (this.ftpSupplierProperties.getFilenameRegex() != null) { - chainFileListFilter.addFilter(new FtpRegexPatternFileListFilter(this.ftpSupplierProperties.getFilenameRegex())); + else if (filenameRegex != null) { + chainFileListFilter.addFilter(new FtpRegexPatternFileListFilter(filenameRegex)); } chainFileListFilter.addFilter(new FtpPersistentAcceptOnceFileListFilter(this.metadataStore, "ftpSource/")); messageSourceBuilder.filter(chainFileListFilter); + if (ftpInboundChannelAdapterSpecCustomizer != null) { + ftpInboundChannelAdapterSpecCustomizer.customize(messageSourceBuilder); + } return messageSourceBuilder; } @@ -123,19 +135,24 @@ public class FtpSupplierConfiguration { @Bean @ConditionalOnExpression("environment['file.consumer.mode'] != 'ref'") - public Publisher> ftpReadingFlow() { + public Publisher> ftpReadingFlow(FtpInboundFileSynchronizingMessageSource ftpMessageSource) { return FileUtils.enhanceFlowForReadingMode(IntegrationFlows - .from(IntegrationReactiveUtils.messageSourceToFlux(ftpMessageSource().get())), fileConsumerProperties) + .from(IntegrationReactiveUtils.messageSourceToFlux(ftpMessageSource)), fileConsumerProperties) .toReactivePublisher(); } @Bean - public Supplier>> ftpSupplier() { + public Supplier>> ftpSupplier(@Nullable Publisher> ftpReadingFlow) { if (this.fileConsumerProperties.getMode() == FileReadingMode.ref) { return this::ftpMessageFlux; } + else if (ftpReadingFlow != null) { + return () -> Flux.from(ftpReadingFlow); + } else { - return () -> Flux.from(ftpReadingFlow()); + throw new BeanInitializationException( + "Cannot creat 'ftpSupplier' bean: no 'ftpReadingFlow' dependency and is not 'FileReadingMode.ref'."); } } + } diff --git a/supplier/jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierConfiguration.java b/supplier/jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierConfiguration.java index 79c281c9..a10013b1 100644 --- a/supplier/jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierConfiguration.java +++ b/supplier/jdbc-supplier/src/main/java/org/springframework/cloud/fn/supplier/jdbc/JdbcSupplierConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2020 the original author or authors. + * Copyright 2019-2022 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. @@ -26,6 +26,7 @@ import reactor.core.publisher.Flux; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.cloud.fn.splitter.SplitterFunctionConfiguration; import org.springframework.cloud.function.context.PollableBean; import org.springframework.context.annotation.Bean; @@ -33,13 +34,14 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; import org.springframework.integration.core.MessageSource; import org.springframework.integration.jdbc.JdbcPollingChannelAdapter; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** * @author Soby Chacko * @author Artem Bilan */ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(JdbcSupplierProperties.class) @Import(SplitterFunctionConfiguration.class) public class JdbcSupplierConfiguration { @@ -54,20 +56,26 @@ public class JdbcSupplierConfiguration { } @Bean - public MessageSource jdbcMessageSource() { + public MessageSource jdbcMessageSource( + @Nullable ComponentCustomizer jdbcPollingChannelAdapterCustomizer) { + JdbcPollingChannelAdapter jdbcPollingChannelAdapter = new JdbcPollingChannelAdapter(this.dataSource, this.properties.getQuery()); jdbcPollingChannelAdapter.setMaxRows(this.properties.getMaxRows()); jdbcPollingChannelAdapter.setUpdateSql(this.properties.getUpdate()); + if (jdbcPollingChannelAdapterCustomizer != null) { + jdbcPollingChannelAdapterCustomizer.customize(jdbcPollingChannelAdapter); + } return jdbcPollingChannelAdapter; } @Bean(name = "jdbcSupplier") - @PollableBean(splittable = true) + @PollableBean @ConditionalOnProperty(prefix = "jdbc.supplier", name = "split", matchIfMissing = true) - public Supplier>> splittedSupplier(Function, List>> splitterFunction) { + public Supplier>> splittedSupplier(MessageSource jdbcMessageSource, + Function, List>> splitterFunction) { return () -> { - Message received = jdbcMessageSource().receive(); + Message received = jdbcMessageSource.receive(); if (received != null) { return Flux.fromIterable(splitterFunction.apply(received)); // multiple Message> } @@ -79,8 +87,8 @@ public class JdbcSupplierConfiguration { @Bean @ConditionalOnProperty(prefix = "jdbc.supplier", name = "split", havingValue = "false") - public Supplier> jdbcSupplier() { - return () -> jdbcMessageSource().receive(); + public Supplier> jdbcSupplier(MessageSource jdbcMessageSource) { + return jdbcMessageSource::receive; } } diff --git a/supplier/jms-supplier/src/main/java/org/springframework/cloud/fn/supplier/jms/JmsSupplierConfiguration.java b/supplier/jms-supplier/src/main/java/org/springframework/cloud/fn/supplier/jms/JmsSupplierConfiguration.java index 0a89c4e4..8702d8e1 100644 --- a/supplier/jms-supplier/src/main/java/org/springframework/cloud/fn/supplier/jms/JmsSupplierConfiguration.java +++ b/supplier/jms-supplier/src/main/java/org/springframework/cloud/fn/supplier/jms/JmsSupplierConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2022 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. @@ -26,17 +26,19 @@ import reactor.core.publisher.Flux; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.jms.JmsProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.dsl.IntegrationFlows; -import org.springframework.integration.jms.JmsMessageDrivenEndpoint; import org.springframework.integration.jms.dsl.Jms; +import org.springframework.integration.jms.dsl.JmsMessageDrivenChannelAdapterSpec; import org.springframework.jms.listener.AbstractMessageListenerContainer; import org.springframework.jms.listener.DefaultMessageListenerContainer; import org.springframework.jms.listener.SimpleMessageListenerContainer; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(JmsSupplierProperties.class) public class JmsSupplierConfiguration { @@ -50,18 +52,24 @@ public class JmsSupplierConfiguration { private ConnectionFactory connectionFactory; @Bean - public Supplier>> jmsSupplier(Publisher> jmsPublisher, JmsMessageDrivenEndpoint adapter) { - return () -> Flux.from(jmsPublisher) - .doOnSubscribe(subscription -> adapter.start()) - .doOnTerminate(adapter::stop); + public Supplier>> jmsSupplier(Publisher> jmsPublisher) { + return () -> Flux.from(jmsPublisher); } @Bean - public Publisher> jmsPublisher() { - return IntegrationFlows.from( - Jms.messageDrivenChannelAdapter(container()) - .autoStartup(false)) - .toReactivePublisher(); + public Publisher> jmsPublisher( + AbstractMessageListenerContainer container, + @Nullable ComponentCustomizer> + jmsMessageDrivenChannelAdapterSpecCustomizer) { + + JmsMessageDrivenChannelAdapterSpec messageProducerSpec = Jms.messageDrivenChannelAdapter(container); + + if (jmsMessageDrivenChannelAdapterSpecCustomizer != null) { + jmsMessageDrivenChannelAdapterSpecCustomizer.customize(messageProducerSpec); + } + + return IntegrationFlows.from(messageProducerSpec) + .toReactivePublisher(true); } @Bean @@ -111,4 +119,5 @@ public class JmsSupplierConfiguration { } return container; } + } diff --git a/supplier/mail-supplier/src/main/java/org/springframework/cloud/fn/supplier/mail/MailSupplierConfiguration.java b/supplier/mail-supplier/src/main/java/org/springframework/cloud/fn/supplier/mail/MailSupplierConfiguration.java index 1c12e3c4..877a44ce 100644 --- a/supplier/mail-supplier/src/main/java/org/springframework/cloud/fn/supplier/mail/MailSupplierConfiguration.java +++ b/supplier/mail-supplier/src/main/java/org/springframework/cloud/fn/supplier/mail/MailSupplierConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2020-2021 the original author or authors. + * Copyright 2020-2022 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,27 +23,29 @@ import java.util.function.Supplier; import javax.mail.URLName; +import org.reactivestreams.Publisher; + import reactor.core.publisher.Flux; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; -import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.core.MessageSource; -import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.MessageProducerSpec; import org.springframework.integration.dsl.MessageSourceSpec; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.endpoint.ReactiveMessageSourceProducer; import org.springframework.integration.mail.MailHeaders; +import org.springframework.integration.mail.dsl.ImapIdleChannelAdapterSpec; import org.springframework.integration.mail.dsl.Mail; import org.springframework.integration.mail.dsl.MailInboundChannelAdapterSpec; import org.springframework.integration.transformer.support.AbstractHeaderValueMessageProcessor; import org.springframework.integration.transformer.support.HeaderValueMessageProcessor; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -54,7 +56,7 @@ import org.springframework.messaging.Message; * @author Chris Schaefer * @author Soby Chacko */ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(MailSupplierProperties.class) public class MailSupplierConfiguration { @@ -62,28 +64,21 @@ public class MailSupplierConfiguration { private MailSupplierProperties properties; @Bean - public Supplier>> mailSupplier(@Qualifier("mailChannelAdapter") MessageProducerSupport mailChannelAdapter) { - return () -> Flux.from(mailInputChannel()) - .doOnSubscribe(subscription -> mailChannelAdapter.start()); - } + public Publisher> mailInboundFlow(MessageProducerSupport messageProducer) { - @Bean - public IntegrationFlow mailInboundFlow(MessageProducerSupport messageProducer) { return IntegrationFlows.from(messageProducer) .transform(Mail.toStringTransformer(this.properties.getCharset())) - .enrichHeaders(h -> { - h.defaultOverwrite(true) - .header(MailHeaders.TO, arrayToListProcessor(MailHeaders.TO)) - .header(MailHeaders.CC, arrayToListProcessor(MailHeaders.CC)) - .header(MailHeaders.BCC, arrayToListProcessor(MailHeaders.BCC)); - }) - .channel(mailInputChannel()) - .get(); + .enrichHeaders(h -> h + .defaultOverwrite(true) + .header(MailHeaders.TO, arrayToListProcessor(MailHeaders.TO)) + .header(MailHeaders.CC, arrayToListProcessor(MailHeaders.CC)) + .header(MailHeaders.BCC, arrayToListProcessor(MailHeaders.BCC))) + .toReactivePublisher(true); } @Bean - public FluxMessageChannel mailInputChannel() { - return new FluxMessageChannel(); + public Supplier>> mailSupplier(Publisher> messagePublisher) { + return () -> Flux.from(messagePublisher); } private HeaderValueMessageProcessor arrayToListProcessor(final String header) { @@ -99,20 +94,29 @@ public class MailSupplierConfiguration { @Bean("mailChannelAdapter") @ConditionalOnProperty("mail.supplier.idle-imap") - MessageProducerSpec imapIdleProducer() { + MessageProducerSpec imapIdleProducer( + @Nullable ComponentCustomizer imapIdleChannelAdapterSpecCustomizer) { + URLName urlName = this.properties.getUrl(); - return Mail.imapIdleAdapter(urlName.toString()) - .autoStartup(false) + ImapIdleChannelAdapterSpec imapIdleChannelAdapterSpec = Mail.imapIdleAdapter(urlName.toString()) .shouldDeleteMessages(this.properties.isDelete()) .userFlag(this.properties.getUserFlag()) .javaMailProperties(getJavaMailProperties(urlName)) .selectorExpression(this.properties.getExpression()) .shouldMarkMessagesAsRead(this.properties.isMarkAsRead()); + + if (imapIdleChannelAdapterSpecCustomizer != null) { + imapIdleChannelAdapterSpecCustomizer.customize(imapIdleChannelAdapterSpec); + } + return imapIdleChannelAdapterSpec; } @Bean @ConditionalOnProperty(value = "mail.supplier.idle-imap", matchIfMissing = true, havingValue = "false") - MessageSourceSpec mailMessageSource() { + MessageSourceSpec mailMessageSource( + @Nullable ComponentCustomizer> + mailInboundChannelAdapterSpecCustomizer) { + MailInboundChannelAdapterSpec adapterSpec; URLName urlName = this.properties.getUrl(); switch (urlName.getProtocol().toUpperCase()) { @@ -128,18 +132,22 @@ public class MailSupplierConfiguration { throw new IllegalArgumentException( "Unsupported mail protocol: " + urlName.getProtocol()); } - return adapterSpec.javaMailProperties(getJavaMailProperties(urlName)) + adapterSpec.javaMailProperties(getJavaMailProperties(urlName)) .userFlag(this.properties.getUserFlag()) .selectorExpression(this.properties.getExpression()) .shouldDeleteMessages(this.properties.isDelete()); + + if (mailInboundChannelAdapterSpecCustomizer != null) { + mailInboundChannelAdapterSpecCustomizer.customize(adapterSpec); + } + + return adapterSpec; } @Bean("mailChannelAdapter") @ConditionalOnProperty(value = "mail.supplier.idle-imap", matchIfMissing = true, havingValue = "false") MessageProducerSupport mailMessageProducer(MessageSource mailMessageSource) { - final ReactiveMessageSourceProducer reactiveMessageSourceProducer = new ReactiveMessageSourceProducer(mailMessageSource); - reactiveMessageSourceProducer.setAutoStartup(false); - return reactiveMessageSourceProducer; + return new ReactiveMessageSourceProducer(mailMessageSource); } /** @@ -195,4 +203,5 @@ public class MailSupplierConfiguration { javaMailProperties.putAll(this.properties.getJavaMailProperties()); return javaMailProperties; } + } diff --git a/supplier/mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierConfiguration.java b/supplier/mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierConfiguration.java index 0d3f6f37..f63ae1b0 100644 --- a/supplier/mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierConfiguration.java +++ b/supplier/mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/MongodbSupplierConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2019-2021 the original author or authors. + * Copyright 2019-2022 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. @@ -24,6 +24,7 @@ import reactor.core.publisher.Flux; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.cloud.fn.splitter.SplitterFunctionConfiguration; import org.springframework.cloud.function.context.PollableBean; import org.springframework.context.annotation.Bean; @@ -33,6 +34,7 @@ import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.expression.Expression; import org.springframework.expression.common.LiteralExpression; import org.springframework.integration.mongodb.inbound.MongoDbMessageSource; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -44,7 +46,7 @@ import org.springframework.messaging.Message; * @author Artem Bilan * @author David Turanski */ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties({ MongodbSupplierProperties.class }) @Import(SplitterFunctionConfiguration.class) public class MongodbSupplierConfiguration { @@ -59,11 +61,13 @@ public class MongodbSupplierConfiguration { } @Bean(name = "mongodbSupplier") - @PollableBean(splittable = true) + @PollableBean @ConditionalOnProperty(prefix = "mongodb", name = "split", matchIfMissing = true) - public Supplier>> splittedSupplier(Function, List>> splitterFunction) { + public Supplier>> splittedSupplier(MongoDbMessageSource mongoDbSource, + Function, List>> splitterFunction) { + return () -> { - Message received = mongoSource().receive(); + Message received = mongoDbSource.receive(); if (received != null) { return Flux.fromIterable(splitterFunction.apply(received)); // multiple Message> } @@ -75,30 +79,27 @@ public class MongodbSupplierConfiguration { @Bean @ConditionalOnProperty(prefix = "mongodb", name = "split", havingValue = "false") - public Supplier> mongodbSupplier() { - return () -> mongoSource().receive(); + public Supplier> mongodbSupplier(MongoDbMessageSource mongoDbSource) { + return mongoDbSource::receive; } - /** - * The inheritors can consider to override this method for their purpose or just adjust - * options for the returned instance. - * @return a {@link MongoDbMessageSource} instance - */ @Bean - public MongoDbMessageSource mongoDbSource() { + public MongoDbMessageSource mongoDbSource( + @Nullable ComponentCustomizer mongoDbMessageSourceCustomizer) { + Expression queryExpression = (this.properties.getQueryExpression() != null ? this.properties.getQueryExpression() : new LiteralExpression(this.properties.getQuery())); MongoDbMessageSource mongoDbMessageSource = new MongoDbMessageSource(this.mongoTemplate, queryExpression); mongoDbMessageSource.setCollectionNameExpression(new LiteralExpression(this.properties.getCollection())); mongoDbMessageSource.setEntityClass(String.class); + mongoDbMessageSource.setUpdateExpression(this.properties.getUpdateExpression()); + + if (mongoDbMessageSourceCustomizer != null) { + mongoDbMessageSourceCustomizer.customize(mongoDbMessageSource); + } + return mongoDbMessageSource; } - @Bean - public UpdatingMongoDbMessageSource mongoSource() { - return new UpdatingMongoDbMessageSource(mongoDbSource(), this.mongoTemplate, - this.properties.getUpdateExpression()); - } - } diff --git a/supplier/mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/UpdatingMongoDbMessageSource.java b/supplier/mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/UpdatingMongoDbMessageSource.java deleted file mode 100644 index 7f805679..00000000 --- a/supplier/mongodb-supplier/src/main/java/org/springframework/cloud/fn/supplier/mongo/UpdatingMongoDbMessageSource.java +++ /dev/null @@ -1,118 +0,0 @@ -/* - * Copyright 2021-2021 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. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.springframework.cloud.fn.supplier.mongo; - -import java.util.Collections; -import java.util.List; - -import org.springframework.data.mongodb.core.MongoTemplate; -import org.springframework.data.mongodb.core.query.BasicQuery; -import org.springframework.data.mongodb.core.query.BasicUpdate; -import org.springframework.data.mongodb.core.query.Query; -import org.springframework.data.mongodb.core.query.Update; -import org.springframework.expression.EvaluationContext; -import org.springframework.expression.Expression; -import org.springframework.expression.TypeLocator; -import org.springframework.expression.spel.support.StandardTypeLocator; -import org.springframework.integration.endpoint.AbstractMessageSource; -import org.springframework.integration.expression.ExpressionUtils; -import org.springframework.integration.mongodb.inbound.MongoDbMessageSource; -import org.springframework.integration.mongodb.support.MongoHeaders; -import org.springframework.lang.Nullable; -import org.springframework.messaging.Message; -import org.springframework.util.Assert; - -/** - * An {@link AbstractMessageSource} extension for MongoDB updates - * based on the query result from the {@link MongoDbMessageSource} delegate. - * - * @author Artem Bilan - */ -class UpdatingMongoDbMessageSource extends AbstractMessageSource { - - private final MongoDbMessageSource delegate; - - private final MongoTemplate mongoTemplate; - - @Nullable - private final Expression updateExpression; - - private EvaluationContext evaluationContext; - - - UpdatingMongoDbMessageSource(MongoDbMessageSource delegate, MongoTemplate mongoTemplate, - @Nullable Expression updateExpression) { - - this.delegate = delegate; - this.mongoTemplate = mongoTemplate; - this.updateExpression = updateExpression; - } - - @Override - public String getComponentType() { - return "mongo:updating-inbound-channel-adapter"; - } - - @Override - protected void onInit() { - this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); - TypeLocator typeLocator = this.evaluationContext.getTypeLocator(); - if (typeLocator instanceof StandardTypeLocator) { - //Register MongoDB query API package so FQCN can be avoided in query-expression. - ((StandardTypeLocator) typeLocator).registerImport("org.springframework.data.mongodb.core.query"); - } - } - - @Override - protected Object doReceive() { - final Message message = this.delegate.receive(); - if (message != null && this.updateExpression != null) { - String collectionName = message.getHeaders().get(MongoHeaders.COLLECTION_NAME, String.class); - Object payload = message.getPayload(); - List dataToUpdate; - if (payload instanceof List) { - dataToUpdate = (List) payload; - } - else { - dataToUpdate = Collections.singletonList(payload); - } - - for (Object data : dataToUpdate) { - Query query = new BasicQuery((String) data); - - Object value = this.updateExpression.getValue(this.evaluationContext, data); - Assert.notNull(value, "'updateExpression' must not evaluate to null"); - Update update; - if (value instanceof String) { - update = new BasicUpdate((String) value); - } - else if (value instanceof Update) { - update = ((Update) value); - } - else { - throw new IllegalStateException("'updateExpression' must evaluate to String " + - "or org.springframework.data.mongodb.core.query.Update"); - } - - this.mongoTemplate.updateFirst(query, update, collectionName); - } - } - - return message; - } - -} diff --git a/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierConfiguration.java b/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierConfiguration.java index 96be0d7e..4b4cff95 100644 --- a/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierConfiguration.java +++ b/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2017-2021 the original author or authors. + * Copyright 2017-2022 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. @@ -24,6 +24,7 @@ import reactor.core.publisher.Flux; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.cloud.fn.common.mqtt.MqttConfiguration; import org.springframework.cloud.fn.common.mqtt.MqttProperties; import org.springframework.context.annotation.Bean; @@ -33,6 +34,7 @@ import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.mqtt.core.MqttPahoClientFactory; import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter; import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; /** @@ -56,32 +58,39 @@ public class MqttSupplierConfiguration { private BeanFactory beanFactory; @Bean - public Supplier>> mqttSupplier(Publisher> mqttPublisher, MqttPahoMessageDrivenChannelAdapter mqttInbound) { - return () -> Flux.from(mqttPublisher) - .doOnSubscribe(subscription -> mqttInbound.start()) - .doOnTerminate(mqttInbound::stop); + public Supplier>> mqttSupplier(Publisher> mqttPublisher) { + return () -> Flux.from(mqttPublisher); } - private MqttPahoMessageDrivenChannelAdapter mqttInbound() { - MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(properties.getClientId(), - mqttClientFactory, properties.getTopics()); - adapter.setQos(properties.getQos()); - adapter.setConverter(pahoMessageConverter(beanFactory)); + @Bean + public MqttPahoMessageDrivenChannelAdapter mqttInbound( + @Nullable ComponentCustomizer mqttMessageProducerCustomizer) { + + MqttPahoMessageDrivenChannelAdapter adapter = + new MqttPahoMessageDrivenChannelAdapter(this.properties.getClientId(), this.mqttClientFactory, + this.properties.getTopics()); + adapter.setQos(this.properties.getQos()); + adapter.setConverter(pahoMessageConverter(this.beanFactory)); adapter.setAutoStartup(false); + + if (mqttMessageProducerCustomizer != null) { + mqttMessageProducerCustomizer.customize(adapter); + } + return adapter; } @Bean - public Publisher> mqttPublisher() { - return IntegrationFlows.from( - mqttInbound()) - .toReactivePublisher(); + public Publisher> mqttPublisher(MqttPahoMessageDrivenChannelAdapter mqttInbound) { + return IntegrationFlows.from(mqttInbound) + .toReactivePublisher(true); } - public DefaultPahoMessageConverter pahoMessageConverter(BeanFactory beanFactory) { + private DefaultPahoMessageConverter pahoMessageConverter(BeanFactory beanFactory) { DefaultPahoMessageConverter converter = new DefaultPahoMessageConverter(properties.getCharset()); converter.setPayloadAsBytes(properties.isBinary()); converter.setBeanFactory(beanFactory); return converter; } + } diff --git a/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java b/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java index 6506b025..0c3c8739 100644 --- a/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java +++ b/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2021 the original author or authors. + * Copyright 2016-2022 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. @@ -43,11 +43,14 @@ import org.springframework.boot.autoconfigure.amqp.ConnectionFactoryCustomizer; import org.springframework.boot.autoconfigure.amqp.RabbitConnectionFactoryBeanConfigurer; import org.springframework.boot.autoconfigure.amqp.RabbitProperties; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; import org.springframework.core.io.ResourceLoader; import org.springframework.integration.amqp.dsl.Amqp; -import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter; +import org.springframework.integration.amqp.dsl.AmqpInboundChannelAdapterSMLCSpec; import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.retry.interceptor.RetryOperationsInterceptor; import org.springframework.util.Assert; @@ -59,7 +62,9 @@ import org.springframework.util.Assert; * @author Chris Schaefer * @author Roger Perez * @author Chris Bono + * @author Artem Bilan */ +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties(RabbitSupplierProperties.class) public class RabbitSupplierConfiguration implements DisposableBean { @@ -68,12 +73,13 @@ public class RabbitSupplierConfiguration implements DisposableBean { @Override public MessageProperties toMessageProperties(AMQP.BasicProperties source, - Envelope envelope, - String charset) { + Envelope envelope, String charset) { + MessageProperties properties = super.toMessageProperties(source, envelope, charset); properties.setDeliveryMode(null); return properties; } + }; @Autowired @@ -100,7 +106,7 @@ public class RabbitSupplierConfiguration implements DisposableBean { private CachingConnectionFactory ownConnectionFactory; @Bean - public SimpleMessageListenerContainer container() { + public SimpleMessageListenerContainer container(RetryOperationsInterceptor rabbitSourceRetryInterceptor) { ConnectionFactory connectionFactory = this.properties.isOwnConnection() ? buildLocalConnectionFactory() : this.rabbitConnectionFactory; @@ -135,26 +141,31 @@ public class RabbitSupplierConfiguration implements DisposableBean { Assert.noNullElements(queues, "queues cannot have null elements"); container.setQueueNames(queues); if (this.properties.isEnableRetry()) { - container.setAdviceChain(rabbitSourceRetryInterceptor()); + container.setAdviceChain(rabbitSourceRetryInterceptor); } container.setMessagePropertiesConverter(inboundMessagePropertiesConverter); return container; } @Bean - public Publisher> rabbitPublisher(SimpleMessageListenerContainer container) { - return IntegrationFlows.from( - Amqp.inboundAdapter(container) - .autoStartup(false) - .mappedRequestHeaders(properties.getMappedRequestHeaders())) - .toReactivePublisher(); + public Publisher> rabbitPublisher(SimpleMessageListenerContainer container, + @Nullable ComponentCustomizer amqpMessageProducerCustomizer) { + + AmqpInboundChannelAdapterSMLCSpec messageProducerSpec = + Amqp.inboundAdapter(container) + .mappedRequestHeaders(properties.getMappedRequestHeaders()); + + if (amqpMessageProducerCustomizer != null) { + amqpMessageProducerCustomizer.customize(messageProducerSpec); + } + + return IntegrationFlows.from(messageProducerSpec) + .toReactivePublisher(true); } @Bean - public Supplier>> rabbitSupplier(Publisher> rabbitPublisher, AmqpInboundChannelAdapter adapter) { - return () -> Flux.from(rabbitPublisher) - .doOnSubscribe((subscription) -> adapter.start()) - .doOnTerminate(adapter::stop); + public Supplier>> rabbitSupplier(Publisher> rabbitPublisher) { + return () -> Flux.from(rabbitPublisher); } @Bean @@ -168,7 +179,7 @@ public class RabbitSupplierConfiguration implements DisposableBean { } @Override - public void destroy() throws Exception { + public void destroy() { if (this.ownConnectionFactory != null) { this.ownConnectionFactory.destroy(); } @@ -192,13 +203,15 @@ public class RabbitSupplierConfiguration implements DisposableBean { * https://github.com/spring-projects/spring-boot/blob/c820ad01a108d419d8548265b8a34ed7c5591f7c/spring-boot-project/spring-boot-autoconfigure/src/main/java/org/springframework/boot/autoconfigure/amqp/RabbitAutoConfiguration.java#L95 * [UPGRADE_CONSIDERATION] this should stay somewhat in sync w/ the functionality provided by its original source. */ - private CachingConnectionFactory rabbitConnectionFactory(RabbitProperties properties, ResourceLoader resourceLoader, + private static CachingConnectionFactory rabbitConnectionFactory(RabbitProperties properties, + ResourceLoader resourceLoader, ObjectProvider credentialsProvider, ObjectProvider credentialsRefreshService, ObjectProvider connectionFactoryCustomizers) throws Exception { RabbitConnectionFactoryBean connectionFactoryBean = new RabbitConnectionFactoryBean(); - RabbitConnectionFactoryBeanConfigurer connectionFactoryBeanConfigurer = new RabbitConnectionFactoryBeanConfigurer(resourceLoader, properties); + RabbitConnectionFactoryBeanConfigurer connectionFactoryBeanConfigurer = + new RabbitConnectionFactoryBeanConfigurer(resourceLoader, properties); connectionFactoryBeanConfigurer.setCredentialsProvider(credentialsProvider.getIfUnique()); connectionFactoryBeanConfigurer.setCredentialsRefreshService(credentialsRefreshService.getIfUnique()); connectionFactoryBeanConfigurer.configure(connectionFactoryBean); @@ -209,11 +222,13 @@ public class RabbitSupplierConfiguration implements DisposableBean { .forEach((customizer) -> customizer.customize(connectionFactory)); CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory(connectionFactory); - CachingConnectionFactoryConfigurer cachingConnectionFactoryConfigurer = new CachingConnectionFactoryConfigurer(properties); + CachingConnectionFactoryConfigurer cachingConnectionFactoryConfigurer = + new CachingConnectionFactoryConfigurer(properties); cachingConnectionFactoryConfigurer.setConnectionNameStrategy(cf -> "rabbit.supplier.own.connection"); cachingConnectionFactoryConfigurer.configure(cachingConnectionFactory); cachingConnectionFactory.afterPropertiesSet(); return cachingConnectionFactory; } + } diff --git a/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java b/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java index fc499871..7979da61 100644 --- a/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java +++ b/supplier/s3-supplier/src/main/java/org/springframework/cloud/fn/supplier/s3/AwsS3SupplierConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2020 the original author or authors. + * Copyright 2016-2022 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. @@ -27,12 +27,11 @@ import com.amazonaws.services.s3.AmazonS3; import com.amazonaws.services.s3.model.ListObjectsRequest; import com.amazonaws.services.s3.model.S3ObjectSummary; import org.reactivestreams.Publisher; -import org.reactivestreams.Subscription; import reactor.core.publisher.Flux; -import reactor.core.publisher.MonoProcessor; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.cloud.fn.common.config.ComponentCustomizer; import org.springframework.cloud.fn.common.file.FileConsumerProperties; import org.springframework.cloud.fn.common.file.FileUtils; import org.springframework.context.annotation.Bean; @@ -49,6 +48,7 @@ import org.springframework.integration.endpoint.ReactiveMessageSourceProducer; import org.springframework.integration.file.filters.ChainFileListFilter; import org.springframework.integration.metadata.ConcurrentMetadataStore; import org.springframework.integration.util.IntegrationReactiveUtils; +import org.springframework.lang.Nullable; import org.springframework.messaging.Message; import org.springframework.messaging.support.GenericMessage; import org.springframework.util.StringUtils; @@ -57,7 +57,7 @@ import org.springframework.util.StringUtils; * @author Artem Bilan * @author David Turanski */ -@Configuration +@Configuration(proxyBeanMethods = false) @EnableConfigurationProperties({ AwsS3SupplierProperties.class, FileConsumerProperties.class }) public abstract class AwsS3SupplierConfiguration { @@ -77,6 +77,7 @@ public abstract class AwsS3SupplierConfiguration { FileConsumerProperties fileConsumerProperties, AmazonS3 amazonS3, S3SessionFactory s3SessionFactory, ConcurrentMetadataStore metadataStore) { + this.awsS3SupplierProperties = awsS3SupplierProperties; this.fileConsumerProperties = fileConsumerProperties; this.amazonS3 = amazonS3; @@ -88,11 +89,9 @@ public abstract class AwsS3SupplierConfiguration { @ConditionalOnProperty(prefix = "s3.supplier", name = "list-only", havingValue = "false", matchIfMissing = true) static class SynchronizingConfiguration extends AwsS3SupplierConfiguration { - private final MonoProcessor downstreamSubscription = MonoProcessor.create(); - @Bean public Supplier>> s3Supplier(Publisher> s3SupplierFlow) { - return () -> Flux.from(s3SupplierFlow).doOnSubscribe(this.downstreamSubscription::onNext); + return () -> Flux.from(s3SupplierFlow); } @Bean @@ -116,6 +115,7 @@ public abstract class AwsS3SupplierConfiguration { AmazonS3 amazonS3, S3SessionFactory s3SessionFactory, ConcurrentMetadataStore concurrentMetadataStore) { + super(awsS3SupplierProperties, fileConsumerProperties, amazonS3, s3SessionFactory, concurrentMetadataStore); } @@ -123,11 +123,9 @@ public abstract class AwsS3SupplierConfiguration { @Bean public Publisher> s3SupplierFlow(MessageSource s3MessageSource) { return FileUtils.enhanceFlowForReadingMode( - IntegrationFlows.from( - IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource) - .delaySubscription(this.downstreamSubscription)), - fileConsumerProperties) - .toReactivePublisher(); + IntegrationFlows.from(IntegrationReactiveUtils.messageSourceToFlux(s3MessageSource)), + fileConsumerProperties) + .toReactivePublisher(true); } @Bean @@ -146,11 +144,17 @@ public abstract class AwsS3SupplierConfiguration { } @Bean - public MessageSource s3MessageSource(S3InboundFileSynchronizer s3InboundFileSynchronizer) { + public MessageSource s3MessageSource(S3InboundFileSynchronizer s3InboundFileSynchronizer, + @Nullable ComponentCustomizer s3MessageSourceCustomizer) { + S3InboundFileSynchronizingMessageSource s3MessageSource = new S3InboundFileSynchronizingMessageSource( s3InboundFileSynchronizer); s3MessageSource.setLocalDirectory(this.awsS3SupplierProperties.getLocalDir()); s3MessageSource.setAutoCreateLocalDirectory(this.awsS3SupplierProperties.isAutoCreateLocalDir()); + + if (s3MessageSourceCustomizer != null) { + s3MessageSourceCustomizer.customize(s3MessageSource); + } return s3MessageSource; } @@ -163,21 +167,19 @@ public abstract class AwsS3SupplierConfiguration { ListOnlyConfiguration(AwsS3SupplierProperties awsS3SupplierProperties, FileConsumerProperties fileConsumerProperties, AmazonS3 amazonS3, S3SessionFactory s3SessionFactory, ConcurrentMetadataStore metadataStore) { + super(awsS3SupplierProperties, fileConsumerProperties, amazonS3, s3SessionFactory, metadataStore); } - private final MonoProcessor downstreamSubscription = MonoProcessor.create(); - @Bean public Supplier>> s3Supplier(Publisher> s3SupplierFlow) { - return () -> Flux.from(s3SupplierFlow) - .doOnSubscribe(downstreamSubscription::onNext); + return () -> Flux.from(s3SupplierFlow); } @Bean public Publisher> s3SupplierFlow(ReactiveMessageSourceProducer s3ListingProducer) { - return IntegrationFlows.from(s3ListingProducer).split().toReactivePublisher(); + return IntegrationFlows.from(s3ListingProducer).split().toReactivePublisher(true); } @Bean @@ -208,6 +210,7 @@ public abstract class AwsS3SupplierConfiguration { @Bean ReactiveMessageSourceProducer s3ListingMessageProducer(AmazonS3 amazonS3, AwsS3SupplierProperties awsS3SupplierProperties, Predicate filter) { + ListObjectsRequest listObjectsRequest = new ListObjectsRequest(); listObjectsRequest.setBucketName(awsS3SupplierProperties.getRemoteDir()); return new ReactiveMessageSourceProducer( @@ -216,12 +219,9 @@ public abstract class AwsS3SupplierConfiguration { .getObjectSummaries().stream() .filter(filter).collect(Collectors.toList()); return summaryList.isEmpty() ? null : new GenericMessage<>(summaryList); - }) { - @Override - protected void subscribeToPublisher(Publisher> publisher) { - super.subscribeToPublisher(Flux.from(publisher).delaySubscription(downstreamSubscription)); - } - }; + }); } + } + }