getProcessingBuffer() {
- return Collections.unmodifiableList(this.processingBuffer.get());
- }
-
- /**
- * Asks the backlog if there are any more items to process. This means that
- * this method is intended to return different results in different threads
- * when at least one of the thread is processing. It is unlikely that it is
- * useful to call this method during processing.
- * @return true if both the thread local processing buffer and
- * the backlog are empty.
- */
- public boolean isEmpty() {
- return this.backlog.isEmpty() && this.processingBuffer.get().isEmpty();
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FTPClientFactory.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FTPClientFactory.java
deleted file mode 100644
index 430caa88a7..0000000000
--- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FTPClientFactory.java
+++ /dev/null
@@ -1,38 +0,0 @@
-/*
- * 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.ftp;
-
-import java.io.IOException;
-import java.net.SocketException;
-
-import org.apache.commons.net.ftp.FTPClient;
-
-/**
- * Factory for {@link FTPClient}.
- *
- * @author Iwein Fuld
- */
-public interface FTPClientFactory {
-
- /**
- * @return Fully configured and connected FTPClient.
- * @throws SocketException
- * @throws IOException
- */
- FTPClient getClient() throws SocketException, IOException;
-
-}
diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FTPClientPool.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FTPClientPool.java
deleted file mode 100644
index fc351afd7b..0000000000
--- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FTPClientPool.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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.ftp;
-
-import org.apache.commons.net.ftp.FTPClient;
-
-/**
- * A pool of {@link FTPClient} instances. The pool can be used to control the
- * number of open FTP connections and reuse these connections efficiently.
- *
- * @author Iwein Fuld
- */
-public interface FTPClientPool extends FTPClientFactory {
-
- /**
- * Releases the client back to the pool. When calling this method the caller
- * is no longer responsible for the connection. The pool is free to do with
- * it as it sees fit, which means either recycling or disconnecting it most
- * probably.
- *
- * The caller should NOT disconnect the client before calling this method.
- *
- * The caller is NOT expected to use the client after calling this method.
- * Doing so can lead to unexpected behavior.
- *
- * @param client
- */
- void releaseClient(FTPClient client);
-
-}
diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FileSnapshot.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FileSnapshot.java
deleted file mode 100644
index fb227ce229..0000000000
--- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FileSnapshot.java
+++ /dev/null
@@ -1,101 +0,0 @@
-/*
- * 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.ftp;
-
-import java.io.File;
-
-import org.springframework.util.Assert;
-
-/**
- * Information about a file.
- *
- * The FileSnapshot takes a snapshot of certain mutable properties of a file and
- * stores them in an immutable way. This can be useful to determine if files
- * have been changed.
- *
- * @author Marius Bogoevici
- * @author Iwein Fuld
- */
-public class FileSnapshot implements Comparable {
-
- private final File file;
-
- private final long modificationTimestamp;
-
- private final long size;
-
- public FileSnapshot(File file) {
- Assert.notNull(file, "Can't take a snapshot of file that is null");
- this.file = file;
- this.modificationTimestamp = file.lastModified();
- this.size = file.length();
- }
-
- public FileSnapshot(String fileName, long modificationTimestamp, long size) {
- this.modificationTimestamp = modificationTimestamp;
- this.size = size;
- this.file = new File(fileName);
- }
-
- public String getFileName() {
- // this could be cached for better performance
- return file.getName();
- }
-
- public long getModificationTimestamp() {
- return modificationTimestamp;
- }
-
- public long getSize() {
- return size;
- }
-
- /**
- *
- * Be careful to note that the file that this snapshot refers to might have
- * changed. In particular:
- * snapshot.getModificationTimestamp() != snapshot.getFile().lastModified()
- * will evalutate to true in
- * many scenarios.
- *
- * @return the file that the snapshot was based on.
- */
- public File getFile() {
- return file;
- }
-
- @Override
- public boolean equals(Object other) {
- if (other == null || !(other instanceof FileSnapshot)) {
- return false;
- }
- FileSnapshot otherInfo = (FileSnapshot) other;
- return this.getSize() == otherInfo.getSize()
- && this.getModificationTimestamp() == otherInfo.getModificationTimestamp()
- && this.file.getName().equals(otherInfo.getFileName());
- }
-
- @Override
- public int hashCode() {
- return file.getPath().hashCode() ^ new Long(modificationTimestamp).hashCode() ^ new Long(size).hashCode();
- }
-
- public int compareTo(FileSnapshot other) {
- return this.getFile().compareTo(other.getFile());
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSendingMessageHandler.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSendingMessageHandler.java
deleted file mode 100644
index d00ec04422..0000000000
--- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSendingMessageHandler.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.ftp;
-
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-
-import org.apache.commons.net.ftp.FTPClient;
-
-import org.springframework.integration.core.Message;
-import org.springframework.integration.message.MessageHandler;
-import org.springframework.integration.message.MessageDeliveryException;
-import org.springframework.util.Assert;
-
-/**
- * A {@link MessageHandler} implementation that sends files to an FTP server.
- *
- * @author Iwein Fuld
- * @author Mark Fisher
- */
-public class FtpSendingMessageHandler implements MessageHandler {
-
- private final FTPClientPool ftpClientPool;
-
-
- public FtpSendingMessageHandler(FTPClientPool ftpClientPool) {
- Assert.notNull(ftpClientPool, "ftpClientPool must not be null");
- this.ftpClientPool = ftpClientPool;
- }
-
-
- public void handleMessage(Message message) {
- Assert.notNull(message, "message must not be null");
- Object payload = message.getPayload();
- Assert.notNull(payload, "message payload must not be null");
- Assert.isInstanceOf(File.class, payload, "Message payload must be an instance of [java.io.File]");
- File file = (File) payload;
- if (file != null && file.exists()) {
- FTPClient client = null;
- try {
- FileInputStream fileInputStream = new FileInputStream(file);
- client = this.ftpClientPool.getClient();
- boolean sent = client.storeFile(file.getName(), fileInputStream);
- fileInputStream.close();
- if (!sent) {
- throw new MessageDeliveryException(message, "Failed to store file '" + file + "'");
- }
- }
- catch (FileNotFoundException e) {
- throw new MessageDeliveryException(message, "File [" + file + "] lost from local working directory", e);
- }
- catch (IOException e) {
- throw new MessageDeliveryException(message, "Error transferring File [" + file
- + "] from local working directory to remote FTP directory", e);
- }
- finally {
- ftpClientPool.releaseClient(client);
- }
- }
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSource.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSource.java
deleted file mode 100644
index 39f97985bd..0000000000
--- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSource.java
+++ /dev/null
@@ -1,114 +0,0 @@
-/*
- * 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.ftp;
-
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.commons.net.ftp.FTPClient;
-import org.apache.commons.net.ftp.FTPFile;
-
-import org.springframework.util.Assert;
-
-/**
- * A source adapter for receiving files via FTP.
- *
- * @author Marius Bogoevici
- * @author Mark Fisher
- * @author Iwein Fuld
- */
-public class FtpSource extends AbstractDirectorySource> {
-
- private volatile File localWorkingDirectory;
-
- private volatile int maxFilesPerMessage = -1;
-
- private final FTPClientPool clientPool;
-
-
- public FtpSource(FTPClientPool clientPool) {
- this.clientPool = clientPool;
- }
-
-
- public void setMaxFilesPerMessage(int maxFilesPerMessage) {
- Assert.isTrue(maxFilesPerMessage > 0, "'maxFilesPerMessage' must be greater than 0");
- this.maxFilesPerMessage = maxFilesPerMessage;
- }
-
- public void setLocalWorkingDirectory(File localWorkingDirectory) {
- Assert.notNull(localWorkingDirectory, "'localWorkingDirectory' must not be null");
- this.localWorkingDirectory = localWorkingDirectory;
- }
-
- @Override
- protected void refreshSnapshotAndMarkProcessing(Backlog directoryContentManager) throws IOException {
- List snapshot = new ArrayList();
- populateSnapshot(snapshot);
- directoryContentManager.processSnapshot(snapshot);
- directoryContentManager.prepareForProcessing(maxFilesPerMessage);
- }
-
- @Override
- protected void populateSnapshot(List snapshot) throws IOException {
- FTPClient client = this.clientPool.getClient();
- FTPFile[] fileList = client.listFiles();
- try {
- for (FTPFile ftpFile : fileList) {
- /*
- * according to the FTPFile javadoc the list can contain nulls
- * if files couldn't be parsed
- */
- if (ftpFile != null) {
- FileSnapshot fileSnapshot = new FileSnapshot(ftpFile.getName(),
- ftpFile.getTimestamp().getTimeInMillis(), ftpFile.getSize());
- snapshot.add(fileSnapshot);
- }
- }
- }
- finally {
- this.clientPool.releaseClient(client);
- }
- }
-
- protected List retrieveNextPayload() throws IOException {
- FTPClient client = this.clientPool.getClient();
- try {
- List files = new ArrayList();
- List toDo = this.getBacklog().getProcessingBuffer();
- for (FileSnapshot fileSnapshot : toDo) {
- // local path may be different from the remote path
- File file = new File(this.localWorkingDirectory, fileSnapshot.getFileName());
- if (file.exists()) {
- file.delete();
- }
- FileOutputStream fileOutputStream = new FileOutputStream(file);
- client.retrieveFile(fileSnapshot.getFileName(), fileOutputStream);
- fileOutputStream.close();
- files.add(file);
- }
- return files;
- }
- finally {
- this.clientPool.releaseClient(client);
- }
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java
deleted file mode 100644
index e7c9944099..0000000000
--- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java
+++ /dev/null
@@ -1,198 +0,0 @@
-/*
- * 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.ftp;
-
-import java.io.IOException;
-import java.net.SocketException;
-import java.util.Queue;
-import java.util.concurrent.ArrayBlockingQueue;
-
-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.FTPClientConfig;
-import org.apache.commons.net.ftp.FTPReply;
-
-import org.springframework.integration.core.MessagingException;
-import org.springframework.util.Assert;
-import org.springframework.util.StringUtils;
-
-/**
- * FTPClientPool implementation based on a Queue. This implementation has a
- * default pool size of 5, but this is configurable with a constructor argument.
- *
- * This implementation pools released clients, but gives no guarantee to the
- * number of clients open at the same time.
- *
- * @author Iwein Fuld
- */
-public class QueuedFTPClientPool implements FTPClientPool {
-
- private static final int DEFAULT_POOL_SIZE = 5;
-
- private static final String DEFAULT_REMOTE_WORKING_DIRECTORY = "/";
-
-
- private final Queue pool;
-
- private volatile FTPClientConfig config;
-
- private volatile String host;
-
- private volatile int port = FTP.DEFAULT_PORT;
-
- private volatile String username;
-
- private volatile String password;
-
- private volatile FTPClientFactory factory = new DefaultFTPClientFactory();
-
- private final Log log = LogFactory.getLog(this.getClass());
-
- private volatile String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY;
-
-
- public QueuedFTPClientPool() {
- this(DEFAULT_POOL_SIZE);
- }
-
- /**
- * @param maxPoolSize the maximum size of the pool
- */
- public QueuedFTPClientPool(int maxPoolSize) {
- pool = new ArrayBlockingQueue(maxPoolSize);
- }
-
-
- public void setConfig(FTPClientConfig config) {
- Assert.notNull(config);
- this.config = config;
- }
-
- public void setHost(String host) {
- Assert.hasText(host);
- this.host = host;
- }
-
- public void setPort(int port) {
- Assert.isTrue(port > 0, "Port number should be > 0");
- this.port = port;
- }
-
- public void setUsername(String user) {
- Assert.hasText(user, "'user' should be a nonempty string");
- this.username = user;
- }
-
- public void setPassword(String pass) {
- Assert.notNull(pass, "password should not be null");
- this.password = pass;
- }
-
- public void setRemoteWorkingDirectory(String remoteWorkingDirectory) {
- Assert.notNull(remoteWorkingDirectory, "remote directory should not be null");
- this.remoteWorkingDirectory = remoteWorkingDirectory.replaceAll("^$", "/");
- }
-
- public void setFactory(FTPClientFactory factory) {
- Assert.notNull(factory);
- this.factory = factory;
- }
-
- /**
- * 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.
- *
- * 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
- * reason large pools are not recommended in poor networking conditions.
- */
- public FTPClient getClient() throws SocketException, IOException {
- FTPClient client = pool.poll();
- if (client == null) {
- client = factory.getClient();
- }
- else {
- client = isClientAlive(client) ? client : getClient();
- }
- return client;
- }
-
- private boolean isClientAlive(FTPClient client) {
- try {
- if (client.sendNoOp()) {
- return true;
- }
- }
- catch (IOException e) {
- log.warn("Client [" + client + "] discarded: ", e);
- }
- return false;
- }
-
- public void releaseClient(FTPClient client) {
- Assert.notNull(client, "'client' cannot be null");
- if (!pool.offer(client)) {
- try {
- client.disconnect();
- }
- catch (IOException e) {
- log.warn("Error disconnecting ftpclient", e);
- }
- }
- }
-
- private class DefaultFTPClientFactory implements FTPClientFactory {
-
- public FTPClient getClient() throws SocketException, IOException {
- FTPClient client = new FTPClient();
- client.configure(config);
- if (!StringUtils.hasText(username)) {
- throw new MessagingException("username is required");
- }
- client.connect(host, port);
- if (!FTPReply.isPositiveCompletion(client.getReplyCode())) {
- throw new MessagingException("Connecting to server [" + host + ":" + port
- + "] failed, please check the connection");
- }
- if (log.isDebugEnabled()) {
- log.debug("Connected to server [" + host + ":" + port + "]");
- }
- if (!client.login(username, password)) {
- throw new MessagingException("Login failed. Please check the username and password.");
- }
- if (log.isDebugEnabled()) {
- log.debug("login successful");
- }
- client.setFileType(FTP.BINARY_FILE_TYPE);
-
- if (!remoteWorkingDirectory.equals(client.printWorkingDirectory())
- && !client.changeWorkingDirectory(remoteWorkingDirectory)) {
- throw new MessagingException("Could not change directory to '" + remoteWorkingDirectory
- + "'. Please check the path.");
- }
- if (log.isDebugEnabled()) {
- log.debug("working directory is: " + client.printWorkingDirectory());
- }
- return client;
- }
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java
deleted file mode 100644
index 334deb992e..0000000000
--- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java
+++ /dev/null
@@ -1,58 +0,0 @@
-/*
- * 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.ftp.config;
-
-import org.w3c.dom.Element;
-
-import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.support.BeanDefinitionReaderUtils;
-import org.springframework.beans.factory.xml.ParserContext;
-import org.springframework.integration.config.xml.AbstractPollingInboundChannelAdapterParser;
-import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
-import org.springframework.integration.ftp.FtpSource;
-import org.springframework.integration.ftp.QueuedFTPClientPool;
-
-/**
- * Parser for the <inbound-channel-adapter/> element of the 'ftp' namespace.
- *
- * @author Mark Fisher
- * @author Marius Bogoevici
- * @author Iwein Fuld
- */
-public class FtpInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
-
- @Override
- protected String parseSource(Element element, ParserContext parserContext) {
- BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(FtpSource.class);
- String username = element.getAttribute("username");
- String password = element.getAttribute("password");
- String host = element.getAttribute("host");
- String port = element.getAttribute("port");
- String remoteWorkingDirectory = element.getAttribute("remote-working-directory");
- QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool();
- queuedFTPClientPool.setUsername(username);
- queuedFTPClientPool.setPassword(password);
- queuedFTPClientPool.setHost(host);
- 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());
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java
deleted file mode 100644
index 37c7f8ad49..0000000000
--- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * 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.ftp.config;
-
-import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
-
-/**
- * Namespace handler for the 'ftp' namespace.
- *
- * @author Mark Fisher
- */
-public class FtpNamespaceHandler extends NamespaceHandlerSupport {
-
- public void init() {
- this.registerBeanDefinitionParser("inbound-channel-adapter", new FtpInboundChannelAdapterParser());
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParser.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParser.java
deleted file mode 100644
index e905e39ca3..0000000000
--- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParser.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.ftp.config;
-
-import org.w3c.dom.Element;
-
-import org.springframework.beans.factory.support.AbstractBeanDefinition;
-import org.springframework.beans.factory.support.BeanDefinitionBuilder;
-import org.springframework.beans.factory.xml.ParserContext;
-import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
-import org.springframework.integration.ftp.FtpSendingMessageHandler;
-import org.springframework.integration.ftp.QueuedFTPClientPool;
-
-/**
- * Parser for the <outbound-channel-adapter/> element of the 'ftp' namespace.
- *
- * @author Iwein Fuld
- * @author Mark Fisher
- */
-public class FtpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
-
- @Override
- protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
- BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(FtpSendingMessageHandler.class);
- String username = element.getAttribute("username");
- String password = element.getAttribute("password");
- String host = element.getAttribute("host");
- String port = element.getAttribute("port");
- String remoteWorkingDirectory = element.getAttribute("remote-working-directory");
- QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool();
- queuedFTPClientPool.setUsername(username);
- queuedFTPClientPool.setPassword(password);
- queuedFTPClientPool.setHost(host);
- queuedFTPClientPool.setPort(Integer.parseInt(port));
- queuedFTPClientPool.setRemoteWorkingDirectory(remoteWorkingDirectory);
- builder.addConstructorArgValue(queuedFTPClientPool);
- return builder.getBeanDefinition();
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/spring-integration-ftp-1.0.xsd b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/spring-integration-ftp-1.0.xsd
deleted file mode 100644
index 1098325c3f..0000000000
--- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/spring-integration-ftp-1.0.xsd
+++ /dev/null
@@ -1,37 +0,0 @@
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- Defines an inbound FTP-polling Channel Adapter.
-
-
-
-
-
-
-
-
-
-
-
-
-
-
\ No newline at end of file
diff --git a/org.springframework.integration.ftp/src/main/resources/META-INF/spring.handlers b/org.springframework.integration.ftp/src/main/resources/META-INF/spring.handlers
deleted file mode 100644
index 4c588412b9..0000000000
--- a/org.springframework.integration.ftp/src/main/resources/META-INF/spring.handlers
+++ /dev/null
@@ -1 +0,0 @@
-http\://www.springframework.org/schema/integration/ftp=org.springframework.integration.ftp.config.FtpNamespaceHandler
\ No newline at end of file
diff --git a/org.springframework.integration.ftp/src/main/resources/META-INF/spring.schemas b/org.springframework.integration.ftp/src/main/resources/META-INF/spring.schemas
deleted file mode 100644
index 150f391de3..0000000000
--- a/org.springframework.integration.ftp/src/main/resources/META-INF/spring.schemas
+++ /dev/null
@@ -1 +0,0 @@
-http\://www.springframework.org/schema/integration/ftp/spring-integration-ftp-1.0.xsd=org/springframework/integration/ftp/config/spring-integration-ftp-1.0.xsd
\ No newline at end of file
diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/BacklogTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/BacklogTests.java
deleted file mode 100644
index da8b5993c7..0000000000
--- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/BacklogTests.java
+++ /dev/null
@@ -1,168 +0,0 @@
-/*
- * 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.ftp;
-
-import java.util.ArrayList;
-import java.util.concurrent.PriorityBlockingQueue;
-
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-import org.springframework.beans.DirectFieldAccessor;
-import org.springframework.integration.ftp.Backlog;
-import org.springframework.integration.ftp.FileSnapshot;
-
-/**
- * @author Marius Bogoevici
- * @author Iwein Fuld
- */
-@SuppressWarnings("unchecked")
-public class BacklogTests {
-
- private Backlog backlog;
-
- private ArrayList remoteSnapshot;
-
- private FileSnapshot[] process;
-
- @Before
- public void setUp() {
- backlog = new Backlog();
- process = new FileSnapshot[5];
- }
-
- @Test
- public void testInitialization() {
- Assert.assertTrue(backlog.isEmpty());
- backlog.processSnapshot(remoteSnapshot);
- PriorityBlockingQueue queue = (PriorityBlockingQueue) new DirectFieldAccessor(
- backlog).getPropertyValue("backlog");
- Assert.assertEquals(3, queue.size());
- Assert.assertTrue(queue.containsAll(remoteSnapshot));
- }
-
- @Test
- public void testFullProcessingInOneStep() {
- backlog.processSnapshot(remoteSnapshot);
- backlog.fileProcessed( remoteSnapshot.toArray(process));
- Assert.assertTrue(backlog.isEmpty());
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertTrue(backlog.isEmpty());
- }
-
- @Test
- public void testFullProcessingInTwoSteps() {
- backlog.processSnapshot(remoteSnapshot);
- backlog.fileProcessed( remoteSnapshot.subList(0, 2).toArray(process));
- PriorityBlockingQueue queue = (PriorityBlockingQueue) new DirectFieldAccessor(
- backlog).getPropertyValue("backlog");
- Assert.assertEquals(1, queue.size());
- Assert.assertTrue(queue.contains(remoteSnapshot.get(2)));
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertEquals(1, queue.size());
- Assert.assertTrue(queue.contains(remoteSnapshot.get(2)));
- backlog.fileProcessed(remoteSnapshot.get(2));
- Assert.assertTrue(backlog.isEmpty());
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertTrue(backlog.isEmpty());
- }
-
- @Test
- public void testOneFileChangedSize() {
- PriorityBlockingQueue queue = (PriorityBlockingQueue) new DirectFieldAccessor(
- backlog).getPropertyValue("backlog");
- backlog.processSnapshot(remoteSnapshot);
- backlog.fileProcessed( remoteSnapshot.toArray(process));
- Assert.assertTrue(backlog.isEmpty());
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertTrue(backlog.isEmpty());
- remoteSnapshot.remove(2);
- FileSnapshot modifiedC = new FileSnapshot("c.txt", 1001, 112);
- remoteSnapshot.add(modifiedC);
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertEquals(1, queue.size());
- Assert.assertTrue(queue.contains(modifiedC));
- }
-
- @Test
- public void testOneFileChangedDate() {
- PriorityBlockingQueue queue = (PriorityBlockingQueue) new DirectFieldAccessor(
- backlog).getPropertyValue("backlog");
-
- backlog.processSnapshot(remoteSnapshot);
- backlog.fileProcessed( remoteSnapshot.toArray(process));
- Assert.assertTrue(backlog.isEmpty());
- backlog.processSnapshot(remoteSnapshot);
- remoteSnapshot.remove(2);
- FileSnapshot modifiedC = new FileSnapshot("c.txt", 1011, 102);
- remoteSnapshot.add(modifiedC);
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertEquals(1, queue.size());
- Assert.assertTrue(queue.contains(modifiedC));
- }
-
- @Test
- public void testOneFileAdded() {
- PriorityBlockingQueue queue = (PriorityBlockingQueue) new DirectFieldAccessor(
- backlog).getPropertyValue("backlog");
- backlog.processSnapshot(remoteSnapshot);
- backlog.fileProcessed( remoteSnapshot.toArray(process));
- Assert.assertTrue(backlog.isEmpty());
- backlog.processSnapshot(remoteSnapshot);
- FileSnapshot newD = new FileSnapshot("d.txt", 1003, 103);
- remoteSnapshot.add(newD);
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertEquals(1, queue.size());
- Assert.assertTrue(queue.contains(newD));
- }
-
- @Test
- public void testOneFileRemoved() {
- backlog.processSnapshot(remoteSnapshot);
- backlog.fileProcessed( remoteSnapshot.toArray(process));
- Assert.assertTrue(backlog.isEmpty());
- backlog.processSnapshot(remoteSnapshot);
- remoteSnapshot.remove(2);
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertTrue(backlog.isEmpty());
- }
-
- @Test
- public void testOneFileRemovedBeforeBeingProcessedInTheNextStep() {
- PriorityBlockingQueue queue = (PriorityBlockingQueue) new DirectFieldAccessor(
- backlog).getPropertyValue("backlog");
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertEquals(3, queue.size());
- remoteSnapshot.remove(2);
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertEquals(2, queue.size());
- backlog.processSnapshot(remoteSnapshot);
- Assert.assertEquals(2, queue.size());
- }
-
- // @Test selectForProcessing success/failure
-
- @Before
- public void generateInitialSnapshot() {
- this.remoteSnapshot = new ArrayList();
- remoteSnapshot.add(new FileSnapshot("a.txt", 1000, 100));
- remoteSnapshot.add(new FileSnapshot("b.txt", 1001, 101));
- remoteSnapshot.add(new FileSnapshot("c.txt", 1002, 102));
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/ConcurrentBacklogTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/ConcurrentBacklogTests.java
deleted file mode 100644
index 4cce730c17..0000000000
--- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/ConcurrentBacklogTests.java
+++ /dev/null
@@ -1,190 +0,0 @@
-/*
- * 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.ftp;
-
-import static org.junit.Assert.assertTrue;
-
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import org.junit.Test;
-
-import org.springframework.beans.DirectFieldAccessor;
-import org.springframework.integration.ftp.Backlog;
-
-@SuppressWarnings("unchecked")
-public class ConcurrentBacklogTests {
-
- @Test(timeout = 1000)
- public void simultaneousPreparation() throws Exception {
- final Backlog backlog = new Backlog();
- backlog.processSnapshot(Arrays.asList(new String[] { "bert", "ernie", "pino", "whatsherface" }));
- Runnable todo = new Runnable() {
- public void run() {
- backlog.prepareForProcessing(1);
- }
- };
- CountDownLatch start = new CountDownLatch(1);
- CountDownLatch done = doConcurrently(5, todo, start);
- start.countDown();
- try {
- done.await();
- }
- catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- assertTrue(backlog.isEmpty());
- }
-
- @Test(timeout = 1000)
- public void concurrentUnloading() throws Exception {
- final Backlog backlog = new Backlog();
- List items = Arrays.asList(new String[] { "bert", "ernie", "pino", "whatsherface", "kaas", "pasf" });
- backlog.processSnapshot(items);
- final AtomicBoolean properlyUnloaded = new AtomicBoolean(false);
- Runnable todo = new Runnable() {
- public void run() {
- backlog.prepareForProcessing(2);
- backlog.processed();
- properlyUnloaded.set(backlog.isEmpty() && backlog.getProcessingBuffer().isEmpty());
- }
- };
- CountDownLatch start = new CountDownLatch(1);
- CountDownLatch done = doConcurrently(3, todo, start);
- start.countDown();
- try {
- done.await();
- }
- catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- assertTrue("currentlyProcessing not emptied correctly", ((Collection) new DirectFieldAccessor(backlog)
- .getPropertyValue("currentlyProcessing")).isEmpty());
- assertTrue("doneProcessing not populated correctly", ((Collection) new DirectFieldAccessor(backlog)
- .getPropertyValue("doneProcessing")).containsAll(items));
- }
-
- @Test(timeout = 1000)
- public void concurrentFailing() throws Exception {
- final Backlog backlog = new Backlog();
- List items = Arrays.asList(new String[] { "bert", "ernie", "pino", "whatsherface", "kaas", "pasf" });
- backlog.processSnapshot(items);
- final AtomicBoolean properlyBackedUp = new AtomicBoolean(false);
- Runnable todo = new Runnable() {
- public void run() {
- backlog.prepareForProcessing(2);
- backlog.processingFailed();
- properlyBackedUp.set(backlog.getProcessingBuffer().isEmpty());
- }
- };
- CountDownLatch start = new CountDownLatch(1);
- CountDownLatch done = doConcurrently(3, todo, start);
- start.countDown();
- try {
- done.await();
- }
- catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- assertTrue("currentlyProcessing not emptied correctly", ((Collection) new DirectFieldAccessor(backlog)
- .getPropertyValue("currentlyProcessing")).isEmpty());
- assertTrue("backlog not repopulated correctly", ((Collection) new DirectFieldAccessor(backlog)
- .getPropertyValue("backlog")).containsAll(items));
- }
-
- @Test(timeout = 1000)
- public void concurrentSuccessFailure() throws Exception {
- final Backlog backlog = new Backlog();
- List items = Arrays.asList(new String[] { "ham", "chicken", "burger", "cheeze" });
- backlog.processSnapshot(items);
- final AtomicBoolean properlyBackedUp = new AtomicBoolean(true);
- final AtomicBoolean properlyUnloaded = new AtomicBoolean(true);
- Runnable doFailure = new Runnable() {
- public void run() {
- backlog.prepareForProcessing(1);
- backlog.processingFailed();
- properlyBackedUp.set(backlog.getProcessingBuffer().isEmpty() && properlyBackedUp.get());
- }
- };
- Runnable doSuccess = new Runnable() {
- public void run() {
- backlog.prepareForProcessing(1);
- //make sure we process a message
- while (backlog.getProcessingBuffer().size() == 0) {
- Thread.yield();
- backlog.prepareForProcessing(1);
- }
- backlog.processed();
- properlyUnloaded.set(backlog.getProcessingBuffer().isEmpty() && properlyUnloaded.get());
- }
- };
- CountDownLatch start = new CountDownLatch(1);
- CountDownLatch doneFailure = doConcurrently(20, doFailure, start);
- CountDownLatch doneSuccess = doConcurrently(2, doSuccess, start);
- start.countDown();
- try {
- doneSuccess.await();
- doneFailure.await();
- }
- catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- assertTrue(properlyBackedUp.get());
- assertTrue(properlyUnloaded.get());
- assertTrue("currentlyProcessing not emptied correctly", ((Collection) new DirectFieldAccessor(backlog)
- .getPropertyValue("currentlyProcessing")).isEmpty());
- Collection backlogQueue = (Collection) new DirectFieldAccessor(backlog).getPropertyValue("backlog");
- assertTrue("backlog not repopulated correctly size is " + backlogQueue.size(), backlogQueue.size() == 2);
- Collection doneProcessing = (Collection) new DirectFieldAccessor(backlog).getPropertyValue("doneProcessing");
- assertTrue("doneProcessing not repopulated correctly size is " + doneProcessing.size(),
- doneProcessing.size() == 2);
- }
-
- /**
- * Convenience method to run part of a test concurrently in multiple threads
- *
- * @param numberOfThreads
- * @param todo the runnable that should be run by all the threads
- * @return a latch that will be counted down once all threads have run their
- * runnable.
- */
- private CountDownLatch doConcurrently(int numberOfThreads, final Runnable todo, final CountDownLatch start) {
- final CountDownLatch started = new CountDownLatch(numberOfThreads);
- final CountDownLatch done = new CountDownLatch(numberOfThreads);
- for (int i = 0; i < numberOfThreads; i++) {
- new Thread(new Runnable() {
-
- public void run() {
- started.countDown();
- try {
- started.await();
- start.await();
- }
- catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- todo.run();
- done.countDown();
- }
- }).start();
- }
- return done;
- }
-}
diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSendingMessageHandlerTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSendingMessageHandlerTests.java
deleted file mode 100644
index 85adee3ed9..0000000000
--- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSendingMessageHandlerTests.java
+++ /dev/null
@@ -1,92 +0,0 @@
-/*
- * 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.ftp;
-
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.isA;
-import static org.easymock.classextension.EasyMock.createMock;
-import static org.easymock.classextension.EasyMock.createNiceMock;
-import static org.easymock.classextension.EasyMock.replay;
-import static org.easymock.classextension.EasyMock.verify;
-
-import java.io.File;
-import java.io.FileInputStream;
-
-import org.apache.commons.net.ftp.FTPClient;
-import org.junit.Before;
-import org.junit.Test;
-
-import org.springframework.integration.core.Message;
-import org.springframework.integration.message.GenericMessage;
-import org.springframework.integration.message.MessageDeliveryException;
-
-/**
- * @author Iwein Fuld
- * @author Mark Fisher
- */
-public class FtpSendingMessageHandlerTests {
-
- private FtpSendingMessageHandler handler;
-
- private FTPClient ftpClient = createMock(FTPClient.class);
-
- /*
- * We don't want tests to worry about interaction with the pool (with the
- * exception of one dedicated test), so let's make the pool as transparent
- * as possible.
- */
- private FTPClientPool ftpClientPool = createNiceMock(FTPClientPool.class);
-
- /*
- * Handle to all mocks in this test so you can't forget to include one in a
- * replay, verify or reset call.
- */
- private Object[] allMocks = new Object[] { ftpClient, ftpClientPool };
-
-
- @Before
- public void liberalPool() throws Exception {
- expect(ftpClientPool.getClient()).andReturn(ftpClient).anyTimes();
- }
-
- @Before
- public void intitializeSubject() {
- this.handler = new FtpSendingMessageHandler(ftpClientPool);
- }
-
-
- // Tests
-
- @Test
- public void send() throws Exception {
- Message message = new GenericMessage(File.createTempFile("test", ".tmp"));
- expect(ftpClient.storeFile(isA(String.class), isA(FileInputStream.class))).andReturn(true);
- replay(allMocks);
- handler.handleMessage(message);
- verify(allMocks);
- }
-
- @Test(expected = MessageDeliveryException.class)
- public void sendFailed_negative() throws Exception {
- Message message = new GenericMessage(File.createTempFile("test", ".tmp"));
- expect(ftpClient.storeFile(isA(String.class), isA(FileInputStream.class))).andReturn(false);
- replay(allMocks);
- handler.handleMessage(message);
- verify(allMocks);
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSourceTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSourceTests.java
deleted file mode 100644
index b30117633a..0000000000
--- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSourceTests.java
+++ /dev/null
@@ -1,253 +0,0 @@
-/*
- * 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.ftp;
-
-import static org.easymock.EasyMock.eq;
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.isA;
-import static org.easymock.classextension.EasyMock.createMock;
-import static org.easymock.classextension.EasyMock.createNiceMock;
-import static org.easymock.classextension.EasyMock.replay;
-import static org.easymock.classextension.EasyMock.reset;
-import static org.easymock.classextension.EasyMock.verify;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-
-import java.io.File;
-import java.io.FilenameFilter;
-import java.io.OutputStream;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Calendar;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-
-import org.apache.commons.net.ftp.FTPClient;
-import org.apache.commons.net.ftp.FTPFile;
-import org.apache.oro.io.Perl5FilenameFilter;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import org.springframework.integration.core.Message;
-
-/**
- * @author Iwein Fuld
- */
-@SuppressWarnings("unchecked")
-public class FtpSourceTests {
-
- private FTPClient ftpClient = createMock(FTPClient.class);
-
- private FTPFile ftpFile = createMock(FTPFile.class);
-
- private FTPClientPool ftpClientPool = createNiceMock(FTPClientPool.class);
-
- private Object[] globalMocks = new Object[] { ftpClient, ftpFile, ftpClientPool };
-
- private FtpSource ftpSource;
-
- private Long size = 100l;
-
-
- @Before
- public void liberalPool() throws Exception {
- expect(ftpClientPool.getClient()).andReturn(ftpClient).anyTimes();
- }
-
- @Before
- public void initializeFtpSource() {
- ftpSource = new FtpSource(ftpClientPool);
- }
-
- @Before
- public void clearState() {
- reset(globalMocks);
- }
-
-
- @Test
- public void retrieveSingleFile() throws Exception {
- expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test1"));
- expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true);
- replay(globalMocks);
- Message> received = ftpSource.receive();
- ftpSource.onSend(received);
- verify(globalMocks);
- }
-
- private FTPFile[] mockedFTPFilesNamed(String... names) {
- List files = new ArrayList();
- // ensure difference by increasing size
- Calendar timestamp = Calendar.getInstance();
- size++;
- 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(size).anyTimes();
- files.add(ftpFile);
- replay(ftpFile);
- }
- return files.toArray(new FTPFile[] {});
- }
-
- @Test
- public void retrieveMultipleFiles() throws Exception {
- // get files
- expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test1", "test2")).times(2);
- expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true);
- expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true);
- List files = Arrays.asList(new File("test1"), new File("test2"));
-
- replay(globalMocks);
- Message receivedFiles = ftpSource.receive();
- ftpSource.onSend(receivedFiles);
- Message> secondReceived = ftpSource.receive();
- verify(globalMocks);
- assertEquals(files, receivedFiles.getPayload());
- assertNull(secondReceived);
- }
-
- @Test
- public void retrieveMultipleChangingFiles() throws Exception {
- // first run
- FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2");
- expect(ftpClient.listFiles()).andReturn(mockedFTPFiles);
- expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true);
- expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true);
-
- // second run, change the date so the messages should be retrieved again
- FTPFile[] mockedFTPFiles2 = mockedFTPFilesNamed("test1", "test2");
- expect(ftpClient.listFiles()).andReturn(mockedFTPFiles2);
- expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true);
- expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true);
- List files = Arrays.asList(new File("test1"), new File("test2"));
-
- replay(globalMocks);
- Message receivedFiles = ftpSource.receive();
- ftpSource.onSend(receivedFiles);
- ftpSource.onSend(ftpSource.receive());
- verify(globalMocks);
- assertEquals(files, receivedFiles.getPayload());
- }
-
- @Test
- public void retrieveMaxFilesPerMessage() throws Exception {
-
- this.ftpSource.setMaxFilesPerMessage(2);
- // assume client already connected
- FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2", "test3");
-
- // expect two receive runs
- expect(ftpClient.listFiles()).andReturn(mockedFTPFiles).times(2);
-
- // first run
- expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true);
- expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true);
- // second run
- expect(ftpClient.retrieveFile(eq("test3"), isA(OutputStream.class))).andReturn(true);
-
- replay(globalMocks);
- Message> receivedFiles1 = ftpSource.receive();
- ftpSource.onSend(receivedFiles1);
- Message> receivedFiles2 = ftpSource.receive();
- ftpSource.onSend(receivedFiles2);
- verify(globalMocks);
- List allReceived = new ArrayList(receivedFiles1.getPayload());
- allReceived.addAll(receivedFiles2.getPayload());
- assertEquals(2, receivedFiles1.getPayload().size());
- assertEquals(1, receivedFiles2.getPayload().size());
- assertTrue(allReceived.containsAll(Arrays.asList(new File[] { new File("test1"), new File("test2"),
- new File("test3") })));
- }
-
- @Test(timeout = 6000)
- @Ignore //not reliable
- public void concurrentPollingSunnyDay() throws Exception {
- final CountDownLatch recorded = new CountDownLatch(1);
- this.ftpSource.setMaxFilesPerMessage(2);
- // first run
- FTPFile[] mockedFTPFiles = mockedFTPFilesNamed("test1", "test2", "test3", "test4", "test5");
- expect(ftpClient.listFiles()).andReturn(mockedFTPFiles);
- expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true);
- expect(ftpClient.retrieveFile(eq("test2"), isA(OutputStream.class))).andReturn(true);
-
- // second poll
- expect(ftpClient.listFiles()).andReturn(mockedFTPFiles);
- expect(ftpClient.retrieveFile(eq("test3"), isA(OutputStream.class))).andReturn(true);
- expect(ftpClient.retrieveFile(eq("test4"), isA(OutputStream.class))).andReturn(true);
-
- expect(ftpClient.listFiles()).andReturn(mockedFTPFiles);
- expect(ftpClient.retrieveFile(eq("test5"), isA(OutputStream.class))).andReturn(true);
- replay(globalMocks);
- recorded.countDown();
-
- final CountDownLatch receivesDone = new CountDownLatch(3);
-
- for (int i = 0; i < 3; i++) {
- new Thread(new Runnable() {
- public void run() {
- Message> recievedFiles = null;
- try {
- // make sure receive happens after recording
- recorded.await();
- recievedFiles = ftpSource.receive();
- receivesDone.countDown();
- // make sure onSend happens after all receives
- receivesDone.await();
- }
- catch (InterruptedException e) {
- }
- finally {
- ftpSource.onSend(recievedFiles);
- }
- }
- }).start();
- }
- try {
- receivesDone.await();
- }
- catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- verify(globalMocks);
- }
-
- @Test
- public void onFailure() throws Exception {
- expect(ftpClient.listFiles()).andReturn(mockedFTPFilesNamed("test1")).times(2);
- expect(ftpClient.retrieveFile(eq("test1"), isA(OutputStream.class))).andReturn(true).times(2);
- replay(globalMocks);
- Message> received = ftpSource.receive();
- ftpSource.onFailure(received, new Exception("just a test"));
- assertEquals(received.getPayload(), ftpSource.receive().getPayload());
- verify(globalMocks);
- }
-
- @AfterClass
- public static void deleteFiles() {
- File file = new File("./");
- File[] files = file.listFiles((FilenameFilter) new Perl5FilenameFilter("test\\d"));
- for (File file2 : files) {
- file2.delete();
- }
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/QueuedFTPClientPoolTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/QueuedFTPClientPoolTests.java
deleted file mode 100644
index a21f297a01..0000000000
--- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/QueuedFTPClientPoolTests.java
+++ /dev/null
@@ -1,113 +0,0 @@
-/*
- * 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.ftp;
-
-import static junit.framework.Assert.assertEquals;
-import static junit.framework.Assert.assertSame;
-import static junit.framework.Assert.assertTrue;
-import static org.easymock.EasyMock.expect;
-import static org.easymock.classextension.EasyMock.createMock;
-import static org.easymock.classextension.EasyMock.createNiceMock;
-import static org.easymock.classextension.EasyMock.replay;
-import static org.easymock.classextension.EasyMock.verify;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.commons.net.ftp.FTPClient;
-import org.junit.Before;
-import org.junit.Test;
-
-import org.springframework.integration.ftp.FTPClientFactory;
-import org.springframework.integration.ftp.QueuedFTPClientPool;
-
-/**
- * @author Iwein Fuld
- */
-public class QueuedFTPClientPoolTests {
-
- private QueuedFTPClientPool pool;
-
- private FTPClientFactory factoryMock = createMock(FTPClientFactory.class);
-
- private Object[] allMocks = new Object[] { factoryMock };
-
-
- @Before
- public void initializeSubject() throws Exception {
- this.pool = new QueuedFTPClientPool(5);
- pool.setFactory(factoryMock);
- }
-
-
- @Test
- public void get() throws Exception {
- FTPClient expectedClient = new FTPClient();
- expect(factoryMock.getClient()).andReturn(expectedClient);
- replay(allMocks);
- FTPClient client = pool.getClient();
- assertEquals(expectedClient, client);
- verify(allMocks);
- }
-
- @Test
- public void getMultipleGet() throws Exception {
- FTPClient[] expectedClients = new FTPClient[] { mockedFTPClient(), mockedFTPClient(),
- mockedFTPClient(), mockedFTPClient(), mockedFTPClient(), mockedFTPClient() };
- for (FTPClient client : expectedClients) {
- expect(factoryMock.getClient()).andReturn(client);
- }
- replay(allMocks);
- for (int i = 0; i < 6; i++) {
- assertSame(expectedClients[i], pool.getClient());
- }
- verify(allMocks);
- }
-
- @Test
- public void getMultipleGetReleaseGet() throws Exception {
- FTPClient[] expectedClients = new FTPClient[] { mockedFTPClient(), mockedFTPClient(),
- mockedFTPClient(), mockedFTPClient(), mockedFTPClient() };
- for (FTPClient client : expectedClients) {
- expect(factoryMock.getClient()).andReturn(client);
- }
- replay(allMocks);
- List fromPool = new ArrayList();
- for (int i = 0; i < 5; i++) {
- fromPool.add(pool.getClient());
- }
- for (FTPClient client2 : fromPool) {
- pool.releaseClient(client2);
- }
- for (int i = 0; i < 5; i++) {
- FTPClient client = pool.getClient();
- boolean removed = fromPool.remove(client);
- assertTrue("Failed on element " + i, removed);
- }
- verify(allMocks);
- }
-
-
- private FTPClient mockedFTPClient() throws Exception {
- FTPClient mock = createNiceMock(FTPClient.class);
- expect(mock.isConnected()).andReturn(true).anyTimes();
- expect(mock.sendNoOp()).andReturn(true).anyTimes();
- replay(mock);
- return mock;
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java
deleted file mode 100644
index 2e5b228a85..0000000000
--- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.ftp.config;
-
-import static org.junit.Assert.assertEquals;
-
-import java.io.File;
-
-import org.junit.Test;
-
-import org.springframework.beans.DirectFieldAccessor;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.support.ClassPathXmlApplicationContext;
-
-/**
- * @author Mark Fisher
- * @author Marius Bogoevici
- * @author Iwein Fuld
- */
-public class FtpInboundChannelAdapterParserTests {
-
- @Test
- public void ftpInboundChannelAdapter() {
- ApplicationContext context = new ClassPathXmlApplicationContext(
- "ftpInboundChannelAdapterParserTests.xml", this.getClass());
- Object adapter = context.getBean("adapter");
- DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(
- new DirectFieldAccessor(adapter).getPropertyValue("source"));
- DirectFieldAccessor poolAccessor = new DirectFieldAccessor(
- sourceAccessor.getPropertyValue("clientPool"));
- assertEquals("testHost", poolAccessor.getPropertyValue("host"));
- assertEquals(2121, poolAccessor.getPropertyValue("port"));
- assertEquals(new File("/local"), sourceAccessor.getPropertyValue("localWorkingDirectory"));
- assertEquals("/remote", poolAccessor.getPropertyValue("remoteWorkingDirectory"));
- assertEquals("testUser", poolAccessor.getPropertyValue("username"));
- assertEquals("testPassword", poolAccessor.getPropertyValue("password"));
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSendingMessageConsumerIntegrationTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSendingMessageConsumerIntegrationTests.java
deleted file mode 100644
index e046798920..0000000000
--- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSendingMessageConsumerIntegrationTests.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * 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.ftp.config;
-
-import java.io.File;
-import java.io.FilenameFilter;
-
-import org.apache.oro.io.Perl5FilenameFilter;
-import org.junit.AfterClass;
-import org.junit.Before;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import org.springframework.integration.ftp.FtpSendingMessageHandler;
-import org.springframework.integration.ftp.QueuedFTPClientPool;
-import org.springframework.integration.message.GenericMessage;
-
-/**
- * @author Iwein Fuld
- */
-@Ignore
-public class FtpSendingMessageConsumerIntegrationTests {
-
- private FtpSendingMessageHandler handler;
-
- @Before
- public void initFtpTarget() {
- QueuedFTPClientPool clientPool = new QueuedFTPClientPool();
- clientPool.setHost("localhost");
- clientPool.setUsername("ftp-user");
- clientPool.setPassword("kaas");
- clientPool.setRemoteWorkingDirectory("ftp-test");
- handler = new FtpSendingMessageHandler(clientPool);
- }
-
- @Test
- public void send() throws Exception {
- File file = File.createTempFile("test", "");
- handler.handleMessage(new GenericMessage(file));
- }
-
- @AfterClass
- public static void deleteTestFiles() {
- File tmpDir = new File(System.getProperty("java.io.tmpdir"));
- File[] files = tmpDir.listFiles((FilenameFilter) new Perl5FilenameFilter("test\\d"));
- for (File file : files) {
- file.delete();
- }
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSourceIntegrationTests.java b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSourceIntegrationTests.java
deleted file mode 100644
index 5685ea5bfa..0000000000
--- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSourceIntegrationTests.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * 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.ftp.config;
-
-import static org.junit.Assert.assertTrue;
-
-import java.io.File;
-import java.util.List;
-
-import org.junit.Before;
-import org.junit.BeforeClass;
-import org.junit.Ignore;
-import org.junit.Test;
-
-import org.springframework.integration.core.Message;
-import org.springframework.integration.ftp.FtpSource;
-import org.springframework.integration.ftp.QueuedFTPClientPool;
-
-/**
- * @author Iwein Fuld
- */
-/*
- * These tests assume you have a local ftp server running. The whole class
- * should be disabled and only run when you have started your ftp server and are
- * in need of experimenting.
- *
- * To pass the test you should have an ftp server running at localhost that
- * accepts a login for ftp-user/kaas and has a remote directory ftp-test with at
- * least one file in it. Nothing is stopping you from changing the code to your
- * needs of course, this is just a starting point for local testing.
- */
-// ftp server dependency. comment away Ignore if you want to run this
-@Ignore
-public class FtpSourceIntegrationTests {
-
- private static File localWorkDir;
-
- private FtpSource ftpSource;
-
-
- @BeforeClass
- public static void initializeEnvironment() {
- localWorkDir = new File(System.getProperty("java.io.tmpdir") + "/" + FtpSourceIntegrationTests.class.getName());
- localWorkDir.mkdir();
- }
-
- @Before
- public void initializeFtpSource() throws Exception {
- QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool();
- ftpSource = new FtpSource(queuedFTPClientPool);
- queuedFTPClientPool.setHost("localhost");
- queuedFTPClientPool.setUsername("ftp-user");
- queuedFTPClientPool.setPassword("kaas");
- ftpSource.setLocalWorkingDirectory(localWorkDir);
- queuedFTPClientPool.setRemoteWorkingDirectory("ftp-test");
- }
-
- @Test
- public void receive() {
- Message> received = ftpSource.receive();
- assertTrue(received.getPayload().iterator().next().exists());
- }
-
-}
diff --git a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/ftpInboundChannelAdapterParserTests.xml b/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/ftpInboundChannelAdapterParserTests.xml
deleted file mode 100644
index 527faabb22..0000000000
--- a/org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/ftpInboundChannelAdapterParserTests.xml
+++ /dev/null
@@ -1,24 +0,0 @@
-
-
-
-
-
-
-
-
diff --git a/org.springframework.integration.ftp/template.mf b/org.springframework.integration.ftp/template.mf
deleted file mode 100644
index 6ce11594f8..0000000000
--- a/org.springframework.integration.ftp/template.mf
+++ /dev/null
@@ -1,13 +0,0 @@
-Bundle-SymbolicName: org.springframework.integration.ftp
-Bundle-Name: Spring Integration FTP Support
-Bundle-Vendor: SpringSource
-Bundle-ManifestVersion: 2
-Import-Template:
- org.springframework.integration.*;version="[1.0.0, 1.0.1)",
- org.springframework.beans.*;version="[2.5.6, 3.0.0)",
- org.springframework.util;version="[2.5.6, 3.0.0)",
- org.apache.commons.logging;version="[1.1.1, 2.0.0)",
- org.apache.commons.net.ftp;version="[1.4.1, 2.0.0)";resolution:=optional
-Unversioned-Imports:
- org.w3c.dom
-