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

INT-3204 RemoteFileTemplate Phase II Part 1

Browse Source 
- AbstractInboundFileSynchronizer

INT-3204 RemoteFileTemplate Phase II, Part 2

- AbstractRemoteFileOutboundGateway
- Encapsulate Session.readRaw and .finalizeRaw in template.get()

INT-3204 Polishing - PR Comments

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

INT-3204: Polishing
This commit is contained in:
Gary Russell
2013-11-15 15:13:07 +02:00
committed by Artem Bilan
parent d662e7ee42
commit 1d0c28852f
12 changed files with 446 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
spring-integration-file/src/main/java/org/springframework/integration/file/remote/InputStreamCallback.java Normal file
View File

@@ -0,0 +1,40 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.file.remote;
import java.io.IOException;
import java.io.InputStream;
/**
* Callback for stream-based file retrieval using a RemoteFileOperations.
*
* @author Gary Russell
* @since 3.0
*
*/
public interface InputStreamCallback {
/**
* Called with the InputStream for the remote file. The caller will
* take care of closing the stream and finalizing the file retrieval operation after
* this method exits.
*
* @param stream The InputStream.
* @throws IOException
*/
void doWithInputStream(InputStream stream) throws IOException;
}

45
spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java
View File

@@ -18,12 +18,53 @@ package org.springframework.integration.file.remote;
import org.springframework.integration.Message;
/**
* Strategy for performing operations on remote files.
*
* @author Gary Russell
* @since 3.0
*
*/
public interface RemoteFileOperations {
public interface RemoteFileOperations<F> {
public abstract void send(Message<?> message) throws Exception;
/**
* Send a file to a remote server, based on information in a message.
*
* @param message The message
* @throws Exception
*/
void send(Message<?> message);
/**
* Retrieve a remote file as an InputStream, based on information in a message.
*
* @param callback the callback.
* @return true if the operation was successful.
*/
boolean get(Message<?> message, InputStreamCallback callback);
/**
* Remove a remote file.
*
* @param path The full path to the file.
* @return true when successful
*/
boolean remove(String path);
/**
* Rename a remote file, creating directories if needed.
*
* @param fromPath The current path.
* @param toPath The new path.
*/
void rename(String fromPath, String toPath);
/**
* Execute the callback's doInSession method after obtaining a session.
* Reliably closes the session when the method exits.
*
* @param callback the SessionCallback.
* @return The result of the callback method.
*/
<T> T execute(SessionCallback<F, T> callback);
}

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

@@ -52,10 +52,13 @@ import org.springframework.util.StringUtils;
* @since 3.0
*
*/
public class RemoteFileTemplate<F> implements RemoteFileOperations, InitializingBean, BeanFactoryAware {
public class RemoteFileTemplate<F> implements RemoteFileOperations<F>, InitializingBean, BeanFactoryAware {
private final Log logger = LogFactory.getLog(this.getClass());
/**
* the {@link SessionFactory} for acquiring remote file Sessions.
*/
private final SessionFactory<F> sessionFactory;
private volatile String temporaryFileSuffix =".writing";
@@ -68,6 +71,8 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations, Initializing
private volatile ExpressionEvaluatingMessageProcessor<String> temporaryDirectoryExpressionProcessor;
private volatile ExpressionEvaluatingMessageProcessor<String> fileNameProcessor;
private volatile FileNameGenerator fileNameGenerator = new DefaultFileNameGenerator();
private volatile boolean fileNameGeneratorSet;
@@ -104,6 +109,11 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations, Initializing
this.temporaryDirectoryExpressionProcessor = new ExpressionEvaluatingMessageProcessor<String>(temporaryRemoteDirectoryExpression, String.class);
}
public void setFileNameExpression(Expression fileNameExpression) {
Assert.notNull(fileNameExpression, "fileNameExpression must not be null");
this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor<String>(fileNameExpression, String.class);
}
public String getTemporaryFileSuffix() {
return this.temporaryFileSuffix;
}
@@ -139,16 +149,20 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations, Initializing
@Override
public void afterPropertiesSet() throws Exception {
Assert.notNull(this.directoryExpressionProcessor, "remoteDirectoryExpression is required");
BeanFactory beanFactory = this.beanFactory;
if (beanFactory != null) {
this.directoryExpressionProcessor.setBeanFactory(beanFactory);
if (this.directoryExpressionProcessor != null) {
this.directoryExpressionProcessor.setBeanFactory(beanFactory);
}
if (this.temporaryDirectoryExpressionProcessor != null) {
this.temporaryDirectoryExpressionProcessor.setBeanFactory(beanFactory);
}
if (!this.fileNameGeneratorSet && this.fileNameGenerator instanceof BeanFactoryAware) {
((BeanFactoryAware) this.fileNameGenerator).setBeanFactory(beanFactory);
}
if (this.fileNameProcessor != null) {
this.fileNameProcessor.setBeanFactory(beanFactory);
}
}
if (this.autoCreateDirectory){
Assert.hasText(this.remoteFileSeparator, "'remoteFileSeparator' must not be empty when 'autoCreateDirectory' is set to 'true'");
@@ -159,37 +173,42 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations, Initializing
}
@Override
public void send(Message<?> message) throws Exception {
StreamHolder inputStreamHolder = this.payloadToInputStream(message);
public void send(final Message<?> message) {
Assert.notNull(this.directoryExpressionProcessor, "'remoteDirectoryExpression' is required");
final 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);
this.execute(new SessionCallbackWithoutResult<F>() {
@Override
public void doInSessionWithoutResult(Session<F> session) throws IOException {
String fileName = inputStreamHolder.getName();
try {
String remoteDirectory = RemoteFileTemplate.this.directoryExpressionProcessor
.processMessage(message);
String temporaryRemoteDirectory = remoteDirectory;
if (RemoteFileTemplate.this.temporaryDirectoryExpressionProcessor != null) {
temporaryRemoteDirectory = RemoteFileTemplate.this.temporaryDirectoryExpressionProcessor
.processMessage(message);
}
fileName = RemoteFileTemplate.this.fileNameGenerator.generateFileName(message);
RemoteFileTemplate.this.sendFileToRemoteDirectory(inputStreamHolder.getStream(),
temporaryRemoteDirectory, remoteDirectory, fileName, session);
}
catch (FileNotFoundException e) {
throw new MessageDeliveryException(message, "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 ["
+ inputStreamHolder.getName() + " -> " + fileName
+ "] from local directory to remote directory.", e);
}
catch (Exception e) {
throw new MessageDeliveryException(message, "Error handling message for file ["
+ inputStreamHolder.getName() + " -> " + fileName + "]", e);
}
}
fileName = this.fileNameGenerator.generateFileName(message);
this.sendFileToRemoteDirectory(inputStreamHolder.getStream(), temporaryRemoteDirectory, remoteDirectory, fileName, session);
}
catch (FileNotFoundException e) {
throw new MessageDeliveryException(message,
"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 [" + inputStreamHolder.getName() + " -> " + fileName + "] from local directory to remote directory.", e);
}
catch (Exception e) {
throw new MessageDeliveryException(message,
"Error handling message for file [" + inputStreamHolder.getName() + " -> " + fileName + "]", e);
}
finally {
if (session != null) {
session.close();
}
}
});
}
else {
// A null holder means a File payload that does not exist.
@@ -199,6 +218,79 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations, Initializing
}
}
@Override
public boolean remove(final String path) {
return this.execute(new SessionCallback<F, Boolean>() {
@Override
public Boolean doInSession(Session<F> session) throws IOException {
return session.remove(path);
}
});
}
@Override
public void rename(final String fromPath, final String toPath) {
Assert.hasText(fromPath, "Old filename cannot be null or empty");
Assert.hasText(toPath, "New filename cannot be null or empty");
this.execute(new SessionCallbackWithoutResult<F>() {
@Override
public void doInSessionWithoutResult(Session<F> session) throws IOException {
int lastSeparator = toPath.lastIndexOf(RemoteFileTemplate.this.remoteFileSeparator);
if (lastSeparator > 0) {
String remoteFileDirectory = toPath.substring(0, lastSeparator + 1);
RemoteFileUtils.makeDirectories(remoteFileDirectory, session,
RemoteFileTemplate.this.remoteFileSeparator, RemoteFileTemplate.this.logger);
}
session.rename(fromPath, toPath);
}
});
}
@Override
public boolean get(final Message<?> message, final InputStreamCallback callback) {
Assert.notNull(this.fileNameProcessor, "'fileNameProcessor' needed to use get");
return this.execute(new SessionCallback<F, Boolean>() {
@Override
public Boolean doInSession(Session<F> session) throws IOException {
final String remotePath = RemoteFileTemplate.this.fileNameProcessor.processMessage(message);
InputStream inputStream = session.readRaw(remotePath);
callback.doWithInputStream(inputStream);
inputStream.close();
return session.finalizeRaw();
}
});
}
@Override
public <T> T execute(SessionCallback<F, T> callback) {
Session<F> session = null;
try {
session = this.sessionFactory.getSession();
Assert.notNull(session, "failed to acquire a Session");
return callback.doInSession(session);
}
catch (IOException e) {
throw new MessagingException("Failed to execute on session", e);
}
finally {
if (session != null) {
try {
session.close();
}
catch (Exception ignored) {
if (logger.isDebugEnabled()) {
logger.debug("failed to close Session", ignored);
}
}
}
}
}
private StreamHolder payloadToInputStream(Message<?> message) throws MessageDeliveryException {
try {
Object payload = message.getPayload();
@@ -240,7 +332,7 @@ public class RemoteFileTemplate<F> implements RemoteFileOperations, Initializing
}
private void sendFileToRemoteDirectory(InputStream inputStream, String temporaryRemoteDirectory,
String remoteDirectory, String fileName, Session<F> session) throws FileNotFoundException, IOException {
String remoteDirectory, String fileName, Session<F> session) throws IOException {
remoteDirectory = this.normalizeDirectoryPath(remoteDirectory);
temporaryRemoteDirectory = this.normalizeDirectoryPath(temporaryRemoteDirectory);

43
spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallback.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.file.remote;
import java.io.IOException;
import org.springframework.integration.file.remote.session.Session;
/**
* Callback invoked by {@code RemoteFileOperations.execute()) - allows multiple operations
* on a session.
*
* @author Gary Russell
* @since 3.0
*
*/
public interface SessionCallback<F, T> {
/**
* Called within the context of a session.
* Perform some operation(s) on the session. The caller will take
* care of closing the session after this method exits.
*
* @param session The session.
* @return The result of type T.
* @throws IOException
*/
T doInSession(Session<F> session) throws IOException;
}

48
spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallbackWithoutResult.java Normal file
View File

@@ -0,0 +1,48 @@
/*
* Copyright 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.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.file.remote;
import java.io.IOException;
import org.springframework.integration.file.remote.session.Session;
/**
* Simple convenience implementation of {@link SessionCallback} for cases where
* no result is returned.
*
* @author Gary Russell
* @since 3.0
*
*/
public abstract class SessionCallbackWithoutResult<F> implements SessionCallback<F, Object> {
@Override
public Object doInSession(Session<F> session) throws IOException {
this.doInSessionWithoutResult(session);
return null;
}
/**
* Called within the context of a session.
* Perform some operation(s) on the session. The caller will take
* care of closing the session after this method exits.
*
* @param session The session.
* @throws IOException
*/
protected abstract void doInSessionWithoutResult(Session<F> session) throws IOException;
}

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

@@ -40,7 +40,8 @@ import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.file.FileHeaders;
import org.springframework.integration.file.filters.FileListFilter;
import org.springframework.integration.file.remote.AbstractFileInfo;
import org.springframework.integration.file.remote.RemoteFileUtils;
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.handler.AbstractReplyProducingMessageHandler;
@@ -59,7 +60,7 @@ import org.springframework.util.StringUtils;
*/
public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReplyProducingMessageHandler {
protected final SessionFactory<F> sessionFactory;
private final RemoteFileTemplate<F> remoteFileTemplate;
protected final Command command;
@@ -205,7 +206,8 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
public AbstractRemoteFileOutboundGateway(SessionFactory<F> sessionFactory, String command,
String expression) {
this.sessionFactory = sessionFactory;
Assert.notNull(sessionFactory, "'sessionFactory' cannot be null");
this.remoteFileTemplate = new RemoteFileTemplate<F>(sessionFactory);
this.command = Command.toCommand(command);
this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor<String>(
new SpelExpressionParser().parseExpression(expression));
@@ -213,7 +215,8 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
public AbstractRemoteFileOutboundGateway(SessionFactory<F> sessionFactory, Command command,
String expression) {
this.sessionFactory = sessionFactory;
Assert.notNull(sessionFactory, "'sessionFactory' cannot be null");
this.remoteFileTemplate = new RemoteFileTemplate<F>(sessionFactory);
this.command = command;
this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor<String>(
new SpelExpressionParser().parseExpression(expression));
@@ -330,88 +333,101 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
if (this.getBeanFactory() != null) {
this.fileNameProcessor.setBeanFactory(this.getBeanFactory());
this.renameProcessor.setBeanFactory(this.getBeanFactory());
this.remoteFileTemplate.setBeanFactory(this.getBeanFactory());
}
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
Session<F> session = this.sessionFactory.getSession();
try {
switch (this.command) {
case LS:
return doLs(requestMessage, session);
case GET:
return doGet(requestMessage, session);
case MGET:
return doMget(requestMessage, session);
case RM:
return doRm(requestMessage, session);
case MV:
return doMv(requestMessage, session);
default:
return null;
}
}
catch (IOException e) {
throw new MessagingException(requestMessage, e);
}
finally {
session.close();
switch (this.command) {
case LS:
return doLs(requestMessage);
case GET:
return doGet(requestMessage);
case MGET:
return doMget(requestMessage);
case RM:
return doRm(requestMessage);
case MV:
return doMv(requestMessage);
default:
return null;
}
}
private Object doLs(Message<?> requestMessage, Session<F> session) throws IOException {
private Object doLs(Message<?> requestMessage) {
String dir = this.fileNameProcessor.processMessage(requestMessage);
if (!dir.endsWith(this.remoteFileSeparator)) {
dir += this.remoteFileSeparator;
}
List<?> payload = ls(session, dir);
final String fullDir = dir;
List<?> payload = this.remoteFileTemplate.execute(new SessionCallback<F, List<?>>() {
@Override
public List<?> doInSession(Session<F> session) throws IOException {
return AbstractRemoteFileOutboundGateway.this.ls(session, fullDir);
}
});
return MessageBuilder.withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, dir)
.build();
}
private Object doGet(Message<?> requestMessage, Session<F> session) throws IOException {
String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
String remoteFilename = this.getRemoteFilename(remoteFilePath);
String remoteDir = this.getRemoteDirectory(remoteFilePath, remoteFilename);
File payload = this.get(requestMessage, session, remoteDir, remoteFilePath, remoteFilename, true);
private Object doGet(final Message<?> requestMessage) {
final String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
final String remoteFilename = this.getRemoteFilename(remoteFilePath);
final String remoteDir = this.getRemoteDirectory(remoteFilePath, remoteFilename);
File payload = this.remoteFileTemplate.execute(new SessionCallback<F, File>() {
@Override
public File doInSession(Session<F> session) throws IOException {
return AbstractRemoteFileOutboundGateway.this.get(requestMessage, session, remoteDir, remoteFilePath,
remoteFilename, true);
}
});
return MessageBuilder.withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.build();
}
private Object doMget(Message<?> requestMessage, Session<F> session) throws IOException {
String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
String remoteFilename = this.getRemoteFilename(remoteFilePath);
String remoteDir = this.getRemoteDirectory(remoteFilePath, remoteFilename);
List<File> payload = this.mGet(requestMessage, session, remoteDir, remoteFilename);
private Object doMget(final Message<?> requestMessage) {
final String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
final String remoteFilename = this.getRemoteFilename(remoteFilePath);
final String remoteDir = this.getRemoteDirectory(remoteFilePath, remoteFilename);
List<File> payload = this.remoteFileTemplate.execute(new SessionCallback<F, List<File>>() {
@Override
public List<File> doInSession(Session<F> session) throws IOException {
return AbstractRemoteFileOutboundGateway.this.mGet(requestMessage, session, remoteDir, remoteFilename);
}
});
return MessageBuilder.withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.build();
}
private Object doRm(Message<?> requestMessage, Session<F> session) throws IOException {
String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
private Object doRm(Message<?> requestMessage) {
final String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
String remoteFilename = this.getRemoteFilename(remoteFilePath);
String remoteDir = this.getRemoteDirectory(remoteFilePath, remoteFilename);
boolean payload = this.rm(session, remoteFilePath);
boolean payload = this.remoteFileTemplate.remove(remoteFilePath);
return MessageBuilder.withPayload(payload)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
.build();
}
private Object doMv(Message<?> requestMessage, Session<F> session) throws IOException {
private Object doMv(Message<?> requestMessage) {
String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage);
String remoteFilename = this.getRemoteFilename(remoteFilePath);
String remoteDir = this.getRemoteDirectory(remoteFilePath, remoteFilename);
String remoteFileNewPath = this.renameProcessor.processMessage(requestMessage);
Assert.hasLength(remoteFileNewPath, "New filename cannot be empty");
this.mv(session, remoteFilePath, remoteFileNewPath);
this.remoteFileTemplate.rename(remoteFilePath, remoteFileNewPath);
return MessageBuilder.withPayload(Boolean.TRUE)
.setHeader(FileHeaders.REMOTE_DIRECTORY, remoteDir)
.setHeader(FileHeaders.REMOTE_FILE, remoteFilename)
@@ -660,20 +676,6 @@ public abstract class AbstractRemoteFileOutboundGateway<F> extends AbstractReply
return remoteFileName;
}
protected boolean rm(Session<?> session, String remoteFilePath)
throws IOException {
return session.remove(remoteFilePath);
}
protected void mv(Session<?> session, String remoteFilePath, String remoteFileNewPath) throws IOException {
int lastSeparator = remoteFileNewPath.lastIndexOf(this.remoteFileSeparator);
if (lastSeparator > 0) {
String remoteFileDirectory = remoteFileNewPath.substring(0, lastSeparator + 1);
RemoteFileUtils.makeDirectories(remoteFileDirectory, session, this.remoteFileSeparator, this.logger);
}
session.rename(remoteFilePath, remoteFileNewPath);
}
private File generateLocalDirectory(Message<?> message, String remoteDirectory) {
EvaluationContext evaluationContext = ExpressionUtils.createStandardEvaluationContext(this.getBeanFactory());
evaluationContext.setVariable("remoteDirectory", remoteDirectory);

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

@@ -34,6 +34,8 @@ import org.springframework.expression.Expression;
import org.springframework.integration.MessagingException;
import org.springframework.integration.expression.IntegrationEvaluationContextAware;
import org.springframework.integration.file.filters.FileListFilter;
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.util.Assert;
@@ -59,6 +61,8 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
protected final Log logger = LogFactory.getLog(this.getClass());
private final RemoteFileTemplate<F> remoteFileTemplate;
private volatile EvaluationContext evaluationContext;
private volatile String remoteFileSeparator = "/";
@@ -75,11 +79,6 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
*/
private volatile String remoteDirectory;
/**
* the {@link SessionFactory} for acquiring remote file Sessions.
*/
private final SessionFactory<F> sessionFactory;
/**
* An {@link FileListFilter} that runs against the <em>remote</em> file system view.
*/
@@ -102,7 +101,7 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
*/
public AbstractInboundFileSynchronizer(SessionFactory<F> sessionFactory) {
Assert.notNull(sessionFactory, "sessionFactory must not be null");
this.sessionFactory = sessionFactory;
this.remoteFileTemplate = new RemoteFileTemplate<F>(sessionFactory);
}
@@ -144,6 +143,7 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
this.evaluationContext = evaluationContext;
}
@Override
public final void afterPropertiesSet() {
Assert.notNull(this.remoteDirectory, "remoteDirectory must not be null");
Assert.notNull(this.evaluationContext, "evaluationContext must not be null");
@@ -157,36 +157,37 @@ public abstract class AbstractInboundFileSynchronizer<F> implements InboundFileS
return temporaryFileSuffix;
}
public void synchronizeToLocalDirectory(File localDirectory) {
Session<F> session = null;
@Override
public void synchronizeToLocalDirectory(final File localDirectory) {
try {
session = this.sessionFactory.getSession();
Assert.notNull(session, "failed to acquire a Session");
F[] files = session.list(this.remoteDirectory);
if (!ObjectUtils.isEmpty(files)) {
Collection<F> filteredFiles = this.filterFiles(files);
for (F file : filteredFiles) {
if (file != null) {
this.copyFileToLocalDirectory(this.remoteDirectory, file, localDirectory, session);
int transferred = this.remoteFileTemplate.execute(new SessionCallback<F, Integer>() {
@Override
public Integer doInSession(Session<F> session) throws IOException {
F[] files = session.list(AbstractInboundFileSynchronizer.this.remoteDirectory);
if (!ObjectUtils.isEmpty(files)) {
Collection<F> filteredFiles = AbstractInboundFileSynchronizer.this.filterFiles(files);
for (F file : filteredFiles) {
if (file != null) {
AbstractInboundFileSynchronizer.this.copyFileToLocalDirectory(
AbstractInboundFileSynchronizer.this.remoteDirectory, file, localDirectory,
session);
}
}
return filteredFiles.size();
}
else {
return 0;
}
}
});
if (logger.isDebugEnabled()) {
logger.debug(transferred + " files transferred");
}
}
catch (IOException e) {
catch (Exception e) {
throw new MessagingException("Problem occurred while synchronizing remote to local directory", e);
}
finally {
if (session != null) {
try {
session.close();
}
catch (Exception ignored) {
if (logger.isDebugEnabled()) {
logger.debug("failed to close Session", ignored);
}
}
}
}
}
private void copyFileToLocalDirectory(String remoteDirectoryPath, F remoteFile, File localDirectory, Session<F> session) throws IOException {

7
spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java
View File

@@ -84,7 +84,7 @@ public class FtpInboundChannelAdapterParserTests {
assertEquals("", remoteFileSeparator);
FtpSimplePatternFileListFilter filter = (FtpSimplePatternFileListFilter) TestUtils.getPropertyValue(fisync, "filter");
assertNotNull(filter);
Object sessionFactory = TestUtils.getPropertyValue(fisync, "sessionFactory");
Object sessionFactory = TestUtils.getPropertyValue(fisync, "remoteFileTemplate.sessionFactory");
assertTrue(DefaultFtpSessionFactory.class.isAssignableFrom(sessionFactory.getClass()));
FileListFilter<?> acceptAllFilter = ac.getBean("acceptAllFilter", FileListFilter.class);
assertTrue(TestUtils.getPropertyValue(inbound, "fileSource.scanner.filter.fileFilters", Collection.class).contains(acceptAllFilter));
@@ -107,7 +107,7 @@ public class FtpInboundChannelAdapterParserTests {
ApplicationContext ac = new ClassPathXmlApplicationContext(
"FtpInboundChannelAdapterParserTests-context.xml", this.getClass());
SourcePollingChannelAdapter adapter = ac.getBean("simpleAdapterWithCachedSessions", SourcePollingChannelAdapter.class);
Object sessionFactory = TestUtils.getPropertyValue(adapter, "source.synchronizer.sessionFactory");
Object sessionFactory = TestUtils.getPropertyValue(adapter, "source.synchronizer.remoteFileTemplate.sessionFactory");
assertEquals(CachingSessionFactory.class, sessionFactory.getClass());
FtpInboundFileSynchronizer fisync =
TestUtils.getPropertyValue(adapter, "source.synchronizer", FtpInboundFileSynchronizer.class);
@@ -142,6 +142,7 @@ public class FtpInboundChannelAdapterParserTests {
public static class TestSessionFactoryBean implements FactoryBean<DefaultFtpSessionFactory> {
@Override
@SuppressWarnings({ "rawtypes", "unchecked" })
public DefaultFtpSessionFactory getObject() throws Exception {
DefaultFtpSessionFactory factory = mock(DefaultFtpSessionFactory.class);
@@ -150,10 +151,12 @@ public class FtpInboundChannelAdapterParserTests {
return factory;
}
@Override
public Class<?> getObjectType() {
return DefaultFtpSessionFactory.class;
}
@Override
public boolean isSingleton() {
return true;
}

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

@@ -20,7 +20,6 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.lang.reflect.Method;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
@@ -75,7 +74,7 @@ public class FtpOutboundGatewayParserTests {
FtpOutboundGateway gateway = TestUtils.getPropertyValue(gateway1,
"handler", FtpOutboundGateway.class);
assertEquals("X", TestUtils.getPropertyValue(gateway, "remoteFileSeparator"));
assertNotNull(TestUtils.getPropertyValue(gateway, "sessionFactory"));
assertNotNull(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory"));
assertNotNull(TestUtils.getPropertyValue(gateway, "outputChannel"));
assertEquals("local-test-dir", TestUtils.getPropertyValue(gateway, "localDirectoryExpression.literalValue"));
assertFalse((Boolean) TestUtils.getPropertyValue(gateway, "autoCreateLocalDirectory"));
@@ -97,8 +96,8 @@ public class FtpOutboundGatewayParserTests {
FtpOutboundGateway gateway = TestUtils.getPropertyValue(gateway2,
"handler", FtpOutboundGateway.class);
assertEquals("X", TestUtils.getPropertyValue(gateway, "remoteFileSeparator"));
assertNotNull(TestUtils.getPropertyValue(gateway, "sessionFactory"));
assertTrue(TestUtils.getPropertyValue(gateway, "sessionFactory") instanceof CachingSessionFactory);
assertNotNull(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory"));
assertTrue(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory") instanceof CachingSessionFactory);
assertNotNull(TestUtils.getPropertyValue(gateway, "outputChannel"));
assertEquals("local-test-dir", TestUtils.getPropertyValue(gateway, "localDirectoryExpression.literalValue"));
assertFalse((Boolean) TestUtils.getPropertyValue(gateway, "autoCreateLocalDirectory"));
@@ -130,7 +129,7 @@ public class FtpOutboundGatewayParserTests {
public void testGatewayMv() {
FtpOutboundGateway gateway = TestUtils.getPropertyValue(gateway3,
"handler", FtpOutboundGateway.class);
assertNotNull(TestUtils.getPropertyValue(gateway, "sessionFactory"));
assertNotNull(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory"));
assertNotNull(TestUtils.getPropertyValue(gateway, "outputChannel"));
assertEquals(Command.MV, TestUtils.getPropertyValue(gateway, "command"));
assertEquals("'foo'", TestUtils.getPropertyValue(gateway, "renameProcessor.expression.expression"));

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

@@ -21,21 +21,30 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.mockito.Mockito.mock;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import org.apache.commons.net.ftp.FTPFile;
import org.hamcrest.Matchers;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.expression.spel.standard.SpelExpressionParser;
import org.springframework.integration.Message;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.file.remote.InputStreamCallback;
import org.springframework.integration.file.remote.RemoteFileTemplate;
import org.springframework.integration.file.remote.session.Session;
import org.springframework.integration.file.remote.session.SessionFactory;
import org.springframework.integration.ftp.TesFtpServer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.test.context.ContextConfiguration;
@@ -44,6 +53,8 @@ import org.springframework.util.FileCopyUtils;
/**
* @author Artem Bilan
* @author Gary Russell
*
* @since 3.0
*/
@ContextConfiguration
@@ -51,7 +62,10 @@ import org.springframework.util.FileCopyUtils;
public class FtpServerOutboundTests {
@Autowired
public TesFtpServer ftpServer;
private TesFtpServer ftpServer;
@Autowired
private SessionFactory<FTPFile> ftpSessionFactory;
@Autowired
private PollableChannel output;
@@ -176,7 +190,7 @@ public class FtpServerOutboundTests {
@Test
public void testInt3100RawGET() throws Exception {
Session<?> session = this.ftpServer.ftpSessionFactory().getSession();
Session<?> session = this.ftpSessionFactory.getSession();
ByteArrayOutputStream baos = new ByteArrayOutputStream();
FileCopyUtils.copy(session.readRaw("ftpSource/ftpSource1.txt"), baos);
assertTrue(session.finalizeRaw());
@@ -190,5 +204,32 @@ public class FtpServerOutboundTests {
session.close();
}
@Test
public void testRawGETWithTemplate() throws Exception {
RemoteFileTemplate<FTPFile> template = new RemoteFileTemplate<FTPFile>(this.ftpSessionFactory);
template.setFileNameExpression(new SpelExpressionParser().parseExpression("payload"));
template.setBeanFactory(mock(BeanFactory.class));
template.afterPropertiesSet();
final ByteArrayOutputStream baos1 = new ByteArrayOutputStream();
assertTrue(template.get(new GenericMessage<String>("ftpSource/ftpSource1.txt"), new InputStreamCallback() {
@Override
public void doWithInputStream(InputStream stream) throws IOException {
FileCopyUtils.copy(stream, baos1);
}
}));
assertEquals("source1", new String(baos1.toByteArray()));
final ByteArrayOutputStream baos2 = new ByteArrayOutputStream();
assertTrue(template.get(new GenericMessage<String>("ftpSource/ftpSource2.txt"), new InputStreamCallback() {
@Override
public void doWithInputStream(InputStream stream) throws IOException {
FileCopyUtils.copy(stream, baos2);
}
}));
assertEquals("source2", new String(baos2.toByteArray()));
}
}

4
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserCachingTests.java
View File

@@ -46,7 +46,7 @@ public class InboundChannelAdapterParserCachingTests {
@Test
public void cachingAdapter() {
Object sessionFactory = TestUtils.getPropertyValue(cachingAdapter, "source.synchronizer.sessionFactory");
Object sessionFactory = TestUtils.getPropertyValue(cachingAdapter, "source.synchronizer.remoteFileTemplate.sessionFactory");
assertEquals(CachingSessionFactory.class, sessionFactory.getClass());
Properties sessionConfig = TestUtils.getPropertyValue(sessionFactory, "sessionFactory.sessionConfig", Properties.class);
assertNotNull(sessionConfig);
@@ -55,7 +55,7 @@ public class InboundChannelAdapterParserCachingTests {
@Test
public void nonCachingAdapter() {
Object sessionFactory = TestUtils.getPropertyValue(nonCachingAdapter, "source.synchronizer.sessionFactory");
Object sessionFactory = TestUtils.getPropertyValue(nonCachingAdapter, "source.synchronizer.remoteFileTemplate.sessionFactory");
assertEquals(DefaultSftpSessionFactory.class, sessionFactory.getClass());
}

9
spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java
View File

@@ -20,7 +20,6 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.lang.reflect.Method;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
@@ -73,7 +72,7 @@ public class SftpOutboundGatewayParserTests {
SftpOutboundGateway gateway = TestUtils.getPropertyValue(gateway1,
"handler", SftpOutboundGateway.class);
assertEquals("X", TestUtils.getPropertyValue(gateway, "remoteFileSeparator"));
assertNotNull(TestUtils.getPropertyValue(gateway, "sessionFactory"));
assertNotNull(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory"));
assertNotNull(TestUtils.getPropertyValue(gateway, "outputChannel"));
assertEquals("local-test-dir", TestUtils.getPropertyValue(gateway, "localDirectoryExpression.literalValue"));
assertFalse((Boolean) TestUtils.getPropertyValue(gateway, "autoCreateLocalDirectory"));
@@ -94,8 +93,8 @@ public class SftpOutboundGatewayParserTests {
SftpOutboundGateway gateway = TestUtils.getPropertyValue(gateway2,
"handler", SftpOutboundGateway.class);
assertEquals("X", TestUtils.getPropertyValue(gateway, "remoteFileSeparator"));
assertNotNull(TestUtils.getPropertyValue(gateway, "sessionFactory"));
assertTrue(TestUtils.getPropertyValue(gateway, "sessionFactory") instanceof CachingSessionFactory);
assertNotNull(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory"));
assertTrue(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory") instanceof CachingSessionFactory);
assertNotNull(TestUtils.getPropertyValue(gateway, "outputChannel"));
assertEquals("local-test-dir", TestUtils.getPropertyValue(gateway, "localDirectoryExpression.literalValue"));
assertFalse((Boolean) TestUtils.getPropertyValue(gateway, "autoCreateLocalDirectory"));
@@ -125,7 +124,7 @@ public class SftpOutboundGatewayParserTests {
public void testGatewayMv() {
SftpOutboundGateway gateway = TestUtils.getPropertyValue(gateway3,
"handler", SftpOutboundGateway.class);
assertNotNull(TestUtils.getPropertyValue(gateway, "sessionFactory"));
assertNotNull(TestUtils.getPropertyValue(gateway, "remoteFileTemplate.sessionFactory"));
assertNotNull(TestUtils.getPropertyValue(gateway, "outputChannel"));
assertEquals(Command.MV, TestUtils.getPropertyValue(gateway, "command"));
assertEquals("'foo'", TestUtils.getPropertyValue(gateway, "renameProcessor.expression.expression"));