FtpSource no longer uses a MessageCreator. Instead, it uses MessageBuilder internally, and any Message customization can be applied with Transformers.

org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/AbstractDirectorySource.java

@@ -25,11 +25,10 @@ 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.MessageBuilder;
 import org.springframework.integration.message.MessageDeliveryAware;
 import org.springframework.integration.message.MessagingException;
 import org.springframework.integration.message.PollableSource;
-import org.springframework.util.Assert;
 
 /**
  * Base class for implementing a PollableSource that creates messages from files
@@ -46,17 +45,13 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 
 	private final Backlog<FileSnapshot> backlog;
 
-	private final MessageCreator<T, T> messageCreator;
 
-
-	public AbstractDirectorySource(MessageCreator<T, T> messageCreator) {
-		this(messageCreator, null);
+	public AbstractDirectorySource() {
+		this(null);
 	}
 
-	public AbstractDirectorySource(MessageCreator<T, T> messageCreator, Comparator<FileSnapshot> comparator) {
+	public AbstractDirectorySource(Comparator<FileSnapshot> comparator) {
 		this.backlog = comparator == null ? new Backlog<FileSnapshot>() : new Backlog<FileSnapshot>(comparator);
-		Assert.notNull(messageCreator, "The MessageCreator must not be null");
-		this.messageCreator = messageCreator;
 	}
 
 
@@ -64,10 +59,6 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 		return this.backlog;
 	}
 
-	public MessageCreator<T, T> getMessageCreator() {
-		return this.messageCreator;
-	}
-
 	public final Message<T> receive() {
 		try {
 			refreshSnapshotAndMarkProcessing(this.backlog);
@@ -89,17 +80,16 @@ public abstract class AbstractDirectorySource<T> implements PollableSource<T>, M
 
 	/**
 	 * 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.
+	 * received. Implementations can use a File by File approach or 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 this.messageCreator.createMessage(retrieveNextPayload());
+		return MessageBuilder.withPayload(retrieveNextPayload()).build();
 	}
 
 	public void onSend(Message<T> message) {

org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSource.java

@@ -25,8 +25,6 @@ import java.util.List;
 import org.apache.commons.net.ftp.FTPClient;
 import org.apache.commons.net.ftp.FTPFile;
 
-import org.springframework.integration.message.DefaultMessageCreator;
-import org.springframework.integration.message.MessageCreator;
 import org.springframework.util.Assert;
 
 /**
@@ -46,11 +44,6 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
 
 
 	public FtpSource(FTPClientPool clientPool) {
-		this(new DefaultMessageCreator<List<File>>(), clientPool);
-	}
-
-	public FtpSource(MessageCreator<List<File>, List<File>> messageCreator, FTPClientPool clientPool) {
-		super(messageCreator);
 		this.clientPool = clientPool;
 	}
 
@@ -84,8 +77,8 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
 				 * if files couldn't be parsed
 				 */
 				if (ftpFile != null) {
-					FileSnapshot fileSnapshot = new FileSnapshot(ftpFile.getName(), ftpFile.getTimestamp()
-							.getTimeInMillis(), ftpFile.getSize());
+					FileSnapshot fileSnapshot = new FileSnapshot(ftpFile.getName(),
+							ftpFile.getTimestamp().getTimeInMillis(), ftpFile.getSize());
 					snapshot.add(fileSnapshot);
 				}
 			}
