Address deprecation warnings

Remove the usage of deprecated `IntegrationFlows` in favor of `IntegrationFlow`

functions/consumer/ftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/ftp/FtpConsumerConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2022 the original author or authors.
+ * Copyright 2015-2023 the original author or authors.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -31,7 +31,6 @@
 import org.springframework.expression.spel.standard.SpelExpressionParser;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
-import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.ftp.dsl.Ftp;
 import org.springframework.integration.ftp.dsl.FtpMessageHandlerSpec;
@@ -54,7 +53,7 @@ public IntegrationFlow ftpInboundFlow(FtpConsumerProperties properties, SessionF
 			@Nullable ComponentCustomizer<FtpMessageHandlerSpec> ftpMessageHandlerSpecCustomizer) {
 
 		IntegrationFlowBuilder integrationFlowBuilder =
-				IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("ftpConsumer"));
+				IntegrationFlow.from(MessageConsumer.class, (gateway) -> gateway.beanName("ftpConsumer"));
 
 		FtpMessageHandlerSpec handlerSpec =
 				Ftp.outboundAdapter(new FtpRemoteFileTemplate(ftpSessionFactory), properties.getMode())

functions/consumer/jdbc-consumer/src/main/java/org/springframework/cloud/fn/consumer/jdbc/JdbcConsumerConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2021 the original author or authors.
+ * Copyright 2020-2023 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.
@@ -48,7 +48,6 @@
 import org.springframework.integration.config.AggregatorFactoryBean;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
-import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.expression.ExpressionUtils;
 import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.jdbc.JdbcMessageHandler;
