From 5d4cee05651dee1bcd1116cff179ba4351297b2b Mon Sep 17 00:00:00 2001
From: Artem Bilan <artem.bilan@broadcom.com>
Date: Mon, 8 Jan 2024 12:07:43 -0500
Subject: [PATCH] Move `sftpMessageSource.start()` out of `Publisher`

The `AbstractRemoteFileStreamingMessageSource.start()` does nothing, but state change.
Therefore, it is safe to call from whenever is convenient for us
---
 .../fn/supplier/sftp/SftpSupplierConfiguration.java      | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/supplier/spring-sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java b/supplier/spring-sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java
index 240d0cb3..a34e9c2b 100644
--- a/supplier/spring-sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java
+++ b/supplier/spring-sftp-supplier/src/main/java/org/springframework/cloud/fn/supplier/sftp/SftpSupplierConfiguration.java
@@ -107,11 +107,10 @@ public Supplier<Flux<? extends Message<?>>> sftpSupplier(
 		Flux<? extends Message<?>> flux = (sftpReadingFlow != null) ? Flux.from(sftpReadingFlow)
 				: sftpMessageFlux(sftpMessageSource, sftpSupplierProperties);
 
-		return () -> flux.doOnSubscribe((sub) -> {
-			if (sftpMessageSource instanceof Lifecycle lifecycle) {
-				lifecycle.start();
-			}
-		});
+		if (sftpMessageSource instanceof Lifecycle lifecycle) {
+			lifecycle.start();
+		}
+		return () -> flux;
 	}
 
 	@Bean