From ae7bc5d4f758befcbcc98889de1c163bd0ba5ac2 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 11 Oct 2013 16:35:27 -0400 Subject: [PATCH] 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 --- .../FileTransferringMessageHandler.java | 98 ++++++++++++------- .../AbstractInboundFileSynchronizer.java | 3 - .../FtpOutboundChannelAdapterParserTests.java | 6 +- ...FtpsOutboundChannelAdapterParserTests.java | 2 - .../ftp/outbound/FtpOutboundTests.java | 53 +++++++++- .../OutboundChannelAdapterParserTests.java | 7 +- .../sftp/outbound/SftpOutboundTests.java | 10 +- 7 files changed, 128 insertions(+), 51 deletions(-) diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java index b314c5d812..c266c1bbd7 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java +++ b/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 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 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 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 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 session) - throws FileNotFoundException, IOException { + private void sendFileToRemoteDirectory(InputStream inputStream, String temporaryRemoteDirectory, + String remoteDirectory, String fileName, Session session) throws FileNotFoundException, IOException { remoteDirectory = this.normalizeDirectoryPath(remoteDirectory); temporaryRemoteDirectory = this.normalizeDirectoryPath(temporaryRemoteDirectory); @@ -253,9 +261,8 @@ public class FileTransferringMessageHandler 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 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 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; + } + + } + } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java index b2bcab056b..cb42377025 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java +++ b/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 implements InboundFileS */ private volatile boolean deleteRemoteFiles; - private volatile BeanFactory beanFactory; - /** * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances. */ diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParserTests.java index 8556c99042..d6b2f8e98a 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParserTests.java +++ b/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; diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsOutboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsOutboundChannelAdapterParserTests.java index 42dfc9bc37..7611b69cb9 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsOutboundChannelAdapterParserTests.java +++ b/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")); diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java index cb088f65da..9514d21cc1 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java +++ b/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("hello")); + handler.handleMessage(new GenericMessage("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("hello".getBytes())); + handler.handleMessage(new GenericMessage("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 handler = new FileTransferringMessageHandler(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 logged = new AtomicReference(); + doAnswer(new Answer(){ + + @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(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"); diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapterParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapterParserTests.java index 0e937b0da6..c9c90a5bbd 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapterParserTests.java +++ b/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")); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java index 8e1b8ec108..aa3f584584 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/outbound/SftpOutboundTests.java +++ b/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("hello")); + handler.handleMessage(new GenericMessage("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("hello".getBytes())); + handler.handleMessage(new GenericMessage("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