Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-3706: Support FileExistModes in (S)FTP Gateway

Browse Source 
JIRA: https://jira.spring.io/browse/INT-3706

Previously, the `mode` attribute on the (S)FTP outbound gateways was ignored.

Take account of the mode on (M)GET and (M)PUT commands.
This commit is contained in:
Gary Russell
2015-05-11 12:58:07 +01:00
committed by Artem Bilan
parent c2ee93f9e1
commit 123b38a008
8 changed files with 222 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-file/src/main/java/org/springframework/integration/file/config/AbstractRemoteFileOutboundGatewayParser.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 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.
@@ -77,6 +77,7 @@ public abstract class AbstractRemoteFileOutboundGatewayParser extends AbstractCo
localFileGeneratorExpressionBuilder.addConstructorArgValue(localFileGeneratorExpression);
builder.addPropertyValue("localFilenameGeneratorExpression", localFileGeneratorExpressionBuilder.getBeanDefinition());
}
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "mode", "fileExistsMode");
return builder;
}

4
spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java
View File

@@ -258,7 +258,9 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations<F>, Initializ
@Override
public String send(final Message<?> message, String subDirectory, FileExistsMode... mode) {
FileExistsMode modeToUse = mode == null || mode.length < 1 ? FileExistsMode.REPLACE : mode[0];
FileExistsMode modeToUse = mode == null || mode.length < 1 || mode[0] == null
? FileExistsMode.REPLACE
: mode[0];
return send(message, subDirectory, modeToUse);
}

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

