INT-3091 Fix Concurrent (S)FTP Transfers

byte[] and String payloads were written to a temporary file.

If the same message is transferred to two different destinations
concurrently, one of the adapters could remove the temporary
file while the other was using it.

Don't use a temporary file for these payloads, simply use the
payload byte[] [or a String.getBytes()] as the InputStream
passed to the Session.write() method.

Polishing

Polishing - Add WARN For Missing File

+ Test

JIRA: https://jira.springsource.org/browse/INT-3091

spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java

@@ -16,10 +16,13 @@
 
 package org.springframework.integration.file.remote.handler;
 
+import java.io.BufferedInputStream;
+import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.io.InputStream;
 
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
@@ -35,7 +38,6 @@ import org.springframework.integration.file.remote.session.SessionFactory;
 import org.springframework.integration.handler.AbstractMessageHandler;
 import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
 import org.springframework.util.Assert;
-import org.springframework.util.FileCopyUtils;
 import org.springframework.util.StringUtils;
 
 /**
@@ -67,8 +69,6 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
 
 	private volatile boolean fileNameGeneratorSet;
 
-	private volatile File temporaryDirectory = new File(System.getProperty("java.io.tmpdir"));
-
 	private volatile String charset = "UTF-8";
 
 	private volatile String remoteFileSeparator = "/";
@@ -105,9 +105,11 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
 		return this.temporaryFileSuffix;
 	}
 
+	/**
+	 * @deprecated This property is no longer used; byte[] and String payloads are written directly
+	 */
+	@Deprecated
 	public void setTemporaryDirectory(File temporaryDirectory) {
-		Assert.notNull(temporaryDirectory, "temporaryDirectory must not be null");
-		this.temporaryDirectory = temporaryDirectory;
 	}
 
 	protected boolean isUseTemporaryFileName() {
@@ -158,81 +160,87 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
 
 	@Override
 	protected void handleMessageInternal(Message<?> message) throws Exception {
-		File file = this.redeemForStorableFile(message);
-		if (file != null && file.exists()) {
+		StreamHolder inputStreamHolder = this.payloadToInputStream(message);
+		if (inputStreamHolder != null) {
 			Session<F> session = this.sessionFactory.getSession();
+			String fileName = inputStreamHolder.getName();
 			try {
 				String remoteDirectory = this.directoryExpressionProcessor.processMessage(message);
 				String temporaryRemoteDirectory = remoteDirectory;
 				if (this.temporaryDirectoryExpressionProcessor != null){
 					temporaryRemoteDirectory = this.temporaryDirectoryExpressionProcessor.processMessage(message);
 				}
-				String fileName = this.fileNameGenerator.generateFileName(message);
-				this.sendFileToRemoteDirectory(file, temporaryRemoteDirectory, remoteDirectory, fileName, session);
+				fileName = this.fileNameGenerator.generateFileName(message);
+				this.sendFileToRemoteDirectory(inputStreamHolder.getStream(), temporaryRemoteDirectory, remoteDirectory, fileName, session);
 			}
 			catch (FileNotFoundException e) {
 				throw new MessageDeliveryException(message,
-						"File [" + file + "] not found in local working directory; it was moved or deleted unexpectedly.", e);
+						"File [" + inputStreamHolder.getName() + "] not found in local working directory; it was moved or deleted unexpectedly.", e);
 			}
 			catch (IOException e) {
 				throw new MessageDeliveryException(message,
-						"Failed to transfer file [" + file + "] from local working directory to remote FTP directory.", e);
+						"Failed to transfer file [" + inputStreamHolder.getName() + " -> " + fileName + "] from local directory to remote directory.", e);
 			}
 			catch (Exception e) {
 				throw new MessageDeliveryException(message,
-						"Error handling message for file [" + file + "]", e);
+						"Error handling message for file [" + inputStreamHolder.getName() + " -> " + fileName + "]", e);
 			}
 			finally {
-				if (!(message.getPayload() instanceof File)) {
-					// we created the File, so we need to delete it
-					if (file.exists()) {
-						try {
-							file.delete();
-						}
-						catch (Throwable t) {
-							// ignore
-						}
-					}
-				}
 				if (session != null) {
 					session.close();
 				}
 			}
 		}
+		else {
+			// A null holder means a File payload that does not exist.
+			if (logger.isWarnEnabled()) {
+				logger.warn("File " + message.getPayload() + " does not exist");
+			}
+		}
 	}
 