@@ -101,8 +94,7 @@ public class FtpSource extends AbstractDirectorySource<List<File>> {
 			List<File> files = new ArrayList<File>();
 			List<FileSnapshot> toDo = this.getBacklog().getProcessingBuffer();
 			for (FileSnapshot fileSnapshot : toDo) {
-				// some awkwardness here because the local path may be different
-				// from the remote path
+				// local path may be different from the remote path
 				File file = new File(this.localWorkingDirectory, fileSnapshot.getFileName());
 				if (file.exists()) {
 					file.delete();

org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java

@@ -47,6 +47,7 @@ public class QueuedFTPClientPool implements FTPClientPool {
 
 	private static final String DEFAULT_REMOTE_WORKING_DIRECTORY = "/";
 
+
 	private final Queue<FTPClient> pool;
 
 	private volatile FTPClientConfig config;
@@ -59,13 +60,25 @@ public class QueuedFTPClientPool implements FTPClientPool {
 
 	private volatile String password;
 
-	private volatile FTPClientFactory factory = new DefaultFactory();
+	private volatile FTPClientFactory factory = new DefaultFTPClientFactory();
 
 	private final Log log = LogFactory.getLog(this.getClass());
 
 	private volatile String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY;
 
-	// setters
+
+	public QueuedFTPClientPool() {
+		this(DEFAULT_POOL_SIZE);
+	}
+
+	/**
+	 * @param maxPoolSize the maximum size of the pool
+	 */
+	public QueuedFTPClientPool(int maxPoolSize) {
+		pool = new ArrayBlockingQueue<FTPClient>(maxPoolSize);
+	}
+
+
 	public void setConfig(FTPClientConfig config) {
 		Assert.notNull(config);
 		this.config = config;
@@ -101,22 +114,11 @@ public class QueuedFTPClientPool implements FTPClientPool {
 		this.factory = factory;
 	}
 
-	public QueuedFTPClientPool() {
-		this(DEFAULT_POOL_SIZE);
-	}
-
-	/**
-	 * @param maxPoolSize the maximum size of the pool
-	 */
-	public QueuedFTPClientPool(int maxPoolSize) {
-		pool = new ArrayBlockingQueue<FTPClient>(maxPoolSize);
-	}
-
 	/**
 	 * Returns an active FTPClient connected to the configured server. When no
 	 * clients are available in the queue a new client is created with the
 	 * factory.
-	 * 
+	 * <p>
 	 * It is possible that released clients are disconnected by the remote
 	 * server (@see {@link FTPClient#sendNoOp()}. In this case getClient is
 	 * called recursively to obtain a client that is still alive. For this
@@ -157,7 +159,7 @@ public class QueuedFTPClientPool implements FTPClientPool {
 		}
 	}
 
-	private class DefaultFactory implements FTPClientFactory {
+	private class DefaultFTPClientFactory implements FTPClientFactory {
 
 		public FTPClient getClient() throws SocketException, IOException {
 			FTPClient client = new FTPClient();
@@ -192,4 +194,5 @@ public class QueuedFTPClientPool implements FTPClientPool {
 			return client;
 		}
 	}
+
 }

org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java

@@ -25,7 +25,6 @@ import org.springframework.integration.config.AbstractPollingInboundChannelAdapt
 import org.springframework.integration.config.IntegrationNamespaceUtils;
 import org.springframework.integration.ftp.FtpSource;
 import org.springframework.integration.ftp.QueuedFTPClientPool;
-import org.springframework.util.StringUtils;
 
 /**
  * Parser for the <inbound-channel-adapter/> element of the 'ftp' namespace.
@@ -39,11 +38,6 @@ public class FtpInboundChannelAdapterParser extends AbstractPollingInboundChanne
 	@Override
 	protected String parseSource(Element element, ParserContext parserContext) {
 		BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(FtpSource.class);
-		String messageCreatorReference = element.getAttribute("message-creator");
-		if (StringUtils.hasText(messageCreatorReference)) {
-			builder.addConstructorArgReference(messageCreatorReference);
-		}
-		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "local-working-directory");
 		String username = element.getAttribute("username");
 		String password = element.getAttribute("password");
 		String host = element.getAttribute("host");
@@ -56,6 +50,7 @@ public class FtpInboundChannelAdapterParser extends AbstractPollingInboundChanne
 		queuedFTPClientPool.setPort(Integer.parseInt(port));
 		queuedFTPClientPool.setRemoteWorkingDirectory(remoteWorkingDirectory);
 		builder.addConstructorArgValue(queuedFTPClientPool);
+		IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "local-working-directory");
 		return BeanDefinitionReaderUtils.registerWithGeneratedName(
 				builder.getBeanDefinition(), parserContext.getRegistry());
 	}

org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/spring-integration-ftp-1.0.xsd

@@ -24,13 +24,13 @@
 				</xsd:documentation>
 			</xsd:annotation>
 			<xsd:attribute name="id" type="xsd:string"/>
+			<xsd:attribute name="channel" type="xsd:string"/>
 			<xsd:attribute name="username" type="xsd:string" use="optional"/>
 			<xsd:attribute name="password" type="xsd:string" use="optional"/>
 			<xsd:attribute name="host" type="xsd:string" use="required"/>
 			<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="message-creator" type="xsd:string" use="optional"/>
 		</xsd:complexType>
 	</xsd:element>