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)); - } - }; + }); } + } + }