@@ -130,7 +129,7 @@ IntegrationFlow jdbcConsumerFlow(@Qualifier("aggregator") MessageHandler aggrega
 									JdbcMessageHandler jdbcMessageHandler) {
 
 		final IntegrationFlowBuilder builder =
-				IntegrationFlows.from(Consumer.class, gateway -> gateway.beanName("jdbcConsumer"));
+				IntegrationFlow.from(Consumer.class, gateway -> gateway.beanName("jdbcConsumer"));
 		if (properties.getBatchSize() > 1 || properties.getIdleTimeout() > 0) {
 			builder.handle(aggregator);
 		}

functions/consumer/log-consumer/src/main/java/org/springframework/cloud/fn/consumer/log/LogConsumerConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 the original author or authors.
+ * Copyright 2020-2023 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,7 +22,6 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.integration.dsl.IntegrationFlow;
-import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.messaging.Message;
 
 /**
@@ -41,7 +40,7 @@ public class LogConsumerConfiguration {
 
 	@Bean
 	IntegrationFlow logConsumerFlow(LogConsumerProperties logSinkProperties) {
-		return IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("logConsumer"))
+		return IntegrationFlow.from(MessageConsumer.class, (gateway) -> gateway.beanName("logConsumer"))
 				.log(logSinkProperties.getLevel(), logSinkProperties.getName(), logSinkProperties.getExpression())
 				.nullChannel();
 	}

functions/consumer/sftp-consumer/src/main/java/org/springframework/cloud/fn/consumer/sftp/SftpConsumerConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2022 the original author or authors.
+ * Copyright 2015-2023 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,7 +27,6 @@
 import org.springframework.context.annotation.Import;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
-import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.sftp.dsl.Sftp;
 import org.springframework.integration.sftp.dsl.SftpMessageHandlerSpec;
@@ -51,7 +50,7 @@ public IntegrationFlow ftpOutboundFlow(SftpConsumerProperties properties,
 			@Nullable ComponentCustomizer<SftpMessageHandlerSpec> sftpMessageHandlerSpecCustomizer) {
 
 		IntegrationFlowBuilder integrationFlowBuilder =
-				IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("sftpConsumer"));
+				IntegrationFlow.from(MessageConsumer.class, (gateway) -> gateway.beanName("sftpConsumer"));
 
 		SftpMessageHandlerSpec handlerSpec =
 				Sftp.outboundAdapter(new SftpRemoteFileTemplate(ftpSessionFactory), properties.getMode())

functions/supplier/file-supplier/src/main/java/org/springframework/cloud/fn/supplier/file/FileSupplierConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2020-2022 the original author or authors.
+ * Copyright 2020-2023 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,8 +35,8 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Lazy;
+import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
-import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.integration.file.dsl.FileInboundChannelAdapterSpec;
 import org.springframework.integration.file.dsl.Files;
@@ -120,7 +120,7 @@ public Flux<Message<?>> fileMessageFlux() {
 	@Bean
 	@ConditionalOnExpression("environment['file.consumer.mode'] != 'ref'")
 	public Publisher<Message<Object>> fileReadingFlow() {
-		IntegrationFlowBuilder flowBuilder = IntegrationFlows.from(fileMessageFlux());
+		IntegrationFlowBuilder flowBuilder = IntegrationFlow.from(fileMessageFlux());
 		return FileUtils.enhanceFlowForReadingMode(flowBuilder, this.fileConsumerProperties)
 				.toReactivePublisher();
 	}

functions/supplier/ftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/ftp/FtpSupplierConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2022 the original author or authors.
+ * Copyright 2015-2023 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,7 +39,7 @@
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.Lazy;
-import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.file.filters.ChainFileListFilter;
 import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.ftp.dsl.Ftp;
@@ -136,7 +136,7 @@ public Flux<Message<?>> ftpMessageFlux() {
 	@Bean
 	@ConditionalOnExpression("environment['file.consumer.mode'] != 'ref'")
 	public Publisher<Message<Object>> ftpReadingFlow(FtpInboundFileSynchronizingMessageSource ftpMessageSource) {
-		return FileUtils.enhanceFlowForReadingMode(IntegrationFlows
+		return FileUtils.enhanceFlowForReadingMode(IntegrationFlow
 				.from(IntegrationReactiveUtils.messageSourceToFlux(ftpMessageSource)), fileConsumerProperties)
 				.toReactivePublisher();
 	}

functions/supplier/http-supplier/src/main/java/org/springframework/cloud/fn/supplier/http/HttpSupplierConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2011-2021 the original author or authors.
+ * Copyright 2011-2023 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.
@@ -29,7 +29,7 @@
 import org.springframework.http.HttpStatus;
 import org.springframework.http.MediaType;
 import org.springframework.http.codec.ServerCodecConfigurer;
-import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.expression.ValueExpression;
 import org.springframework.integration.http.support.DefaultHttpHeaderMapper;
 import org.springframework.integration.mapping.HeaderMapper;
@@ -60,7 +60,7 @@ public Publisher<Message<byte[]>> httpSupplierFlow(HttpSupplierProperties httpSu
 			HeaderMapper<HttpHeaders> httpHeaderMapper,
 			ServerCodecConfigurer serverCodecConfigurer) {
 
-		return IntegrationFlows.from(
+		return IntegrationFlow.from(
 						WebFlux.inboundChannelAdapter(httpSupplierProperties.getPathPattern())
 								.requestPayloadType(byte[].class)
 								.statusCodeExpression(new ValueExpression<>(HttpStatus.ACCEPTED))

functions/supplier/jms-supplier/src/main/java/org/springframework/cloud/fn/supplier/jms/JmsSupplierConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2023 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.
@@ -28,7 +28,7 @@
 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.dsl.IntegrationFlow;
 import org.springframework.integration.jms.dsl.Jms;
 import org.springframework.integration.jms.dsl.JmsMessageDrivenChannelAdapterSpec;
 import org.springframework.jms.listener.AbstractMessageListenerContainer;
@@ -67,7 +67,7 @@ public Publisher<Message<byte[]>> jmsPublisher(
 			jmsMessageDrivenChannelAdapterSpecCustomizer.customize(messageProducerSpec);
 		}
 
-		return IntegrationFlows.from(messageProducerSpec)
+		return IntegrationFlow.from(messageProducerSpec)
 				.toReactivePublisher(true);
 	}
 

functions/supplier/mqtt-supplier/src/main/java/org/springframework/cloud/fn/supplier/mqtt/MqttSupplierConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2022 the original author or authors.
+ * Copyright 2017-2023 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.
@@ -30,7 +30,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Import;
-import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
 import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
 import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
@@ -82,7 +82,7 @@ public MqttPahoMessageDrivenChannelAdapter mqttInbound(
 
 	@Bean
 	public Publisher<Message<byte[]>> mqttPublisher(MqttPahoMessageDrivenChannelAdapter mqttInbound) {
-		return IntegrationFlows.from(mqttInbound)
+		return IntegrationFlow.from(mqttInbound)
 				.toReactivePublisher(true);
 	}
 

functions/supplier/rabbit-supplier/src/main/java/org/springframework/cloud/fn/supplier/rabbit/RabbitSupplierConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2016-2022 the original author or authors.
+ * Copyright 2016-2023 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.
@@ -49,7 +49,7 @@
 import org.springframework.core.io.ResourceLoader;
 import org.springframework.integration.amqp.dsl.Amqp;
 import org.springframework.integration.amqp.dsl.AmqpInboundChannelAdapterSMLCSpec;
-import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.retry.interceptor.RetryOperationsInterceptor;
@@ -159,7 +159,7 @@ public Publisher<Message<byte[]>> rabbitPublisher(SimpleMessageListenerContainer
 			amqpMessageProducerCustomizer.customize(messageProducerSpec);
 		}
 
-		return IntegrationFlows.from(messageProducerSpec)
+		return IntegrationFlow.from(messageProducerSpec)
 				.toReactivePublisher(true);
 	}
 

functions/supplier/sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2023 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.
@@ -50,7 +50,6 @@
 import org.springframework.integration.core.MessageSource;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
-import org.springframework.integration.dsl.IntegrationFlows;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.file.FileHeaders;
 import org.springframework.integration.file.filters.ChainFileListFilter;
@@ -211,7 +210,7 @@ public Publisher<Message<Object>> sftpReadingFlow(
 				SftpSupplierProperties sftpSupplierProperties,
 				FileConsumerProperties fileConsumerProperties) {
 
-			return FileUtils.enhanceStreamFlowForReadingMode(IntegrationFlows
+			return FileUtils.enhanceStreamFlowForReadingMode(IntegrationFlow
 							.from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource)
 									.delaySubscription(subscriptionBarrier)
 									.contextWrite(Context.of(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY,
@@ -258,7 +257,7 @@ public Publisher<Message<Object>> sftpReadingFlow(
 				FileConsumerProperties fileConsumerProperties,
 				@Nullable MessageHandler renameRemoteFileHandler) {
 
-			IntegrationFlowBuilder flowBuilder = FileUtils.enhanceFlowForReadingMode(IntegrationFlows
+			IntegrationFlowBuilder flowBuilder = FileUtils.enhanceFlowForReadingMode(IntegrationFlow
 							.from(IntegrationReactiveUtils.messageSourceToFlux(sftpMessageSource)
 									.delaySubscription(subscriptionBarrier)
 									.contextWrite(Context.of(IntegrationReactiveUtils.DELAY_WHEN_EMPTY_KEY,
@@ -367,7 +366,7 @@ public IntegrationFlow listingFlow(MessageProducerSupport listingMessageProducer
 				GenericSelector<Message<?>> duplicateFilter,
 				GenericSelector<String> listOnlyFilter) {
 
-			return IntegrationFlows.from(listingMessageProducer)
+			return IntegrationFlow.from(listingMessageProducer)
 					.split()
 					.transform(lsEntryToStringTransformer)
 					.filter(duplicateFilter)

functions/supplier/tcp-supplier/src/main/java/org/springframework/cloud/fn/supplier/tcp/TcpSupplierConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2015-2021 the original author or authors.
+ * Copyright 2015-2023 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,7 +27,7 @@
 import org.springframework.cloud.fn.common.tcp.TcpConnectionFactoryProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.ip.config.TcpConnectionFactoryFactoryBean;
 import org.springframework.integration.ip.tcp.TcpReceivingChannelAdapter;
@@ -82,7 +82,7 @@ public TcpReceivingChannelAdapter adapter(
 
 	@Bean
 	public Publisher<Message<Object>> tcpSupplierFlow(TcpReceivingChannelAdapter adapter) {
-		return IntegrationFlows.from(adapter)
+		return IntegrationFlow.from(adapter)
 				.headerFilter(IpHeaders.LOCAL_ADDRESS)
 				.toReactivePublisher();
 	}

functions/supplier/websocket-supplier/src/main/java/org/springframework/cloud/fn/supplier/websocket/WebsocketSupplierConfiguration.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2021 the original author or authors.
+ * Copyright 2018-2023 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.
@@ -28,7 +28,7 @@
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.integration.dsl.IntegrationFlows;
+import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.websocket.IntegrationWebSocketContainer;
 import org.springframework.integration.websocket.ServerWebSocketContainer;
 import org.springframework.integration.websocket.inbound.WebSocketInboundChannelAdapter;
@@ -58,7 +58,7 @@ public Supplier<Flux<Message<?>>> websocketSupplier(Publisher<Message<?>> websoc
 
 	@Bean
 	public Publisher<Message<byte[]>> websocketPublisher(IntegrationWebSocketContainer serverWebSocketContainer) {
-		return IntegrationFlows.from(
+		return IntegrationFlow.from(
 				webSocketInboundChannelAdapter(serverWebSocketContainer))
 				.toReactivePublisher();
 	}