FtpSource refactoring: Message<File> -> Message<List<File>>

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<Object>, MessageDeliveryAware {
+public abstract class AbstractDirectorySource<T> implements MessageSource<T>, 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<File, ?> messageCreator;
+	private final MessageCreator<T, T> messageCreator;
 
-
-	public AbstractDirectorySource(MessageCreator<File, ?> messageCreator) {
+	public AbstractDirectorySource(MessageCreator<T, T> messageCreator) {
 		Assert.notNull(messageCreator, "The MessageCreator must not be null");
 		this.messageCreator = messageCreator;
 	}
 
-
 	protected DirectoryContentManager getDirectoryContentManager() {
 		return this.directoryContentManager;
 	}
 
-	public MessageCreator<File, ?> getMessageCreator() {
+	public MessageCreator<T, T> getMessageCreator() {
 		return messageCreator;
 	}
 
 	@SuppressWarnings("unchecked")
-	public final Message receive() {
+	public final Message<T> 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 = 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<T> 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<Object>,
 	 * @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);
+	}
+	
 }

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<Object>, MessageDeliveryAware {
+public class FileSource extends AbstractDirectorySource<File> implements MessageSource<File>, 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<File> buildNextMessage() throws IOException {
+		File file = retrieveNextPayload();
+		Message<File> 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<File, ?> messageCreator) {
+	public FileSource(Resource directory, MessageCreator<File, File> 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<String, FileInfo> 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.");
+		}
+	}
 }

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<List<File>> 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<File>, List<File>> messageCreator) {
+		this(messageCreator, new FTPClient());
+	}
 
-	public FtpSource(MessageCreator<File, ?> messageCreator) {
+	public FtpSource(MessageCreator<List<File>, List<File>> 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<String, FileInfo> 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<File> retrieveNextPayload() throws IOException {
+		establishConnection();
+		List<File> files = new ArrayList<File>();
+		Set<String> 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<File> files = ((Message<List<File>>) message).getPayload();
+		for (File file : files) {
+			fileProcessed(file.getName());
 		}
 	}
 

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<File>, List<File>> 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<FTPFile> files = new ArrayList<FTPFile>();
+		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<File> 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<List<File>> secondReceived = ftpSource.receive();
+		verify(globalMocks);
+		assertEquals(files, receivedFiles.getPayload());
+		assertNull(secondReceived);
+	}
+
+
+}