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

Refactored FtpSource and FileSource so that they inherit from the same base class AbstractDirectorySource. FtpSource namespace support is modified, so that it supports the type/message-creator pair, with the same semantic as FileSource. Implemented a distinct behaviour with respect to deleting files via MessageCreator: FtpSource will delete the local copies after creating the files, whereas FileSource will not.

Browse Source
This commit is contained in:
Marius Bogoevici
2008-05-24 15:34:54 +00:00
parent c2728c2f17
commit 3d00dd9930
19 changed files with 519 additions and 186 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd
View File

@@ -57,7 +57,8 @@
<xsd:attribute name="port" type="xsd:int" use="optional"/>
<xsd:attribute name="local-working-directory" type="xsd:string" use="required"/>
<xsd:attribute name="remote-working-directory" type="xsd:string" use="optional"/>
<xsd:attribute name="text-based" type="xsd:boolean" use="optional"/>
<xsd:attribute name="type" type="fileSourceType" use="optional"/>
<xsd:attribute name="message-creator" type="xsd:string" use="optional"/>
</xsd:complexType>
</xsd:element>

129
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractDirectorySource.java Normal file
View File

@@ -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<Object>, 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<File, ?> messageCreator;
public AbstractDirectorySource(MessageCreator<File, ?> messageCreator) {
Assert.notNull(messageCreator, "The MessageCreator must not be null");
this.messageCreator = messageCreator;
}
protected DirectoryContentManager getDirectoryContentManager() {
return this.directoryContentManager;
}
public MessageCreator<File, ?> getMessageCreator() {
return messageCreator;
}
public final Message receive() {
try {
this.establishConnection();
HashMap<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
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<String, FileInfo> 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();
}

27
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/AbstractFileMessageCreator.java
View File

@@ -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<T> implements MessageCreator<Fi
protected Log logger = LogFactory.getLog(this.getClass());
private final boolean deleteFileAfterCreation;
/**
* @param deleteFileAfterCreation Indicates whether the file should be
* deleted after the message has been created.
*/
public AbstractFileMessageCreator(boolean deleteFileAfterCreation) {
this.deleteFileAfterCreation = deleteFileAfterCreation;
}
public Message<T> createMessage(File file) {
public final Message<T> createMessage(File file) {
try {
T payload = this.readMessagePayload(file);
if (payload == null) {
return null;
}
Message<T> message = new GenericMessage<T>(payload);
message.getHeader().setProperty(FileNameGenerator.FILENAME_PROPERTY_KEY, file.getName());
if (deleteFileAfterCreation) {
file.delete();
}
return message;
}
catch (Exception e) {

10
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/ByteArrayFileMessageCreator.java
View File

@@ -29,6 +29,16 @@ import org.springframework.util.FileCopyUtils;
*/
public class ByteArrayFileMessageCreator extends AbstractFileMessageCreator<byte[]> {
/**
* 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);

2
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java → org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DirectoryContentManager.java
View File

@@ -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;

3
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FileInfo.java → org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileInfo.java
View File

@@ -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;

6
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileMessageCreator.java
View File

@@ -29,6 +29,12 @@ import org.springframework.integration.message.MessageCreator;
*/
public class FileMessageCreator extends AbstractFileMessageCreator<File> {
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;

61
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/FileSource.java
View File

@@ -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<Object>, InitializingBean, MessageDeliveryAware {
private final Log logger = LogFactory.getLog(this.getClass());
public class FileSource extends AbstractDirectorySource implements Source<Object>, InitializingBean, MessageDeliveryAware {
private final File directory;
@@ -53,18 +47,14 @@ public class FileSource implements Source<Object>, 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<File, ?> 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<Object>, 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<String, FileInfo> 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<Object>, InitializingBean, MessageDeli
if (files == null) {
throw new MessagingException("Problem occurred while polling for files. " +
"Is '" + directory.getAbsolutePath() + "' a directory?");
}
HashMap<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
}
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);
}
}

10
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/TextFileMessageCreator.java
View File

@@ -30,6 +30,16 @@ import org.springframework.util.FileCopyUtils;
*/
public class TextFileMessageCreator extends AbstractFileMessageCreator<String> {
/**
* 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));

84
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/config/AbstractDirectorySourceParser.java Normal file
View File

@@ -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));
}
}
}
}

48
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/config/FileSourceParser.java
View File

@@ -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);
}
}

122
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpSource.java
View File

@@ -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<Object>, MessageDeliveryAware, InitializingBean {
public class FtpSource extends AbstractDirectorySource {
private final static String DEFAULT_HOST = "localhost";
@@ -69,15 +62,13 @@ public class FtpSource implements Source<Object>, MessageDeliveryAware, Initiali
private volatile File localWorkingDirectory;
private volatile boolean textBased = true;
private volatile MessageCreator<File, ?> messageCreator;
private final DirectoryContentManager directoryContentManager = new DirectoryContentManager();
private final FTPClient client = new FTPClient();
public FtpSource(MessageCreator<File, ?> messageCreator) {
super(messageCreator);
}
public void setHost(String host) {
this.host = host;
}
@@ -104,67 +95,18 @@ public class FtpSource implements Source<Object>, MessageDeliveryAware, Initiali
this.localWorkingDirectory = localWorkingDirectory;
}
public boolean isTextBased() {
return this.textBased;
}
@Override
protected void populateSnapshot(Map<String, FileInfo> 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<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
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<String, FileInfo> 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<Object>, 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);
}
}

13
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java
View File

@@ -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 &lt;ftp-source/&gt; 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;
}
}

14
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/config/FileTargetParserTests.java
View File

@@ -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);
}
}

6
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/config/invalidFileSourceTests.xml
View File

@@ -10,9 +10,11 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:file-source id="fileSourceCustom" type="text" message-creator="customMessageCreator" directory="${java.io.tmpdir}"/>
<si:file-source id="fileSourceCustom" type="text"
message-creator="customMessageCreator" directory="${java.io.tmpdir}"/>
<bean id="customMessageCreator" class="org.springframework.integration.adapter.file.config.CustomMessageCreator"/>
<bean id="customMessageCreator"
class="org.springframework.integration.adapter.file.config.CustomMessageCreator"/>
<context:property-placeholder/>

2
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/DirectoryContentManagerTests.java
View File

@@ -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

96
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java
View File

@@ -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);
}
}
}

41
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceParserTests.xml
View File

@@ -7,12 +7,51 @@
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:ftp-source id="ftpSource"
<si:ftp-source id="ftpSourceDefault"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"/>
<si:ftp-source id="ftpSourceText"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"
type="text"/>
<si:ftp-source id="ftpSourceBinary"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"
type="binary"/>
<si:ftp-source id="ftpSourceFile"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"
type="file"/>
<si:ftp-source id="ftpSourceCustom"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"
message-creator="customMessageCreator"/>
<bean id="customMessageCreator"
class="org.springframework.integration.adapter.file.config.CustomMessageCreator"/>
</beans>

28
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/invalidFtpSourceTests.xml Normal file
View File

@@ -0,0 +1,28 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="http://www.springframework.org/schema/integration"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:ftp-source id="ftpSourceCustom"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"
message-creator="customMessageCreator"
type="text"/>
<bean id="customMessageCreator"
class="org.springframework.integration.adapter.file.config.CustomMessageCreator"/>
<context:property-placeholder/>
</beans>