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 index cd81771c86..cddf68fbd9 100644 --- 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 @@ -24,6 +24,7 @@ import java.util.Map; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageCreator; @@ -34,85 +35,74 @@ import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** - * Base class for implementing a Source that creates messages from files in a directory, - * either local or remote. + * Base class for implementing a Source that creates messages from files in a + * directory, either local or remote. * * @author Marius Bogoevici + * @author iwein */ -public abstract class AbstractDirectorySource implements MessageSource, MessageDeliveryAware { +public abstract class AbstractDirectorySource implements MessageSource, MessageDeliveryAware { public final static String FILE_INFO_PROPERTY = "file.info"; - private final Log logger = LogFactory.getLog(this.getClass()); private final DirectoryContentManager directoryContentManager = new DirectoryContentManager(); - private final MessageCreator messageCreator; + private final MessageCreator messageCreator; - - public AbstractDirectorySource(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() { + public MessageCreator getMessageCreator() { return messageCreator; } @SuppressWarnings("unchecked") - public final Message receive() { + 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 = MessageBuilder.fromMessage(message) - .setHeader(FileNameGenerator.FILENAME_PROPERTY_KEY, file.getName()) - .setHeader(FILE_INFO_PROPERTY, getDirectoryContentManager().getBacklog().get(file.getName())) - .build(); - return message; + return buildNextMessage(); } return null; } catch (Exception e) { throw new MessagingException("Error while polling for messages.", e); } - finally { - disconnect(); - } } - public void onSend(Message message) { - String filename = message.getHeaders().get(FileNameGenerator.FILENAME_PROPERTY_KEY, String.class); - 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."); - } + /** + * Hook point for implementors to create the next message that should be + * received. Implementations can use a File by File approach like + * FileSource). In cases where retrieval could be expensive because of + * network latency a batched approach could be implemented here. See + * FtpSource for an example. + * + * @return the next message containing (part of) the unprocessed content of + * the directory + * @throws IOException + */ + protected Message buildNextMessage() throws IOException { + return messageCreator.createMessage(retrieveNextPayload()); } + public abstract void onSend(Message message); + 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 @@ -125,11 +115,10 @@ public abstract class AbstractDirectorySource implements MessageSource, * @return * @throws IOException */ - protected abstract File retrieveNextFile() throws IOException; - - /** - * Disconnects from the directory - */ - protected abstract void disconnect(); + protected abstract T retrieveNextPayload() throws IOException; + protected final void fileProcessed(String fileName){ + this.directoryContentManager.fileProcessed(fileName); + } + } 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 e92058c71c..5f5e6ab504 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 @@ -22,13 +22,18 @@ import java.io.FilenameFilter; import java.io.IOException; import java.util.Map; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.springframework.core.io.Resource; import org.springframework.integration.ConfigurationException; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageCreator; import org.springframework.integration.message.MessageDeliveryAware; import org.springframework.integration.message.MessagingException; import org.springframework.integration.message.MessageSource; import org.springframework.util.Assert; +import org.springframework.util.StringUtils; /** * A messaging source that polls a directory to retrieve files. @@ -36,8 +41,10 @@ import org.springframework.util.Assert; * @author Mark Fisher * @author Marius Bogoevici */ -public class FileSource extends AbstractDirectorySource implements MessageSource, MessageDeliveryAware { +public class FileSource extends AbstractDirectorySource implements MessageSource, MessageDeliveryAware { + private final Log logger = LogFactory.getLog(this.getClass()); + private final File directory; private volatile FileFilter fileFilter; @@ -48,8 +55,18 @@ public class FileSource extends AbstractDirectorySource implements MessageSource public FileSource(Resource directory) { this(directory, new FileMessageCreator()); } + + @Override + protected Message buildNextMessage() throws IOException { + File file = retrieveNextPayload(); + Message message = this.getMessageCreator().createMessage(file); + message = MessageBuilder.fromMessage(message) + .setHeader(FileNameGenerator.FILENAME_PROPERTY_KEY, file.getName()).setHeader(FILE_INFO_PROPERTY, + getDirectoryContentManager().getBacklog().get(file.getName())).build(); + return message; + } - public FileSource(Resource directory, MessageCreator messageCreator) { + public FileSource(Resource directory, MessageCreator messageCreator) { super(messageCreator); Assert.notNull(directory, "The directory must not be null"); try { @@ -73,16 +90,6 @@ public class FileSource extends AbstractDirectorySource implements MessageSource this.filenameFilter = filenameFilter; } - @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; @@ -106,9 +113,19 @@ public class FileSource extends AbstractDirectorySource implements MessageSource } @Override - protected File retrieveNextFile() throws IOException { + protected File retrieveNextPayload() throws IOException { String fileName = this.getDirectoryContentManager().getBacklog().keySet().iterator().next(); return new File(directory, fileName); } + @Override + public void onSend(Message message) { + String filename = message.getHeaders().get(FileNameGenerator.FILENAME_PROPERTY_KEY, String.class); + if (StringUtils.hasText(filename)) { + fileProcessed(filename); + } + else if (this.logger.isWarnEnabled()) { + logger.warn("No filename in Message header, cannot send notification of processing."); + } + } } 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 c77588c600..b814dff3b9 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,15 +19,22 @@ package org.springframework.integration.adapter.ftp; import java.io.File; import java.io.FileOutputStream; import java.io.IOException; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; import java.util.Map; +import java.util.Set; import org.apache.commons.logging.Log; 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.DisposableBean; +import org.springframework.beans.factory.annotation.Required; import org.springframework.integration.adapter.file.AbstractDirectorySource; import org.springframework.integration.adapter.file.FileInfo; +import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageCreator; import org.springframework.integration.message.MessagingException; import org.springframework.util.Assert; @@ -38,8 +45,9 @@ import org.springframework.util.StringUtils; * * @author Marius Bogoevici * @author Mark Fisher + * @author Iwein Fuld */ -public class FtpSource extends AbstractDirectorySource { +public class FtpSource extends AbstractDirectorySource> implements DisposableBean { private final static String DEFAULT_HOST = "localhost"; @@ -47,7 +55,6 @@ public class FtpSource extends AbstractDirectorySource { private final static String DEFAULT_REMOTE_WORKING_DIRECTORY = "/"; - private final Log logger = LogFactory.getLog(this.getClass()); private volatile String username; @@ -62,11 +69,15 @@ public class FtpSource extends AbstractDirectorySource { private volatile File localWorkingDirectory; - private final FTPClient client = new FTPClient(); + private final FTPClient client; + public FtpSource(MessageCreator, List> messageCreator) { + this(messageCreator, new FTPClient()); + } - public FtpSource(MessageCreator messageCreator) { + public FtpSource(MessageCreator, List> messageCreator, FTPClient client) { super(messageCreator); + this.client = client; } public void setHost(String host) { @@ -77,6 +88,7 @@ public class FtpSource extends AbstractDirectorySource { this.port = port; } + @Required public void setUsername(String username) { this.username = username; } @@ -86,8 +98,9 @@ public class FtpSource extends AbstractDirectorySource { } public void setRemoteWorkingDirectory(String remoteWorkingDirectory) { - Assert.hasText(remoteWorkingDirectory, "'remoteWorkingDirectory' is required"); - this.remoteWorkingDirectory = remoteWorkingDirectory; + Assert.notNull(remoteWorkingDirectory, "'remoteWorkingDirectory' cannot be null"); + // FtpClient is picky about "", so we make it happy + this.remoteWorkingDirectory = remoteWorkingDirectory.replaceAll("^$", "/"); } public void setLocalWorkingDirectory(File localWorkingDirectory) { @@ -97,6 +110,7 @@ public class FtpSource extends AbstractDirectorySource { @Override protected void populateSnapshot(Map snapshot) throws IOException { + establishConnection(); FTPFile[] fileList = this.client.listFiles(); for (FTPFile ftpFile : fileList) { @@ -107,6 +121,12 @@ public class FtpSource extends AbstractDirectorySource { } protected void establishConnection() throws IOException { + if (this.client.isConnected()) { + if (logger.isDebugEnabled()) { + logger.debug("client already connected"); + } + return; + } if (!StringUtils.hasText(this.username)) { throw new MessagingException("username is required"); } @@ -128,16 +148,22 @@ public class FtpSource extends AbstractDirectorySource { } } - 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(); + protected List retrieveNextPayload() throws IOException { + establishConnection(); + List files = new ArrayList(); + Set backlog = this.getDirectoryContentManager().getBacklog().keySet(); + for (String fileName : backlog) { + File file = new File(this.localWorkingDirectory, fileName); + if (file.exists()) { + file.delete(); + } + FileOutputStream fileOutputStream = new FileOutputStream(file); + this.client.retrieveFile(fileName, fileOutputStream); + fileOutputStream.close(); + files.add(file); } - FileOutputStream fileOutputStream = new FileOutputStream(file); - this.client.retrieveFile(fileName, fileOutputStream); - fileOutputStream.close(); - return file; + disconnect(); + return files; } protected void disconnect() { @@ -150,7 +176,21 @@ public class FtpSource extends AbstractDirectorySource { } } catch (IOException ioe) { - throw new MessagingException("Error when disconnecting from ftp.", ioe); + if (logger.isErrorEnabled()) { + logger.error("Error when disconnecting from ftp.", ioe); + } + } + } + + public void destroy() throws Exception { + disconnect(); + } + + @Override + public void onSend(Message message) { + List files = ((Message>) message).getPayload(); + for (File file : files) { + fileProcessed(file.getName()); } } diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java new file mode 100644 index 0000000000..1b81f05772 --- /dev/null +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpSourceTests.java @@ -0,0 +1,120 @@ +package org.springframework.integration.adapter.ftp; + +import static org.easymock.classextension.EasyMock.*; +import static org.junit.Assert.*; + +import java.io.File; +import java.io.OutputStream; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Calendar; +import java.util.Collections; +import java.util.List; + +import org.apache.commons.net.ftp.FTPClient; +import org.apache.commons.net.ftp.FTPFile; +import org.junit.Before; +import org.junit.Test; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageCreator; + +@SuppressWarnings("unchecked") +public class FtpSourceTests { + + private MessageCreator, List> messageCreator = createMock(MessageCreator.class); + + private FTPClient ftpClient = createMock(FTPClient.class); + + private FTPFile ftpFile = createMock(FTPFile.class); + + private Object[] globalMocks = new Object[] { messageCreator, ftpClient, ftpFile }; + + private static final String HOST = "testHost"; + + private static final String USER = "testUser"; + + private static final String PASS = "testPass"; + + private FtpSource ftpSource; + + @Before + public void initializeFtpSource() { + ftpSource = new FtpSource(messageCreator, ftpClient); + ftpSource.setHost(HOST); + ftpSource.setUsername(USER); + ftpSource.setPassword(PASS); + } + + @Before + public void clearState() { + reset(globalMocks); + } + + @Test + public void retrieveSingleFile() throws Exception { + // connect client and get file + expect(ftpClient.isConnected()).andReturn(false); + expect(ftpClient.isConnected()).andReturn(true).anyTimes(); + + ftpClient.connect(HOST, 21); + expect(ftpClient.login(USER, PASS)).andReturn(true); + expect(ftpClient.setFileType(anyInt())).andReturn(true); + expect(ftpClient.printWorkingDirectory()).andReturn("/"); + expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed(Calendar.getInstance(), "test")); + expect(ftpClient.retrieveFile(eq("test"), isA(OutputStream.class))).andReturn(true); + // create message + expect(messageCreator.createMessage(isA(List.class))).andReturn( + new GenericMessage(Arrays.asList(new File("test")))); + ftpClient.disconnect(); + replay(globalMocks); + ftpSource.onSend(ftpSource.receive()); + verify(globalMocks); + } + + private FTPFile[] mockedFTPFilesNamed(Calendar timestamp, String... names) { + List files = new ArrayList(); + for (String name : names) { + FTPFile ftpFile = createMock(FTPFile.class); + expect(ftpFile.getName()).andReturn(name).anyTimes(); + expect(ftpFile.getTimestamp()).andReturn(timestamp).anyTimes(); + expect(ftpFile.getSize()).andReturn(100l).anyTimes(); + files.add(ftpFile); + replay(ftpFile); + } + return files.toArray(new FTPFile[] {}); + } + + @Test + public void retrieveMultipleFiles() throws Exception { + // connect client + expect(ftpClient.isConnected()).andReturn(false); + expect(ftpClient.isConnected()).andReturn(true).anyTimes(); + + ftpClient.connect(HOST, 21); + expect(ftpClient.login(USER, PASS)).andReturn(true); + expect(ftpClient.setFileType(anyInt())).andReturn(true); + expect(ftpClient.printWorkingDirectory()).andReturn("/"); + + // get files + Calendar timestamp = Calendar.getInstance(); + expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed(timestamp, "test", "test2")).anyTimes(); + + expect(ftpClient.retrieveFile(eq("test"), isA(OutputStream.class))).andReturn(true); + expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true); + // create message + List files = Arrays.asList(new File("test"), new File("test2")); + expect(messageCreator.createMessage(isA(List.class))).andReturn(new GenericMessage(files)); + ftpClient.disconnect(); + + replay(globalMocks); + Message receivedFiles = ftpSource.receive(); + ftpSource.onSend(receivedFiles); + Message> secondReceived = ftpSource.receive(); + verify(globalMocks); + assertEquals(files, receivedFiles.getPayload()); + assertNull(secondReceived); + } + + +}