-	private File redeemForStorableFile(Message<?> message) throws MessageDeliveryException {
+	private StreamHolder payloadToInputStream(Message<?> message) throws MessageDeliveryException {
 		try {
 			Object payload = message.getPayload();
-			File sendableFile = null;
+			InputStream dataInputStream = null;
+			String name = null;
 			if (payload instanceof File) {
-				sendableFile = (File) payload;
+				File inputFile = (File) payload;
+				if (inputFile.exists()) {
+					dataInputStream = new BufferedInputStream(new FileInputStream(inputFile));
+					name = inputFile.getAbsolutePath();
+				}
 			}
 			else if (payload instanceof byte[] || payload instanceof String) {
-				String tempFileName = this.fileNameGenerator.generateFileName(message) + ".tmp";
-				sendableFile = new File(this.temporaryDirectory, tempFileName); // will only create temp file for String/byte[]
 				byte[] bytes = null;
 				if (payload instanceof String) {
 					bytes = ((String) payload).getBytes(this.charset);
+					name = "String payload";
 				}
 				else {
 					bytes = (byte[]) payload;
+					name = "byte[] payload";
 				}
-				FileCopyUtils.copy(bytes, sendableFile);
+				dataInputStream = new ByteArrayInputStream(bytes);
 			}
 			else {
 				throw new IllegalArgumentException("Unsupported payload type. The only supported payloads are " +
 							"java.io.File, java.lang.String, and byte[]");
 			}
-			return sendableFile;
+			if (dataInputStream == null) {
+				return null;
+			}
+			else {
+				return new StreamHolder(dataInputStream, name);
+			}
 		}
 		catch (Exception e) {
 			throw new MessageDeliveryException(message, "Failed to create sendable file.", e);
 		}
 	}
 
-	private void sendFileToRemoteDirectory(File file, String temporaryRemoteDirectory, String remoteDirectory, String fileName, Session<F> session)
-			throws FileNotFoundException, IOException {
+	private void sendFileToRemoteDirectory(InputStream inputStream, String temporaryRemoteDirectory,
+			String remoteDirectory, String fileName, Session<F> session) throws FileNotFoundException, IOException {
 
 		remoteDirectory = this.normalizeDirectoryPath(remoteDirectory);
 		temporaryRemoteDirectory = this.normalizeDirectoryPath(temporaryRemoteDirectory);
@@ -253,9 +261,8 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
 			}
 		}
 
-		FileInputStream fileInputStream = new FileInputStream(file);
 		try {
-			session.write(fileInputStream, tempFilePath);
+			session.write(inputStream, tempFilePath);
 			// then rename it to its final name if necessary
 			if (useTemporaryFileName){
 			   session.rename(tempFilePath, remoteFilePath);
@@ -265,7 +272,7 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
 			throw new MessagingException("Failed to write to '" + tempFilePath + "' while uploading the file", e);
 		}
 		finally {
-			fileInputStream.close();
+			inputStream.close();
 		}
 	}
 
@@ -279,4 +286,25 @@ public class FileTransferringMessageHandler<F> extends AbstractMessageHandler {
 		return directoryPath;
 	}
 
+	private class StreamHolder {
+
+		private final InputStream stream;
+
+		private final String name;
+
+		private StreamHolder(InputStream stream, String name) {
+			this.stream = stream;
+			this.name = name;
+		}
+
+		public InputStream getStream() {
+			return stream;
+		}
+
+		public String getName() {
+			return name;
+		}
+
+	}
+
 }

spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java

@@ -27,7 +27,6 @@ import java.util.List;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
-import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.InitializingBean;
 import org.springframework.expression.EvaluationContext;
 import org.springframework.expression.Expression;
@@ -90,8 +89,6 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
 	 */
 	private volatile boolean deleteRemoteFiles;
 
-	private volatile BeanFactory beanFactory;
-
 	/**
 	 * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances.
 	 */

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParserTests.java

@@ -17,15 +17,16 @@
 package org.springframework.integration.ftp.config;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
 
 import java.util.Iterator;
 import java.util.Set;
 
 import org.junit.Test;
