From 81dcd668a00c7d4dd7780233d66801f0ab603f99 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. Conflicts: spring-integration-file/src/main/java/org/springframework/integration/file/remote/handler/FileTransferringMessageHandler.java spring-integration-file/src/main/java/org/springframework/integration/file/remote/synchronizer/AbstractInboundFileSynchronizer.java spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParserTests.java spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpsOutboundChannelAdapterParserTests.java spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/OutboundChannelAdapterParserTests.java Resolved. Polishing Polishing - Add WARN For Missing File + Test Conflicts: spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpOutboundTests.java Resolved. JIRA: https://jira.springsource.org/browse/INT-3091 --- .../FileTransferringMessageHandler.java | 99 ++++++++++++------- .../AbstractInboundFileSynchronizer.java | 11 +-- .../FtpOutboundChannelAdapterParserTests.java | 8 +- ...FtpsOutboundChannelAdapterParserTests.java | 8 +- .../ftp/outbound/FtpOutboundTests.java | 67 +++++++++++-- .../OutboundChannelAdapterParserTests.java | 9 +- .../sftp/outbound/SftpOutboundTests.java | 10 +- 7 files changed, 149 insertions(+), 63 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 75a426f1f0..6abdaeeef2 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 java.util.LinkedList; import java.util.List; @@ -36,7 +39,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; /** @@ -68,8 +70,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 = "/"; @@ -106,9 +106,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() { @@ -159,81 +161,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); @@ -254,9 +262,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); @@ -266,7 +273,7 @@ public class FileTransferringMessageHandler extends AbstractMessageHandler { throw new MessagingException("Failed to write to '" + tempFilePath + "' while uploading the file", e); } finally { - fileInputStream.close(); + inputStream.close(); } } @@ -311,4 +318,26 @@ public class FileTransferringMessageHandler extends AbstractMessageHandler { } } } + + 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 476b0b99c7..04ca7c49cf 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 @@ -44,7 +44,7 @@ import org.springframework.util.ObjectUtils; * The implementation should run through any configured * {@link org.springframework.integration.file.filters.FileListFilter}s to * ensure the file entry is acceptable. - * + * * @author Josh Long * @author Mark Fisher * @author Oleg Zhurakousky @@ -86,7 +86,6 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS */ private volatile boolean deleteRemoteFiles; - /** * Create a synchronizer with the {@link SessionFactory} used to acquire {@link Session} instances. */ @@ -132,7 +131,7 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS protected final List filterFiles(F[] files) { return (this.filter != null) ? this.filter.filterFiles(files) : Arrays.asList(files); } - + protected String getTemporaryFileSuffix() { return temporaryFileSuffix; } @@ -179,7 +178,7 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS } return; } - + File localFile = new File(localDirectory, localFileName); if (!localFile.exists()) { String tempFileName = localFile.getAbsolutePath() + this.temporaryFileSuffix; @@ -211,7 +210,7 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS catch (Exception ignored2) { } } - + if (tempFile.renameTo(localFile)) { if (this.deleteRemoteFiles) { session.remove(remoteFilePath); @@ -222,7 +221,7 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS } } } - + private String generateLocalFileName(String remoteFileName){ if (this.localFilenameGeneratorExpression != null){ return this.localFilenameGeneratorExpression.getValue(evaluationContext, remoteFileName, String.class); 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 d89bb50aee..6dcb171a00 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 @@ -16,16 +16,17 @@ package org.springframework.integration.ftp.config; -import static junit.framework.Assert.assertEquals; -import static junit.framework.Assert.assertNotNull; -import static junit.framework.Assert.assertTrue; +import static org.junit.Assert.assertEquals; 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; @@ -70,7 +71,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 0e63927c3a..da2d09130f 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 @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2013 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. @@ -16,9 +16,8 @@ package org.springframework.integration.ftp.config; -import static junit.framework.Assert.assertEquals; -import static junit.framework.Assert.assertNotNull; -import static junit.framework.Assert.assertTrue; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; import org.junit.Test; @@ -46,7 +45,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")); CachingSessionFactory cacheSf = TestUtils.getPropertyValue(handler, "sessionFactory", CachingSessionFactory.class); DefaultFtpsSessionFactory sf = TestUtils.getPropertyValue(cacheSf, "sessionFactory", DefaultFtpsSessionFactory.class); assertEquals("localhost", TestUtils.getPropertyValue(sf, "host")); 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 35fbfd4738..e3f2cbe29c 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 @@ -16,18 +16,28 @@ package org.springframework.integration.ftp.outbound; -import static junit.framework.Assert.assertEquals; -import static junit.framework.Assert.assertFalse; -import static junit.framework.Assert.assertTrue; +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; import java.io.FileOutputStream; import java.io.InputStream; import java.io.OutputStream; -import java.util.*; +import java.util.ArrayList; +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; @@ -35,6 +45,9 @@ import org.junit.Test; 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; import org.springframework.expression.common.LiteralExpression; @@ -47,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; /** @@ -83,8 +97,11 @@ public class FtpOutboundTests { } }); 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 @@ -102,8 +119,11 @@ public class FtpOutboundTests { } }); 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 @@ -130,6 +150,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 c00137efef..3dd0397ee0 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 @@ -16,17 +16,18 @@ package org.springframework.integration.sftp.config; -import static junit.framework.Assert.assertEquals; -import static junit.framework.Assert.assertNotNull; -import static junit.framework.Assert.assertTrue; +import static org.junit.Assert.assertEquals; 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; @@ -74,7 +75,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")); @@ -107,7 +107,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 5575ac3650..d00b7ac193 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 @@ -99,8 +99,11 @@ public class SftpOutboundTests { handler.setFileNameGenerator(fGenerator); handler.setRemoteDirectoryExpression(new LiteralExpression("remote-target-dir")); - 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 @@ -116,8 +119,11 @@ public class SftpOutboundTests { handler.setFileNameGenerator(fGenerator); handler.setRemoteDirectoryExpression(new LiteralExpression("remote-target-dir")); - 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