Remove request FILENAME header for MPUT (#8761)

Related to: https://stackoverflow.com/questions/77268009/how-to-use-sftp-spring-integration-mput-with-sftpoutboundgateway-when-file-objec

In some scenarios when the flow starts with a file inbound channel adapter
and then an MPUT operation is performed for remote file outbound gateway,
the populated in the beginning `FileHeaders.FILENAME` is used from
the `DefaultFileNameGenerator` for all the files from local directory to
upload.
Such a behaviour leads only to the last file in the target remote
directory and only with the name from that header.

* Fix the `AbstractRemoteFileOutboundGateway` to remove a `FileHeaders.FILENAME` header
when message is build for specific item from MPUT request.
This way an original local file is used when we upload directory.

spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java

@@ -55,12 +55,13 @@
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.integration.handler.MessageProcessor;
 import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
-import org.springframework.integration.support.MutableMessage;
+import org.springframework.integration.support.MutableMessageBuilder;
 import org.springframework.integration.support.PartialSuccessException;
 import org.springframework.integration.support.utils.IntegrationUtils;
 import org.springframework.lang.Nullable;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHandlingException;
+import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.MessagingException;
 import org.springframework.util.Assert;
 import org.springframework.util.ObjectUtils;
@@ -867,7 +868,8 @@ else if (!(payload instanceof Collection)) {
 		}
 		if (payload instanceof Collection<?> files) {
 			return files.stream()
-					.map(p -> doMput(new MutableMessage<>(p, requestMessage.getHeaders())))
+					.map((filePayload) -> mputItemMessage(filePayload, requestMessage.getHeaders()))
+					.map(this::doMput)
 					.collect(Collectors.toList());
 		}
 		else if (!file.isDirectory()) {
@@ -879,6 +881,13 @@ else if (!file.isDirectory()) {
 		}
 	}
 
+	private static Message<?> mputItemMessage(Object payload, MessageHeaders requestHeaders) {
+		return MutableMessageBuilder.withPayload(payload)
+				.copyHeaders(requestHeaders)
+				.removeHeader(FileHeaders.FILENAME)
+				.build();
+	}
+
 	/**
 	 * Put files from the provided directory to the remote server recursively.
 	 * The message can be consulted to determine some context.
@@ -899,7 +908,7 @@ private List<String> putLocalDirectory(Message<?> requestMessage, File file, Str
 		try {
 			for (File filteredFile : filteredFiles) {
 				if (!filteredFile.isDirectory()) {
-					String path = doPut(new MutableMessage<>(filteredFile, requestMessage.getHeaders()), subDirectory);
+					String path = doPut(mputItemMessage(filteredFile, requestMessage.getHeaders()), subDirectory);
 					if (path != null) {
 						replies.add(path);
 					}

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java

@@ -398,7 +398,13 @@ public void testInt3088MPutNotRecursive() throws IOException {
 		session = TestUtils.getPropertyValue(session, "targetSession", Session.class);
 		FTPClient client = spy(TestUtils.getPropertyValue(session, "client", FTPClient.class));
 		new DirectFieldAccessor(session).setPropertyValue("client", client);
-		this.inboundMPut.send(new GenericMessage<>(getSourceLocalDirectory()));
+		// The FileHeaders.FILENAME is removed by the AbstractRemoteFileOutboundGateway
+		// when MPUT item is prepared for its specific PUT
+		Message<File> mputRequestMessage =
+				MessageBuilder.withPayload(getSourceLocalDirectory())
+						.setHeader(FileHeaders.FILENAME, "inbound_file_name")
+						.build();
+		this.inboundMPut.send(mputRequestMessage);
 		@SuppressWarnings("unchecked")
 		Message<List<String>> out = (Message<List<String>>) this.output.receive(1000);
 		assertThat(out).isNotNull();