+
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -71,7 +72,6 @@ public class FtpOutboundChannelAdapterParserTests {
 		assertEquals("UTF-8", TestUtils.getPropertyValue(handler, "charset"));
 		assertNotNull(TestUtils.getPropertyValue(handler, "directoryExpressionProcessor"));
 		assertNotNull(TestUtils.getPropertyValue(handler, "temporaryDirectoryExpressionProcessor"));
-		assertNotNull(TestUtils.getPropertyValue(handler, "temporaryDirectory"));
 		Object sfProperty = TestUtils.getPropertyValue(handler, "sessionFactory");
 		assertEquals(DefaultFtpSessionFactory.class, sfProperty.getClass());
 		DefaultFtpSessionFactory sessionFactory = (DefaultFtpSessionFactory) sfProperty;

spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsOutboundChannelAdapterParserTests.java

@@ -17,7 +17,6 @@
 package org.springframework.integration.ftp.config;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 
 import org.junit.Test;
@@ -47,7 +46,6 @@ public class FtpsOutboundChannelAdapterParserTests {
 		FileTransferringMessageHandler<?> handler = TestUtils.getPropertyValue(consumer, "handler", FileTransferringMessageHandler.class);
 		assertEquals(ac.getBean("fileNameGenerator"), TestUtils.getPropertyValue(handler, "fileNameGenerator"));
 		assertEquals("UTF-8", TestUtils.getPropertyValue(handler, "charset"));
-		assertNotNull(TestUtils.getPropertyValue(handler, "temporaryDirectory"));
 		DefaultFtpsSessionFactory sf = TestUtils.getPropertyValue(handler, "sessionFactory", DefaultFtpsSessionFactory.class);
 		assertEquals("localhost", TestUtils.getPropertyValue(sf, "host"));
 		assertEquals(22, TestUtils.getPropertyValue(sf, "port"));

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

@@ -18,8 +18,11 @@ package org.springframework.integration.ftp.outbound;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.when;
 
 import java.io.File;
@@ -31,7 +34,10 @@ import java.util.Arrays;
 import java.util.Calendar;
 import java.util.Collection;
 import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicReference;
 
+import org.apache.commons.logging.Log;
 import org.apache.commons.net.ftp.FTPClient;
 import org.apache.commons.net.ftp.FTPFile;
 import org.junit.Before;
@@ -40,6 +46,7 @@ import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
 
+import org.springframework.beans.DirectFieldAccessor;
 import org.springframework.beans.factory.BeanFactory;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -53,6 +60,7 @@ import org.springframework.integration.file.remote.handler.FileTransferringMessa
 import org.springframework.integration.ftp.session.AbstractFtpSessionFactory;
 import org.springframework.integration.message.GenericMessage;
 import org.springframework.integration.support.MessageBuilder;
+import org.springframework.integration.test.util.TestUtils;
 import org.springframework.util.FileCopyUtils;
 
 /**
@@ -92,8 +100,11 @@ public class FtpOutboundTests {
 		});
 		handler.setBeanFactory(mock(BeanFactory.class));
 		handler.afterPropertiesSet();
-		handler.handleMessage(new GenericMessage<String>("hello"));
+		handler.handleMessage(new GenericMessage<String>("String data"));
 		assertTrue(file.exists());
+		byte[] inFile = FileCopyUtils.copyToByteArray(file);
+		assertEquals("String data", new String(inFile));
+		file.delete();
 	}
 
 	@Test
@@ -112,8 +123,11 @@ public class FtpOutboundTests {
 		});
 		handler.setBeanFactory(mock(BeanFactory.class));
 		handler.afterPropertiesSet();
-		handler.handleMessage(new GenericMessage<byte[]>("hello".getBytes()));
+		handler.handleMessage(new GenericMessage<byte[]>("byte[] data".getBytes()));
 		assertTrue(file.exists());
+		byte[] inFile = FileCopyUtils.copyToByteArray(file);
+		assertEquals("byte[] data", new String(inFile));
+		file.delete();
 	}
 
 	@Test
@@ -141,6 +155,41 @@ public class FtpOutboundTests {
 		assertTrue("destination file was not created", destFile.exists());
 	}
 
+	@Test
+	public void testHandleMissingFileMessage() throws Exception {
+		File targetDir = new File("remote-target-dir");
+		assertTrue("target directory does not exist: " + targetDir.getName(), targetDir.exists());
+
+		FileTransferringMessageHandler<FTPFile> handler = new FileTransferringMessageHandler<FTPFile>(sessionFactory);
+		handler.setRemoteDirectoryExpression(new LiteralExpression(targetDir.getName()));
+		handler.setFileNameGenerator(new FileNameGenerator() {
+			public String generateFileName(Message<?> message) {
+				return ((File)message.getPayload()).getName() + ".test";
+			}
+		});
+		handler.setBeanFactory(mock(BeanFactory.class));
+		handler.afterPropertiesSet();
+
+		File srcFile = new File(UUID.randomUUID() + ".txt");
+
+		Log logger = spy(TestUtils.getPropertyValue(handler, "logger", Log.class));
+		when(logger.isWarnEnabled()).thenReturn(true);
+		final AtomicReference<String> logged = new AtomicReference<String>();
+		doAnswer(new Answer<Object>(){
+
+			@Override
+			public Object answer(InvocationOnMock invocation) throws Throwable {
+				logged.set((String) invocation.getArguments()[0]);
+				invocation.callRealMethod();
+				return null;
+			}
+		}).when(logger).warn(Mockito.anyString());
+		new DirectFieldAccessor(handler).setPropertyValue("logger", logger);
+		handler.handleMessage(new GenericMessage<File>(srcFile));
+		assertNotNull(logged.get());
+		assertEquals("File " + srcFile.toString() + " does not exist", logged.get());
+	}
+
 	@Test //INT-2275
 	public void testFtpOutboundChannelAdapterInsideChain() throws Exception {
 		File targetDir = new File("remote-target-dir");

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapterParserTests.java

@@ -17,16 +17,17 @@
 package org.springframework.integration.sftp.config;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
 
 import java.util.Iterator;
 import java.util.Set;
 
 import org.junit.Test;
+
 import org.springframework.beans.factory.BeanDefinitionStoreException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.support.ClassPathXmlApplicationContext;
@@ -75,7 +76,6 @@ public class OutboundChannelAdapterParserTests {
 		assertNotNull(TestUtils.getPropertyValue(handler, "temporaryDirectoryExpressionProcessor"));
 		assertEquals(context.getBean("fileNameGenerator"), TestUtils.getPropertyValue(handler, "fileNameGenerator"));
 		assertEquals("UTF-8", TestUtils.getPropertyValue(handler, "charset"));
-		assertNotNull(TestUtils.getPropertyValue(handler, "temporaryDirectory"));
 		CachingSessionFactory<?> sessionFactory = TestUtils.getPropertyValue(handler, "sessionFactory", CachingSessionFactory.class);
 		DefaultSftpSessionFactory clientFactory = TestUtils.getPropertyValue(sessionFactory, "sessionFactory", DefaultSftpSessionFactory.class);
 		assertEquals("localhost", TestUtils.getPropertyValue(clientFactory, "host"));
@@ -108,7 +108,6 @@ public class OutboundChannelAdapterParserTests {
 		String fileNameGeneratorExpression = (String) TestUtils.getPropertyValue(generator, "expression");
 		assertEquals("payload.getName() + '-foo'", fileNameGeneratorExpression);
 		assertEquals("UTF-8", TestUtils.getPropertyValue(handler, "charset"));
-		assertNotNull(TestUtils.getPropertyValue(handler, "temporaryDirectory"));
 		assertNull(TestUtils.getPropertyValue(handler, "temporaryDirectoryExpressionProcessor"));
 
 	}

spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java

@@ -108,8 +108,11 @@ public class SftpOutboundTests {
 		handler.setBeanFactory(mock(BeanFactory.class));
 		handler.afterPropertiesSet();
 
-		handler.handleMessage(new GenericMessage<String>("hello"));
+		handler.handleMessage(new GenericMessage<String>("String data"));
 		assertTrue(new File("remote-target-dir", "foo.txt").exists());
+		byte[] inFile = FileCopyUtils.copyToByteArray(file);
+		assertEquals("String data", new String(inFile));
+		file.delete();
 	}
 
 	@Test
@@ -128,8 +131,11 @@ public class SftpOutboundTests {
 		handler.setBeanFactory(mock(BeanFactory.class));
 		handler.afterPropertiesSet();
 
-		handler.handleMessage(new GenericMessage<byte[]>("hello".getBytes()));
+		handler.handleMessage(new GenericMessage<byte[]>("byte[] data".getBytes()));
 		assertTrue(new File("remote-target-dir", "foo.txt").exists());
+		byte[] inFile = FileCopyUtils.copyToByteArray(file);
+		assertEquals("byte[] data", new String(inFile));
+		file.delete();
 	}
 
 	@Test //INT-2275