diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd index 97cb80aec5..f1e7b8ab7b 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd @@ -57,7 +57,8 @@ - + + diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java new file mode 100644 index 0000000000..9deb2f1299 --- /dev/null +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java @@ -0,0 +1,129 @@ +/* + * Copyright 2002-2008 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.adapter.file; + +import java.io.File; +import java.io.IOException; +import java.util.HashMap; +import java.util.Map; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageCreator; +import org.springframework.integration.message.MessageDeliveryAware; +import org.springframework.integration.message.MessagingException; +import org.springframework.integration.message.Source; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * Base class for implementing a Source that creates messages from files in a directory, + * eiter local or remote. + * + * @author Marius Bogoevici + */ +public abstract class AbstractDirectorySource implements Source, MessageDeliveryAware { + + public String FILE_INFO_PROPERTY = "file.info"; + + private final Log logger = LogFactory.getLog(this.getClass()); + + private final DirectoryContentManager directoryContentManager = new DirectoryContentManager(); + + private final MessageCreator messageCreator; + + + public AbstractDirectorySource(MessageCreator messageCreator) { + Assert.notNull(messageCreator, "The MessageCreator must not be null"); + this.messageCreator = messageCreator; + } + + protected DirectoryContentManager getDirectoryContentManager() { + return this.directoryContentManager; + } + + public MessageCreator getMessageCreator() { + return messageCreator; + } + + public final Message receive() { + try { + this.establishConnection(); + HashMap snapshot = new HashMap(); + this.populateSnapshot(snapshot); + this.directoryContentManager.processSnapshot(snapshot); + if (!getDirectoryContentManager().getBacklog().isEmpty()) { + File file = retrieveNextFile(); + Message message = this.messageCreator.createMessage(file); + message.getHeader().setProperty(FileNameGenerator.FILENAME_PROPERTY_KEY, file.getName()); + message.getHeader().setAttribute(FILE_INFO_PROPERTY, + getDirectoryContentManager().getBacklog().get(file.getName())); + return message; + } + return null; + } + catch (Exception e) { + throw new MessagingException("Error while polling for messages.", e); + } + finally { + disconnect(); + } + } + + public void onSend(Message message) { + String filename = message.getHeader().getProperty(FileNameGenerator.FILENAME_PROPERTY_KEY); + if (StringUtils.hasText(filename)) { + this.directoryContentManager.fileProcessed(filename); + } + else if (this.logger.isWarnEnabled()) { + logger.warn("No filename in Message header, cannot send notification of processing."); + } + } + + public void onFailure(MessagingException exception) { + if (this.logger.isWarnEnabled()) { + logger.warn("Failure notification received by " + this.getClass().getSimpleName(), exception); + } + } + + + /** + * Connects to the directory, if necessary. + */ + protected abstract void establishConnection() throws IOException; + + /** + * Constructs the snapshot by iterating files. + * @param snapshot + * @throws IOException + */ + protected abstract void populateSnapshot(Map snapshot) throws IOException; + + /** + * Returns the next file, based on the backlog data. + * @return + * @throws IOException + */ + protected abstract File retrieveNextFile() throws IOException; + + /** + * Disconnects from the directory + */ + protected abstract void disconnect(); + +} \ No newline at end of file diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractFileMessageCreator.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractFileMessageCreator.java index 119d7244fc..5fdd46d742 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractFileMessageCreator.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractFileMessageCreator.java @@ -27,7 +27,15 @@ import org.springframework.integration.message.MessageCreator; import org.springframework.integration.message.MessagingException; /** - * Base class providing common behavior for file-based message creators. + * Base class providing common behavior for file-based message creators. The + * subclasses will redefine the {@code readMessagePayload()} method. This class + * allows to choose between keeping the file after message creation and removing + * it, by setting the appropriate value in the constructor. The desired + * behaviour depends on the nature of the created message (i.e. messages with a + * {@link String} payload can safely remove the file after creation, but + * messages with a {@link File} payload cannot do that) or of the collaborator + * that uses the class instance (e.g. if the file is a locally created copy, it + * can be always discarded). * * @author Mark Fisher * @author Marius Bogoevici @@ -36,15 +44,28 @@ public abstract class AbstractFileMessageCreator implements MessageCreator createMessage(File file) { + public final Message createMessage(File file) { try { T payload = this.readMessagePayload(file); if (payload == null) { return null; } Message message = new GenericMessage(payload); - message.getHeader().setProperty(FileNameGenerator.FILENAME_PROPERTY_KEY, file.getName()); + if (deleteFileAfterCreation) { + file.delete(); + } return message; } catch (Exception e) { diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/ByteArrayFileMessageCreator.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/ByteArrayFileMessageCreator.java index 01b3872535..df278d2153 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/ByteArrayFileMessageCreator.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/ByteArrayFileMessageCreator.java @@ -29,6 +29,16 @@ import org.springframework.util.FileCopyUtils; */ public class ByteArrayFileMessageCreator extends AbstractFileMessageCreator { + /** + * Specifies whether the file should be removed after the message has been created. + * See {@link AbstractFileMessageCreator} + * @param deleteFilesAfterMessageCreation + */ + public ByteArrayFileMessageCreator(boolean deleteFilesAfterMessageCreation) { + super(deleteFilesAfterMessageCreation); + } + + @Override protected byte[] readMessagePayload(File file) throws Exception { return FileCopyUtils.copyToByteArray(file); diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DirectoryContentManager.java similarity index 97% rename from org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java rename to org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DirectoryContentManager.java index f1965cd87e..ea5e7ce3a5 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DirectoryContentManager.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.adapter.ftp; +package org.springframework.integration.adapter.file; import java.util.Collections; import java.util.HashMap; diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FileInfo.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileInfo.java similarity index 96% rename from org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FileInfo.java rename to org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileInfo.java index 3e7fd8de20..d847abaf2a 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FileInfo.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileInfo.java @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.adapter.ftp; +package org.springframework.integration.adapter.file; /** * Information about a file in a directory. @@ -29,7 +29,6 @@ public class FileInfo { private final long size; - public FileInfo(String fileName, long modificationTimestamp, long size) { this.fileName = fileName; this.modificationTimestamp = modificationTimestamp; diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileMessageCreator.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileMessageCreator.java index a2caa78ecc..b0ea68515a 100755 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileMessageCreator.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileMessageCreator.java @@ -29,6 +29,12 @@ import org.springframework.integration.message.MessageCreator; */ public class FileMessageCreator extends AbstractFileMessageCreator { + public FileMessageCreator() { + // The file should never be removed, as just the reference to it is + // passed to the message + super(false); + } + @Override protected File readMessagePayload(File file) throws Exception { return file; diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java index d7dfdd9baa..b9faa61254 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java @@ -19,21 +19,17 @@ package org.springframework.integration.adapter.file; import java.io.File; import java.io.FileFilter; import java.io.FilenameFilter; -import java.util.HashMap; +import java.io.IOException; +import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; - import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.adapter.ftp.DirectoryContentManager; -import org.springframework.integration.adapter.ftp.FileInfo; -import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageCreator; import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.Source; import org.springframework.util.Assert; -import org.springframework.util.StringUtils; /** * A messaging source that polls a directory to retrieve files. @@ -41,9 +37,7 @@ import org.springframework.util.StringUtils; * @author Mark Fisher * @author Marius Bogoevici */ -public class FileSource implements Source, InitializingBean, MessageDeliveryAware { - - private final Log logger = LogFactory.getLog(this.getClass()); +public class FileSource extends AbstractDirectorySource implements Source, InitializingBean, MessageDeliveryAware { private final File directory; @@ -53,18 +47,14 @@ public class FileSource implements Source, InitializingBean, MessageDeli private volatile FilenameFilter filenameFilter; - private final DirectoryContentManager directoryContentManager = new DirectoryContentManager(); - - public FileSource(File directory) { this(directory, new FileMessageCreator()); } public FileSource(File directory, MessageCreator messageCreator) { - Assert.notNull(directory, "directory must not be null"); - Assert.notNull(messageCreator, "MessageCreator must not be null"); + super(messageCreator); + Assert.notNull(directory, "The directory must not be null"); this.directory = directory; - this.messageCreator = messageCreator; } @@ -86,7 +76,18 @@ public class FileSource implements Source, InitializingBean, MessageDeli } } - public Message receive() { + @Override + protected void disconnect() { + // No action is necessary + } + + @Override + protected void establishConnection() throws IOException { + // No action is necessary + } + + @Override + protected void populateSnapshot(Map snapshot) throws IOException { File[] files = null; if (this.fileFilter != null) { files = this.directory.listFiles(this.fileFilter); @@ -100,35 +101,17 @@ public class FileSource implements Source, InitializingBean, MessageDeli if (files == null) { throw new MessagingException("Problem occurred while polling for files. " + "Is '" + directory.getAbsolutePath() + "' a directory?"); - } - HashMap snapshot = new HashMap(); + } for (int i = 0; i < files.length; i++) { FileInfo fileInfo = new FileInfo(files[i].getName(), files[i].lastModified(), files[i].length()); snapshot.put(files[i].getName(), fileInfo); } - this.directoryContentManager.processSnapshot(snapshot); - if (!this.directoryContentManager.getBacklog().isEmpty()) { - String fileName = this.directoryContentManager.getBacklog().keySet().iterator().next(); - File file = new File(directory, fileName); - return this.messageCreator.createMessage(file); - } - return null; } - public void onSend(Message message) { - String filename = message.getHeader().getProperty(FileNameGenerator.FILENAME_PROPERTY_KEY); - if (StringUtils.hasText(filename)) { - this.directoryContentManager.fileProcessed(filename); - } - else if (this.logger.isWarnEnabled()) { - logger.warn("No filename in Message header, cannot send notification of processing."); - } - } - - public void onFailure(MessagingException exception) { - if (this.logger.isWarnEnabled()) { - logger.warn("FileSource received failure notification", exception); - } + @Override + protected File retrieveNextFile() throws IOException { + String fileName = this.getDirectoryContentManager().getBacklog().keySet().iterator().next(); + return new File(directory, fileName); } } diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/TextFileMessageCreator.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/TextFileMessageCreator.java index 2de70c4555..01875edfdd 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/TextFileMessageCreator.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/TextFileMessageCreator.java @@ -30,6 +30,16 @@ import org.springframework.util.FileCopyUtils; */ public class TextFileMessageCreator extends AbstractFileMessageCreator { + /** + * Specifies whether the file should be removed after the message has been created. + * See {@link AbstractFileMessageCreator} + * @param deleteFilesAfterMessageCreation + */ + public TextFileMessageCreator(boolean deleteFilesAfterMessageCreation) { + super(deleteFilesAfterMessageCreation); + } + + @Override protected String readMessagePayload(File file) throws Exception { return FileCopyUtils.copyToString(new FileReader(file)); diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/config/AbstractDirectorySourceParser.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/config/AbstractDirectorySourceParser.java new file mode 100644 index 0000000000..fcafa5388b --- /dev/null +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/config/AbstractDirectorySourceParser.java @@ -0,0 +1,84 @@ +/* + * Copyright 2002-2008 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.adapter.file.config; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.adapter.file.ByteArrayFileMessageCreator; +import org.springframework.integration.adapter.file.FileMessageCreator; +import org.springframework.integration.adapter.file.TextFileMessageCreator; +import org.springframework.util.StringUtils; +import org.w3c.dom.Element; + +/** + * Base class for directory-based sources. + * + * @author Marius Bogoevici + */ +public abstract class AbstractDirectorySourceParser extends AbstractSimpleBeanDefinitionParser { + + private static final String FILE_SOURCE_TYPE_ATTRIBUTE = "file"; + + private static final String TEXT_SOURCE_TYPE_ATTRIBUTE = "text"; + + private static final String BINARY_SOURCE_TYPE_ATTRIBUTE = "binary"; + + public static final String MESSAGE_CREATOR_REFERENCE_ATTRIBUTE = "message-creator"; + + public static final String TYPE_ATTRIBUTE = "type"; + + + private final boolean deleteFileAfterMessageCreation; + + public AbstractDirectorySourceParser(boolean deleteFileAfterMessageCreation){ + this.deleteFileAfterMessageCreation = deleteFileAfterMessageCreation; + } + + + @Override + protected boolean isEligibleAttribute(String attributeName) { + return !( MESSAGE_CREATOR_REFERENCE_ATTRIBUTE.equals(attributeName) + || TYPE_ATTRIBUTE.equals(attributeName)) + && super.isEligibleAttribute(attributeName); + } + + @Override + protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) { + String messageCreatorReference = element.getAttribute(MESSAGE_CREATOR_REFERENCE_ATTRIBUTE); + String type = element.getAttribute(TYPE_ATTRIBUTE); + if (StringUtils.hasText(type) && StringUtils.hasText(messageCreatorReference)) { + throw new ConfigurationException( + "Either the 'type' or the 'message-creator' attributes are allowed, but not both"); + } + if (StringUtils.hasText(messageCreatorReference)) { + beanDefinition.addConstructorArgReference(messageCreatorReference); + } + else { + if (!StringUtils.hasText(type) || FILE_SOURCE_TYPE_ATTRIBUTE.equals(type)) { + beanDefinition.addConstructorArgValue(new FileMessageCreator()); + } + else if (TEXT_SOURCE_TYPE_ATTRIBUTE.equals(type)) { + beanDefinition.addConstructorArgValue(new TextFileMessageCreator(deleteFileAfterMessageCreation)); + } + else if (BINARY_SOURCE_TYPE_ATTRIBUTE.equals(type)) { + beanDefinition.addConstructorArgValue(new ByteArrayFileMessageCreator(deleteFileAfterMessageCreation)); + } + } + } + +} diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/config/FileSourceParser.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/config/FileSourceParser.java index 6221785f50..2586f818e3 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/config/FileSourceParser.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/config/FileSourceParser.java @@ -17,13 +17,7 @@ package org.springframework.integration.adapter.file.config; import org.springframework.beans.factory.support.BeanDefinitionBuilder; -import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; -import org.springframework.integration.ConfigurationException; -import org.springframework.integration.adapter.file.ByteArrayFileMessageCreator; -import org.springframework.integration.adapter.file.FileMessageCreator; import org.springframework.integration.adapter.file.FileSource; -import org.springframework.integration.adapter.file.TextFileMessageCreator; -import org.springframework.util.StringUtils; import org.w3c.dom.Element; /** @@ -32,20 +26,13 @@ import org.w3c.dom.Element; * @author Mark Fisher * @author Marius Bogoevici */ -public class FileSourceParser extends AbstractSimpleBeanDefinitionParser { - - private static final String FILE_SOURCE_TYPE_ATTRIBUTE = "file"; - - private static final String TEXT_SOURCE_TYPE_ATTRIBUTE = "text"; - - private static final String BINARY_SOURCE_TYPE_ATTRIBUTE = "binary"; +public class FileSourceParser extends AbstractDirectorySourceParser { public static final String DIRECTORY_ATTRIBUTE = "directory"; - public static final String MESSAGE_CREATOR_REFERENCE_ATTRIBUTE = "message-creator"; - - public static final String TYPE_ATTRIBUTE = "type"; - + public FileSourceParser() { + super(false); + } @Override protected Class getBeanClass(Element element) { @@ -54,34 +41,13 @@ public class FileSourceParser extends AbstractSimpleBeanDefinitionParser { @Override protected boolean isEligibleAttribute(String attributeName) { - return !(DIRECTORY_ATTRIBUTE.equals(attributeName) || MESSAGE_CREATOR_REFERENCE_ATTRIBUTE.equals(attributeName) || TYPE_ATTRIBUTE - .equals(attributeName)) - && super.isEligibleAttribute(attributeName); + return !(DIRECTORY_ATTRIBUTE.equals(attributeName)) && super.isEligibleAttribute(attributeName); } @Override protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) { - beanDefinition.addConstructorArgValue(element.getAttribute(DIRECTORY_ATTRIBUTE)); - String messageCreatorReference = element.getAttribute(MESSAGE_CREATOR_REFERENCE_ATTRIBUTE); - String type = element.getAttribute(TYPE_ATTRIBUTE); - if (StringUtils.hasText(type) && StringUtils.hasText(messageCreatorReference)) { - throw new ConfigurationException( - "Either the 'type' or the 'message-creator' attributes are allowed, but not both"); - } - if (StringUtils.hasText(messageCreatorReference)) { - beanDefinition.addConstructorArgReference(messageCreatorReference); - } - else { - if (!StringUtils.hasText(type) || FILE_SOURCE_TYPE_ATTRIBUTE.equals(type)) { - beanDefinition.addConstructorArgValue(new FileMessageCreator()); - } - else if (TEXT_SOURCE_TYPE_ATTRIBUTE.equals(type)) { - beanDefinition.addConstructorArgValue(new TextFileMessageCreator()); - } - else if (BINARY_SOURCE_TYPE_ATTRIBUTE.equals(type)) { - beanDefinition.addConstructorArgValue(new ByteArrayFileMessageCreator()); - } - } + beanDefinition.addConstructorArgValue(element.getAttribute(DIRECTORY_ATTRIBUTE)); + super.postProcess(beanDefinition, element); } } diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java index ea3c17a2f7..c77588c600 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java @@ -19,7 +19,6 @@ package org.springframework.integration.adapter.ftp; import java.io.File; import java.io.FileOutputStream; import java.io.IOException; -import java.util.HashMap; import java.util.Map; import org.apache.commons.logging.Log; @@ -27,16 +26,10 @@ import org.apache.commons.logging.LogFactory; import org.apache.commons.net.ftp.FTP; import org.apache.commons.net.ftp.FTPClient; import org.apache.commons.net.ftp.FTPFile; - -import org.springframework.beans.factory.InitializingBean; -import org.springframework.integration.adapter.file.ByteArrayFileMessageCreator; -import org.springframework.integration.adapter.file.FileNameGenerator; -import org.springframework.integration.adapter.file.TextFileMessageCreator; -import org.springframework.integration.message.Message; +import org.springframework.integration.adapter.file.AbstractDirectorySource; +import org.springframework.integration.adapter.file.FileInfo; import org.springframework.integration.message.MessageCreator; -import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessagingException; -import org.springframework.integration.message.Source; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -46,7 +39,7 @@ import org.springframework.util.StringUtils; * @author Marius Bogoevici * @author Mark Fisher */ -public class FtpSource implements Source, MessageDeliveryAware, InitializingBean { +public class FtpSource extends AbstractDirectorySource { private final static String DEFAULT_HOST = "localhost"; @@ -69,15 +62,13 @@ public class FtpSource implements Source, MessageDeliveryAware, Initiali private volatile File localWorkingDirectory; - private volatile boolean textBased = true; - - private volatile MessageCreator messageCreator; - - private final DirectoryContentManager directoryContentManager = new DirectoryContentManager(); - private final FTPClient client = new FTPClient(); + public FtpSource(MessageCreator messageCreator) { + super(messageCreator); + } + public void setHost(String host) { this.host = host; } @@ -104,67 +95,18 @@ public class FtpSource implements Source, MessageDeliveryAware, Initiali this.localWorkingDirectory = localWorkingDirectory; } - public boolean isTextBased() { - return this.textBased; - } + @Override + protected void populateSnapshot(Map snapshot) throws IOException { + FTPFile[] fileList = this.client.listFiles(); - public void setTextBased(boolean textBased) { - this.textBased = textBased; - } - - public void afterPropertiesSet() { - if (this.isTextBased()) { - this.messageCreator = new TextFileMessageCreator(); - } - else { - this.messageCreator = new ByteArrayFileMessageCreator(); + for (FTPFile ftpFile : fileList) { + FileInfo fileInfo = new FileInfo(ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(), ftpFile + .getSize()); + snapshot.put(ftpFile.getName(), fileInfo); } } - public final Message receive() { - try { - this.establishConnection(); - FTPFile[] fileList = this.client.listFiles(); - HashMap snapshot = new HashMap(); - for (FTPFile ftpFile : fileList) { - FileInfo fileInfo = new FileInfo( - ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(), ftpFile.getSize()); - snapshot.put(ftpFile.getName(), fileInfo); - } - this.directoryContentManager.processSnapshot(snapshot); - Map backlog = this.directoryContentManager.getBacklog(); - if (backlog.isEmpty()) { - return null; - } - String fileName = backlog.keySet().iterator().next(); - File file = new File(this.localWorkingDirectory, fileName); - if (file.exists()) { - file.delete(); - } - FileOutputStream fileOutputStream = new FileOutputStream(file); - this.client.retrieveFile(fileName, fileOutputStream); - fileOutputStream.close(); - return this.messageCreator.createMessage(file); - } - catch (Exception e) { - throw new MessagingException("Error while polling for messages.", e); - } - finally { - try { - if (this.client.isConnected()) { - this.client.disconnect(); - if (logger.isDebugEnabled()) { - logger.debug("connection closed"); - } - } - } - catch (IOException ioe) { - throw new MessagingException("Error when disconnecting from ftp.", ioe); - } - } - } - - private void establishConnection() throws IOException { + protected void establishConnection() throws IOException { if (!StringUtils.hasText(this.username)) { throw new MessagingException("username is required"); } @@ -178,27 +120,37 @@ public class FtpSource implements Source, MessageDeliveryAware, Initiali this.client.setFileType(FTP.IMAGE_FILE_TYPE); if (!this.remoteWorkingDirectory.equals(this.client.printWorkingDirectory()) && !this.client.changeWorkingDirectory(this.remoteWorkingDirectory)) { - throw new MessagingException("Could not change directory to '" + - remoteWorkingDirectory + "'. Please check the path."); + throw new MessagingException("Could not change directory to '" + remoteWorkingDirectory + + "'. Please check the path."); } if (logger.isDebugEnabled()) { logger.debug("working directory is: " + this.client.printWorkingDirectory()); } } - public void onSend(Message message) { - String filename = message.getHeader().getProperty(FileNameGenerator.FILENAME_PROPERTY_KEY); - if (StringUtils.hasText(filename)) { - this.directoryContentManager.fileProcessed(filename); - } - else if (this.logger.isWarnEnabled()) { - logger.warn("No filename in Message header, cannot send notification of processing."); + protected File retrieveNextFile() throws IOException { + String fileName = this.getDirectoryContentManager().getBacklog().keySet().iterator().next(); + File file = new File(this.localWorkingDirectory, fileName); + if (file.exists()) { + file.delete(); } + FileOutputStream fileOutputStream = new FileOutputStream(file); + this.client.retrieveFile(fileName, fileOutputStream); + fileOutputStream.close(); + return file; } - public void onFailure(MessagingException exception) { - if (this.logger.isWarnEnabled()) { - logger.warn("FtpSource received failure notification", exception); + protected void disconnect() { + try { + if (this.client.isConnected()) { + this.client.disconnect(); + if (logger.isDebugEnabled()) { + logger.debug("connection closed"); + } + } + } + catch (IOException ioe) { + throw new MessagingException("Error when disconnecting from ftp.", ioe); } } diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java index f02df80485..024552d7d6 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java @@ -16,21 +16,26 @@ package org.springframework.integration.adapter.ftp.config; -import org.w3c.dom.Element; - import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; +import org.springframework.integration.adapter.file.config.AbstractDirectorySourceParser; import org.springframework.integration.adapter.ftp.FtpSource; +import org.w3c.dom.Element; /** * Parser for the <ftp-source/> element. * * @author Mark Fisher + * @author Marius Bogoevici */ -public class FtpSourceParser extends AbstractSimpleBeanDefinitionParser { +public class FtpSourceParser extends AbstractDirectorySourceParser { + + public FtpSourceParser() { + super(true); + } @Override protected Class getBeanClass(Element element) { return FtpSource.class; } - + } diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/config/FileTargetParserTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/config/FileTargetParserTests.java index e4198cfbf5..eae2493ca0 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/config/FileTargetParserTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/config/FileTargetParserTests.java @@ -40,9 +40,11 @@ public class FileTargetParserTests { ApplicationContext context = new ClassPathXmlApplicationContext("fileTargetParserTests.xml", this.getClass()); FileTarget target = (FileTarget) context.getBean("target"); DirectFieldAccessor targetFieldAccessor = new DirectFieldAccessor(target); - SimpleFileMessageMapper messageMapper = (SimpleFileMessageMapper) targetFieldAccessor.getPropertyValue("messageMapper"); + SimpleFileMessageMapper messageMapper = (SimpleFileMessageMapper) targetFieldAccessor + .getPropertyValue("messageMapper"); DirectFieldAccessor mapperAccessor = new DirectFieldAccessor(messageMapper); - assertEquals(System.getProperty("java.io.tmpdir"), ((File) mapperAccessor.getPropertyValue("parentDirectory")).getAbsolutePath()); + assertEquals(System.getProperty("java.io.tmpdir"), ((File) mapperAccessor.getPropertyValue("parentDirectory")) + .getAbsolutePath()); assertTrue(mapperAccessor.getPropertyValue("fileNameGenerator") instanceof DefaultFileNameGenerator); } @@ -51,10 +53,12 @@ public class FileTargetParserTests { ApplicationContext context = new ClassPathXmlApplicationContext("fileTargetParserTests.xml", this.getClass()); FileTarget target = (FileTarget) context.getBean("targetWithCustomNameGenerator"); DirectFieldAccessor targetFieldAccessor = new DirectFieldAccessor(target); - SimpleFileMessageMapper messageMapper = (SimpleFileMessageMapper) targetFieldAccessor.getPropertyValue("messageMapper"); + SimpleFileMessageMapper messageMapper = (SimpleFileMessageMapper) targetFieldAccessor + .getPropertyValue("messageMapper"); DirectFieldAccessor mapperAccessor = new DirectFieldAccessor(messageMapper); - assertEquals(System.getProperty("java.io.tmpdir"), ((File) mapperAccessor.getPropertyValue("parentDirectory")).getAbsolutePath()); + assertEquals(System.getProperty("java.io.tmpdir"), ((File) mapperAccessor.getPropertyValue("parentDirectory")) + .getAbsolutePath()); assertTrue(mapperAccessor.getPropertyValue("fileNameGenerator") instanceof CustomNameGenerator); - } + } diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/config/invalidFileSourceTests.xml b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/config/invalidFileSourceTests.xml index 9236043e61..24f1dcab1e 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/config/invalidFileSourceTests.xml +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/config/invalidFileSourceTests.xml @@ -10,9 +10,11 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + - + diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/DirectoryContentManagerTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/DirectoryContentManagerTests.java index c3063ca24b..4f05ddd350 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/DirectoryContentManagerTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/DirectoryContentManagerTests.java @@ -22,6 +22,8 @@ import java.util.Map; import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import org.springframework.integration.adapter.file.DirectoryContentManager; +import org.springframework.integration.adapter.file.FileInfo; /** * @author Marius Bogoevici diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java index 907e2172a1..5218a1bbb9 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java @@ -17,25 +17,33 @@ package org.springframework.integration.adapter.ftp.config; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; import java.io.File; import org.junit.Test; - import org.springframework.beans.DirectFieldAccessor; +import org.springframework.beans.factory.BeanDefinitionStoreException; import org.springframework.context.ApplicationContext; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.ConfigurationException; +import org.springframework.integration.adapter.file.ByteArrayFileMessageCreator; +import org.springframework.integration.adapter.file.FileMessageCreator; +import org.springframework.integration.adapter.file.TextFileMessageCreator; +import org.springframework.integration.adapter.file.config.CustomMessageCreator; import org.springframework.integration.adapter.ftp.FtpSource; /** * @author Mark Fisher + * @author Marius Bogoevici */ public class FtpSourceParserTests { @Test public void testFtpSourceAdapterParser() { ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); - FtpSource ftpSource = (FtpSource) context.getBean("ftpSource"); + FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceDefault"); DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); assertEquals("testHost", accessor.getPropertyValue("host")); assertEquals(2121, accessor.getPropertyValue("port")); @@ -43,6 +51,90 @@ public class FtpSourceParserTests { assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); assertEquals("testUser", accessor.getPropertyValue("username")); assertEquals("testPassword", accessor.getPropertyValue("password")); + Object messageCreator = accessor.getPropertyValue("messageCreator"); + assertTrue(messageCreator instanceof FileMessageCreator); + DirectFieldAccessor messageCreatorAccessor = new DirectFieldAccessor(messageCreator); + assertEquals(messageCreatorAccessor.getPropertyValue("deleteFileAfterCreation"), false); + } + + @Test + public void testFtpSourceTextType() { + ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); + FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceText"); + DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); + assertEquals("testHost", accessor.getPropertyValue("host")); + assertEquals(2121, accessor.getPropertyValue("port")); + assertEquals(new File("/local"), accessor.getPropertyValue("localWorkingDirectory")); + assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); + assertEquals("testUser", accessor.getPropertyValue("username")); + assertEquals("testPassword", accessor.getPropertyValue("password")); + Object messageCreator = accessor.getPropertyValue("messageCreator"); + assertTrue(messageCreator instanceof TextFileMessageCreator); + DirectFieldAccessor messageCreatorAccessor = new DirectFieldAccessor(messageCreator); + assertEquals(messageCreatorAccessor.getPropertyValue("deleteFileAfterCreation"), true); + } + + @Test + public void testFtpSourceBinaryType() { + ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); + FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceBinary"); + DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); + assertEquals("testHost", accessor.getPropertyValue("host")); + assertEquals(2121, accessor.getPropertyValue("port")); + assertEquals(new File("/local"), accessor.getPropertyValue("localWorkingDirectory")); + assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); + assertEquals("testUser", accessor.getPropertyValue("username")); + assertEquals("testPassword", accessor.getPropertyValue("password")); + Object messageCreator = accessor.getPropertyValue("messageCreator"); + assertTrue(messageCreator instanceof ByteArrayFileMessageCreator); + DirectFieldAccessor messageCreatorAccessor = new DirectFieldAccessor(messageCreator); + assertEquals(messageCreatorAccessor.getPropertyValue("deleteFileAfterCreation"), true); + } + + @Test + public void testFtpSourceFileType() { + ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); + FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceFile"); + DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); + assertEquals("testHost", accessor.getPropertyValue("host")); + assertEquals(2121, accessor.getPropertyValue("port")); + assertEquals(new File("/local"), accessor.getPropertyValue("localWorkingDirectory")); + assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); + assertEquals("testUser", accessor.getPropertyValue("username")); + assertEquals("testPassword", accessor.getPropertyValue("password")); + Object messageCreator = accessor.getPropertyValue("messageCreator"); + assertTrue(messageCreator instanceof FileMessageCreator); + DirectFieldAccessor messageCreatorAccessor = new DirectFieldAccessor(messageCreator); + assertEquals(messageCreatorAccessor.getPropertyValue("deleteFileAfterCreation"), false); + } + + @Test + public void testFtpSourceCustomType() { + ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass()); + FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceCustom"); + DirectFieldAccessor accessor = new DirectFieldAccessor(ftpSource); + assertEquals("testHost", accessor.getPropertyValue("host")); + assertEquals(2121, accessor.getPropertyValue("port")); + assertEquals(new File("/local"), accessor.getPropertyValue("localWorkingDirectory")); + assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory")); + assertEquals("testUser", accessor.getPropertyValue("username")); + assertEquals("testPassword", accessor.getPropertyValue("password")); + Object messageCreator = accessor.getPropertyValue("messageCreator"); + assertTrue(messageCreator instanceof CustomMessageCreator); + // not testing for deleteFileAfterCreation - this is completely left up + // to the implementation + } + + @Test + public void testInvalidFtpSource() { + try { + ApplicationContext context = new ClassPathXmlApplicationContext("invalidFtpSourceTests.xml", this + .getClass()); + fail(); + } + catch (BeanDefinitionStoreException e) { + assertTrue(e.getCause() instanceof ConfigurationException); + } } } diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceParserTests.xml b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceParserTests.xml index 0259dbe45f..30a1aea722 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceParserTests.xml +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceParserTests.xml @@ -7,12 +7,51 @@ http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration-1.0.xsd"> - + + + + + + + + + + diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/invalidFtpSourceTests.xml b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/invalidFtpSourceTests.xml new file mode 100644 index 0000000000..137fc79983 --- /dev/null +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/invalidFtpSourceTests.xml @@ -0,0 +1,28 @@ + + + + + + + + + +