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<ZMQ.Socket>`, 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<WebClientCustomizer>` 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 <cbono@vmware.com>

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> correlationStrategy,
-			ObjectProvider<ReleaseStrategy> releaseStrategy,
-			ObjectProvider<MessageGroupProcessor> messageGroupProcessor,
-			ObjectProvider<MessageGroupStore> messageStore,
-			@Qualifier("outputChannel") MessageChannel outputChannel) {
+			@Nullable CorrelationStrategy correlationStrategy,
+			@Nullable ReleaseStrategy releaseStrategy,
+			@Nullable MessageGroupProcessor messageGroupProcessor,
+			@Nullable MessageGroupStore messageStore,
+			@Qualifier("outputChannel") MessageChannel outputChannel,
+			@Nullable ComponentCustomizer<AggregatorFactoryBean> 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;
 	}