@@ -21,6 +21,7 @@ import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -42,9 +43,11 @@ import org.springframework.integration.file.remote.RemoteFileTemplate;
import org.springframework.integration.file.remote.SessionCallback;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
@@ -214,6 +217,8 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
private volatile Expression localFilenameGeneratorExpression;
private volatile FileExistsMode fileExistsMode;
public AbstractRemoteFileOutboundGateway(SessionFactory<F> sessionFactory, String command,
String expression) {
Assert.notNull(sessionFactory, "'sessionFactory' cannot be null");
@@ -338,6 +343,19 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
this.localFilenameGeneratorExpression = localFilenameGeneratorExpression;
}
/**
* Determine the action to take when using GET and MGET operations when the file
* already exists locally, or PUT and MPUT when the file exists on the remote
* system.
* @param fileExistsMode the fileExistsMode to set.
* @since 4.2
*/
public void setFileExistsMode(FileExistsMode fileExistsMode) {
this.fileExistsMode = fileExistsMode;
if (FileExistsMode.APPEND.equals(fileExistsMode)) {
this.remoteFileTemplate.setUseTemporaryFileName(false);
}
}
@Override
protected void doInit() {
@@ -494,7 +512,7 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
}
private String doPut(Message<?> requestMessage, String subDirectory) {
String path = this.remoteFileTemplate.send(requestMessage, subDirectory);
String path = this.remoteFileTemplate.send(requestMessage, subDirectory, this.fileExistsMode);
if (path == null) {
throw new MessagingException(requestMessage, "No local file found for " + requestMessage);
}
@@ -661,10 +679,22 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
}
}
File localFile = new File(this.generateLocalDirectory(message, remoteDir), this.generateLocalFileName(message, remoteFilename));
if (!localFile.exists()) {
FileExistsMode fileExistsMode = this.fileExistsMode;
boolean appending = FileExistsMode.APPEND.equals(fileExistsMode);
boolean replacing = FileExistsMode.REPLACE.equals(fileExistsMode);
if (!localFile.exists() || appending || replacing) {
OutputStream outputStream;
String tempFileName = localFile.getAbsolutePath() + this.remoteFileTemplate.getTemporaryFileSuffix();
File tempFile = new File(tempFileName);
BufferedOutputStream outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
if (appending) {
outputStream = new BufferedOutputStream(new FileOutputStream(localFile, true));
}
else {
outputStream = new BufferedOutputStream(new FileOutputStream(tempFile));
}
if (replacing) {
localFile.delete();
}
try {
session.read(remoteFilePath, outputStream);
}
@@ -690,17 +720,22 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
//Ignore it
}
}
if (!tempFile.renameTo(localFile)) {
if (!appending && !tempFile.renameTo(localFile)) {
throw new MessagingException("Failed to rename local file");
}
if (lsFirst && this.options.contains(Option.PRESERVE_TIMESTAMP)) {
localFile.setLastModified(getModified(files[0]));
}
return localFile;
}
else if (FileExistsMode.IGNORE != fileExistsMode) {
throw new MessageHandlingException(message, "Local file " + localFile + " already exists");
}
else {
throw new MessagingException("Local file " + localFile + " already exists");
if (logger.isDebugEnabled()) {
logger.debug("Existing file skipped: " + localFile);
}
}
return localFile;
}
protected List<File> mGet(Message<?> message, Session<F> session, String remoteDirectory,

169
spring-integration-file/src/test/java/org/springframework/integration/file/remote/gateway/RemoteFileOutboundGatewayTests.java
View File

@@ -16,6 +16,7 @@
package org.springframework.integration.file.remote.gateway;
import static org.hamcrest.Matchers.anyOf;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.not;
@@ -29,10 +30,14 @@ import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
@@ -46,6 +51,7 @@ import java.util.concurrent.atomic.AtomicReference;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import org.mockito.ArgumentCaptor;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -58,8 +64,10 @@ import org.springframework.integration.file.remote.RemoteFileTemplate;
import org.springframework.integration.file.remote.handler.FileTransferringMessageHandler;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandlingException;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
@@ -616,6 +624,78 @@ public class RemoteFileOutboundGatewayTests {
out.getHeaders().get(FileHeaders.REMOTE_FILE));
}
@SuppressWarnings("unchecked")
@Test
public void testGetExists() throws Exception {
SessionFactory sessionFactory = mock(SessionFactory.class);
TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway
(sessionFactory, "get", "payload");
gw.setLocalDirectory(new File(this.tmpDir));
gw.afterPropertiesSet();
File outFile = new File(this.tmpDir + "/f1");
FileOutputStream fos = new FileOutputStream(outFile);
fos.write("foo".getBytes());
fos.close();
when(sessionFactory.getSession()).thenReturn(new TestSession() {
@Override
public TestLsEntry[] list(String path) throws IOException {
return new TestLsEntry[]{
new TestLsEntry("f1", 1234, false, false, 12345, "-rw-r--r--")
};
}
@Override
public void read(String source, OutputStream outputStream)
throws IOException {
outputStream.write("testfile".getBytes());
}
});
// default (null)
Message<File> out;
try {
out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
fail("Exception expected");
}
catch (MessageHandlingException e) {
assertThat(e.getMessage(), containsString("already exists"));
}
gw.setFileExistsMode(FileExistsMode.FAIL);
try {
out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
fail("Exception expected");
}
catch (MessageHandlingException e) {
assertThat(e.getMessage(), containsString("already exists"));
}
gw.setFileExistsMode(FileExistsMode.IGNORE);
out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
assertEquals(outFile, out.getPayload());
assertContents("foo", outFile);
gw.setFileExistsMode(FileExistsMode.APPEND);
out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
assertEquals(outFile, out.getPayload());
assertContents("footestfile", outFile);
gw.setFileExistsMode(FileExistsMode.REPLACE);
out = (Message<File>) gw.handleRequestMessage(new GenericMessage<String>("f1"));
assertEquals(outFile, out.getPayload());
assertContents("testfile", outFile);
outFile.delete();
}
private void assertContents(String expected, File outFile) throws Exception {
BufferedReader reader = new BufferedReader(new FileReader(outFile));
assertEquals(expected, reader.readLine());
reader.close();
}
@Test
public void testGetTempFileDelete() throws Exception {
SessionFactory sessionFactory = mock(SessionFactory.class);
@@ -752,7 +832,14 @@ public class RemoteFileOutboundGatewayTests {
SessionFactory<TestLsEntry> sessionFactory = mock(SessionFactory.class);
@SuppressWarnings("unchecked")
Session<TestLsEntry> session = mock(Session.class);
RemoteFileTemplate<TestLsEntry> template = new RemoteFileTemplate<TestLsEntry>(sessionFactory);
RemoteFileTemplate<TestLsEntry> template = new RemoteFileTemplate<TestLsEntry>(sessionFactory) {
@Override
public boolean exists(String path) {
return false;
}
};
template.setRemoteDirectoryExpression(new LiteralExpression("foo/"));
template.setBeanFactory(mock(BeanFactory.class));
template.afterPropertiesSet();
@@ -763,23 +850,85 @@ public class RemoteFileOutboundGatewayTests {
handler.afterPropertiesSet();
gw.afterPropertiesSet();
when(sessionFactory.getSession()).thenReturn(session);
final AtomicReference<String> written = new AtomicReference<String>();
doAnswer(new Answer<Object>() {
@Override
public Object answer(InvocationOnMock invocation) throws Throwable {
written.set((String) invocation.getArguments()[1]);
return null;
}
}).when(session).write(any(InputStream.class), anyString());
Message<String> requestMessage = MessageBuilder.withPayload("hello")
.setHeader(FileHeaders.FILENAME, "bar.txt")
.build();
String path = (String) gw.handleRequestMessage(requestMessage);
assertEquals("foo/bar.txt", path);
ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
verify(session).write(any(InputStream.class), captor.capture());
assertEquals("foo/bar.txt.writing", captor.getValue());
verify(session).rename("foo/bar.txt.writing", "foo/bar.txt");
}
@Test
public void testPutExists() throws Exception {
@SuppressWarnings("unchecked")
SessionFactory<TestLsEntry> sessionFactory = mock(SessionFactory.class);
@SuppressWarnings("unchecked")
Session<TestLsEntry> session = mock(Session.class);
RemoteFileTemplate<TestLsEntry> template = new RemoteFileTemplate<TestLsEntry>(sessionFactory) {
@Override
public boolean exists(String path) {
return true;
}
};
template.setRemoteDirectoryExpression(new LiteralExpression("foo/"));
template.setBeanFactory(mock(BeanFactory.class));
template.afterPropertiesSet();
TestRemoteFileOutboundGateway gw = new TestRemoteFileOutboundGateway(template, "put", null);
FileTransferringMessageHandler<TestLsEntry> handler = new FileTransferringMessageHandler<TestLsEntry>(sessionFactory);
handler.setRemoteDirectoryExpression(new LiteralExpression("foo/"));
handler.setBeanFactory(mock(BeanFactory.class));
handler.afterPropertiesSet();
gw.afterPropertiesSet();
when(sessionFactory.getSession()).thenReturn(session);
Message<String> requestMessage = MessageBuilder.withPayload("hello")
.setHeader(FileHeaders.FILENAME, "bar.txt")
.build();
// default (null) == REPLACE
String path = (String) gw.handleRequestMessage(requestMessage);
assertEquals("foo/bar.txt", path);
ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
verify(session).write(any(InputStream.class), captor.capture());
assertEquals("foo/bar.txt.writing", captor.getValue());
verify(session).rename("foo/bar.txt.writing", "foo/bar.txt");
gw.setFileExistsMode(FileExistsMode.FAIL);
try {
path = (String) gw.handleRequestMessage(requestMessage);
fail("Exception expected");
}
catch (Exception e) {
assertThat(e.getMessage(), containsString("The destination file already exists"));
}
gw.setFileExistsMode(FileExistsMode.REPLACE);
path = (String) gw.handleRequestMessage(requestMessage);
assertEquals("foo/bar.txt", path);
captor = ArgumentCaptor.forClass(String.class);
verify(session, times(2)).write(any(InputStream.class), captor.capture());
assertEquals("foo/bar.txt.writing", captor.getValue());
verify(session, times(2)).rename("foo/bar.txt.writing", "foo/bar.txt");
gw.setFileExistsMode(FileExistsMode.APPEND);
path = (String) gw.handleRequestMessage(requestMessage);
assertEquals("foo/bar.txt", path);
captor = ArgumentCaptor.forClass(String.class);
verify(session).append(any(InputStream.class), captor.capture());
assertEquals("foo/bar.txt", captor.getValue());
gw.setFileExistsMode(FileExistsMode.IGNORE);
path = (String) gw.handleRequestMessage(requestMessage);
assertEquals("foo/bar.txt", path);
// no more writes/appends
verify(session, times(2)).write(any(InputStream.class), anyString());
verify(session, times(1)).append(any(InputStream.class), anyString());
}
@Test
public void testMput() throws Exception {
@SuppressWarnings("unchecked")

1
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests-context.xml
View File

@@ -33,6 +33,7 @@
command-options="-1 -f"
expression="payload"
order="1"
mode="APPEND"
mput-regex=".*">
<int:poller fixed-delay="1000"/>
</int-ftp:outbound-gateway>

2
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java
View File

@@ -39,6 +39,7 @@ import org.springframework.integration.file.filters.SimplePatternFileListFilter;
import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.Command;
import org.springframework.integration.file.remote.gateway.AbstractRemoteFileOutboundGateway.Option;
import org.springframework.integration.file.remote.session.CachingSessionFactory;
import org.springframework.integration.file.support.FileExistsMode;
import org.springframework.integration.ftp.gateway.FtpOutboundGateway;
import org.springframework.integration.handler.ExpressionEvaluatingMessageProcessor;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
@@ -105,6 +106,7 @@ public class FtpOutboundGatewayParserTests {
assertEquals(Long.valueOf(777), sendTimeout);
assertTrue(TestUtils.getPropertyValue(gateway, "requiresReply", Boolean.class));
assertThat(TestUtils.getPropertyValue(gateway, "mputFilter"), Matchers.instanceOf(RegexPatternFileListFilter.class));
assertEquals(FileExistsMode.APPEND, TestUtils.getPropertyValue(gateway, "fileExistsMode"));
}
@Test

9
src/reference/asciidoc/ftp.adoc
View File

@@ -427,11 +427,18 @@ Here is an example of a gateway configured for an ls command...
reply-channel="toSplitter"/>
----
The payload of the message sent to the toSplitter channel is a list of String objects containing the filename of each file.
The payload of the message sent to the toSplitter channel is a list of String objects containing the filename of each
file.
If the `command-options` was omitted, it would be a list of `FileInfo` objects.
Options are provided space-delimited, e.g.
`command-options="-1 -dirs -links"`.
Starting with _version 4.2_, the `GET`, `MGET`, `PUT` and `MPUT` commands support a `FileExistsMode` property (`mode`
when using the namespace support). This affects the behavior when the local file exists (`GET` and `MGET`) or the remote
file exists (`PUT` and `MPUT`). Supported modes are `REPLACE`, `APPEND`, `FAIL` and `IGNORE`.
For backwards compatibility, the default mode for `PUT` and `MPUT` operations is `REPLACE` and for `GET` and `MGET`
operations, the default is `FAIL`.
[[ftp-session-caching]]
=== FTP Session Caching

6
src/reference/asciidoc/sftp.adoc
View File

@@ -497,6 +497,12 @@ If the `command-options` was omitted, it would be a list of `FileInfo` objects.
Options are provided space-delimited, e.g.
`command-options="-1 -dirs -links"`.
Starting with _version 4.2_, the `GET`, `MGET`, `PUT` and `MPUT` commands support a `FileExistsMode` property (`mode`
when using the namespace support). This affects the behavior when the local file exists (`GET` and `MGET`) or the remote
file exists (`PUT` and `MPUT`). Supported modes are `REPLACE`, `APPEND`, `FAIL` and `IGNORE`.
For backwards compatibility, the default mode for `PUT` and `MPUT` operations is `REPLACE` and for `GET` and `MGET`
operations, the default is `FAIL`.
[[sftp-jsch-logging]]
=== SFTP/JSCH Logging