diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/InputStreamCallback.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/InputStreamCallback.java new file mode 100644 index 0000000000..47bf6284af --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/InputStreamCallback.java @@ -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; + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java index 1b61d9cb23..3f5ca23732 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileOperations.java @@ -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 { - 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 execute(SessionCallback callback); } diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java index f96fe3697d..02e3918ee2 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/RemoteFileTemplate.java @@ -52,10 +52,13 @@ import org.springframework.util.StringUtils; * @since 3.0 * */ -public class RemoteFileTemplate implements RemoteFileOperations, InitializingBean, BeanFactoryAware { +public class RemoteFileTemplate implements RemoteFileOperations, InitializingBean, BeanFactoryAware { private final Log logger = LogFactory.getLog(this.getClass()); + /** + * the {@link SessionFactory} for acquiring remote file Sessions. + */ private final SessionFactory sessionFactory; private volatile String temporaryFileSuffix =".writing"; @@ -68,6 +71,8 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializing private volatile ExpressionEvaluatingMessageProcessor temporaryDirectoryExpressionProcessor; + private volatile ExpressionEvaluatingMessageProcessor fileNameProcessor; + private volatile FileNameGenerator fileNameGenerator = new DefaultFileNameGenerator(); private volatile boolean fileNameGeneratorSet; @@ -104,6 +109,11 @@ public class RemoteFileTemplate implements RemoteFileOperations, Initializing this.temporaryDirectoryExpressionProcessor = new ExpressionEvaluatingMessageProcessor(temporaryRemoteDirectoryExpression, String.class); } + public void setFileNameExpression(Expression fileNameExpression) { + Assert.notNull(fileNameExpression, "fileNameExpression must not be null"); + this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor(fileNameExpression, String.class); + } + public String getTemporaryFileSuffix() { return this.temporaryFileSuffix; } @@ -139,16 +149,20 @@ public class RemoteFileTemplate 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 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 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() { + + @Override + public void doInSessionWithoutResult(Session 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 implements RemoteFileOperations, Initializing } } + @Override + public boolean remove(final String path) { + return this.execute(new SessionCallback() { + + @Override + public Boolean doInSession(Session 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() { + + @Override + public void doInSessionWithoutResult(Session 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() { + + @Override + public Boolean doInSession(Session 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 execute(SessionCallback callback) { + Session 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 implements RemoteFileOperations, Initializing } private void sendFileToRemoteDirectory(InputStream inputStream, String temporaryRemoteDirectory, - String remoteDirectory, String fileName, Session session) throws FileNotFoundException, IOException { + String remoteDirectory, String fileName, Session session) throws IOException { remoteDirectory = this.normalizeDirectoryPath(remoteDirectory); temporaryRemoteDirectory = this.normalizeDirectoryPath(temporaryRemoteDirectory); diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallback.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallback.java new file mode 100644 index 0000000000..2f0fc06277 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallback.java @@ -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 { + + /** + * 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 session) throws IOException; + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallbackWithoutResult.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallbackWithoutResult.java new file mode 100644 index 0000000000..b13af2c978 --- /dev/null +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/SessionCallbackWithoutResult.java @@ -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 implements SessionCallback { + + @Override + public Object doInSession(Session 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 session) throws IOException; + +} diff --git a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java index bf4fd97df2..e1ece85a17 100644 --- a/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java +++ b/spring-integration-file/src/main/java/org/springframework/integration/file/remote/gateway/AbstractRemoteFileOutboundGateway.java @@ -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 extends AbstractReplyProducingMessageHandler { - protected final SessionFactory sessionFactory; + private final RemoteFileTemplate remoteFileTemplate; protected final Command command; @@ -205,7 +206,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply public AbstractRemoteFileOutboundGateway(SessionFactory sessionFactory, String command, String expression) { - this.sessionFactory = sessionFactory; + Assert.notNull(sessionFactory, "'sessionFactory' cannot be null"); + this.remoteFileTemplate = new RemoteFileTemplate(sessionFactory); this.command = Command.toCommand(command); this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor( new SpelExpressionParser().parseExpression(expression)); @@ -213,7 +215,8 @@ public abstract class AbstractRemoteFileOutboundGateway extends AbstractReply public AbstractRemoteFileOutboundGateway(SessionFactory sessionFactory, Command command, String expression) { - this.sessionFactory = sessionFactory; + Assert.notNull(sessionFactory, "'sessionFactory' cannot be null"); + this.remoteFileTemplate = new RemoteFileTemplate(sessionFactory); this.command = command; this.fileNameProcessor = new ExpressionEvaluatingMessageProcessor( new SpelExpressionParser().parseExpression(expression)); @@ -330,88 +333,101 @@ public abstract class AbstractRemoteFileOutboundGateway 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 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 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>() { + + @Override + public List doInSession(Session 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 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() { + + @Override + public File doInSession(Session 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 session) throws IOException { - String remoteFilePath = this.fileNameProcessor.processMessage(requestMessage); - String remoteFilename = this.getRemoteFilename(remoteFilePath); - String remoteDir = this.getRemoteDirectory(remoteFilePath, remoteFilename); - List 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 payload = this.remoteFileTemplate.execute(new SessionCallback>() { + + @Override + public List doInSession(Session 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 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 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 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); 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 ba725a1283..e89d2a131a 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 @@ -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 implements InboundFileS protected final Log logger = LogFactory.getLog(this.getClass()); + private final RemoteFileTemplate remoteFileTemplate; + private volatile EvaluationContext evaluationContext; private volatile String remoteFileSeparator = "/"; @@ -75,11 +79,6 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS */ private volatile String remoteDirectory; - /** - * the {@link SessionFactory} for acquiring remote file Sessions. - */ - private final SessionFactory sessionFactory; - /** * An {@link FileListFilter} that runs against the remote file system view. */ @@ -102,7 +101,7 @@ public abstract class AbstractInboundFileSynchronizer implements InboundFileS */ public AbstractInboundFileSynchronizer(SessionFactory sessionFactory) { Assert.notNull(sessionFactory, "sessionFactory must not be null"); - this.sessionFactory = sessionFactory; + this.remoteFileTemplate = new RemoteFileTemplate(sessionFactory); } @@ -144,6 +143,7 @@ public abstract class AbstractInboundFileSynchronizer 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 implements InboundFileS return temporaryFileSuffix; } - public void synchronizeToLocalDirectory(File localDirectory) { - Session 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 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() { + + @Override + public Integer doInSession(Session session) throws IOException { + F[] files = session.list(AbstractInboundFileSynchronizer.this.remoteDirectory); + if (!ObjectUtils.isEmpty(files)) { + Collection 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 session) throws IOException { diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java index 4d40e37900..e3340a1289 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java @@ -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 { + @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; } diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java index 4c28c196f4..b691b40e8f 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/config/FtpOutboundGatewayParserTests.java @@ -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")); diff --git a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java index 04d1f849e0..aa2ea8ace9 100644 --- a/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java +++ b/spring-integration-ftp/src/test/java/org/springframework/integration/ftp/outbound/FtpServerOutboundTests.java @@ -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 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 template = new RemoteFileTemplate(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("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("ftpSource/ftpSource2.txt"), new InputStreamCallback() { + + @Override + public void doWithInputStream(InputStream stream) throws IOException { + FileCopyUtils.copy(stream, baos2); + } + })); + assertEquals("source2", new String(baos2.toByteArray())); + } + } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserCachingTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserCachingTests.java index 57c2a659f3..b432292dd0 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserCachingTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/InboundChannelAdapterParserCachingTests.java @@ -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()); } diff --git a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java index c1b69f1712..8604639969 100644 --- a/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java +++ b/spring-integration-sftp/src/test/java/org/springframework/integration/sftp/config/SftpOutboundGatewayParserTests.java @@ -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"));