From a1f6d4a4688f0cb173d825385cd07539634c1e4a Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 3 Nov 2008 15:51:22 +0000 Subject: [PATCH] Removed FTP module --- .../.classpath | 21 -- org.springframework.integration.ftp/.project | 17 -- org.springframework.integration.ftp/build.xml | 8 - org.springframework.integration.ftp/ivy.xml | 33 --- .../ftp/AbstractDirectorySource.java | 124 --------- .../integration/ftp/Backlog.java | 227 ---------------- .../integration/ftp/FTPClientFactory.java | 38 --- .../integration/ftp/FTPClientPool.java | 44 --- .../integration/ftp/FileSnapshot.java | 101 ------- .../ftp/FtpSendingMessageHandler.java | 78 ------ .../integration/ftp/FtpSource.java | 114 -------- .../integration/ftp/QueuedFTPClientPool.java | 198 -------------- .../FtpInboundChannelAdapterParser.java | 58 ---- .../ftp/config/FtpNamespaceHandler.java | 32 --- .../FtpOutboundChannelAdapterParser.java | 54 ---- .../ftp/config/spring-integration-ftp-1.0.xsd | 37 --- .../main/resources/META-INF/spring.handlers | 1 - .../main/resources/META-INF/spring.schemas | 1 - .../integration/ftp/BacklogTests.java | 168 ------------ .../ftp/ConcurrentBacklogTests.java | 190 ------------- .../ftp/FtpSendingMessageHandlerTests.java | 92 ------- .../integration/ftp/FtpSourceTests.java | 253 ------------------ .../ftp/QueuedFTPClientPoolTests.java | 113 -------- .../FtpInboundChannelAdapterParserTests.java | 53 ---- ...endingMessageConsumerIntegrationTests.java | 65 ----- .../ftp/config/FtpSourceIntegrationTests.java | 78 ------ .../ftpInboundChannelAdapterParserTests.xml | 24 -- .../template.mf | 13 - 28 files changed, 2235 deletions(-) delete mode 100644 org.springframework.integration.ftp/.classpath delete mode 100644 org.springframework.integration.ftp/.project delete mode 100644 org.springframework.integration.ftp/build.xml delete mode 100644 org.springframework.integration.ftp/ivy.xml delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/AbstractDirectorySource.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/Backlog.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FTPClientFactory.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FTPClientPool.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FileSnapshot.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSendingMessageHandler.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSource.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParser.java delete mode 100644 org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/spring-integration-ftp-1.0.xsd delete mode 100644 org.springframework.integration.ftp/src/main/resources/META-INF/spring.handlers delete mode 100644 org.springframework.integration.ftp/src/main/resources/META-INF/spring.schemas delete mode 100644 org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/BacklogTests.java delete mode 100644 org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/ConcurrentBacklogTests.java delete mode 100644 org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSendingMessageHandlerTests.java delete mode 100644 org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSourceTests.java delete mode 100644 org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/QueuedFTPClientPoolTests.java delete mode 100644 org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java delete mode 100644 org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSendingMessageConsumerIntegrationTests.java delete mode 100644 org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSourceIntegrationTests.java delete mode 100644 org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/ftpInboundChannelAdapterParserTests.xml delete mode 100644 org.springframework.integration.ftp/template.mf diff --git a/org.springframework.integration.ftp/.classpath b/org.springframework.integration.ftp/.classpath deleted file mode 100644 index bf1cd742fc..0000000000 --- a/org.springframework.integration.ftp/.classpath +++ /dev/null @@ -1,21 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - diff --git a/org.springframework.integration.ftp/.project b/org.springframework.integration.ftp/.project deleted file mode 100644 index 77dc70cbd5..0000000000 --- a/org.springframework.integration.ftp/.project +++ /dev/null @@ -1,17 +0,0 @@ - - - org.springframework.integration.ftp - - - - - - org.eclipse.jdt.core.javabuilder - - - - - - org.eclipse.jdt.core.javanature - - diff --git a/org.springframework.integration.ftp/build.xml b/org.springframework.integration.ftp/build.xml deleted file mode 100644 index 1dc7d4c3a8..0000000000 --- a/org.springframework.integration.ftp/build.xml +++ /dev/null @@ -1,8 +0,0 @@ - - - - - - - - diff --git a/org.springframework.integration.ftp/ivy.xml b/org.springframework.integration.ftp/ivy.xml deleted file mode 100644 index 62181010c4..0000000000 --- a/org.springframework.integration.ftp/ivy.xml +++ /dev/null @@ -1,33 +0,0 @@ - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/AbstractDirectorySource.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/AbstractDirectorySource.java deleted file mode 100644 index f94bb65034..0000000000 --- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/AbstractDirectorySource.java +++ /dev/null @@ -1,124 +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.util.ArrayList; -import java.util.Comparator; -import java.util.List; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.integration.core.Message; -import org.springframework.integration.core.MessagingException; -import org.springframework.integration.message.MessageBuilder; -import org.springframework.integration.message.MessageDeliveryAware; -import org.springframework.integration.message.MessageSource; - -/** - * Base class for implementing a PollableSource that creates messages from files - * in a directory, either local or remote. - * - * @author Marius Bogoevici - * @author Iwein Fuld - */ -public abstract class AbstractDirectorySource implements MessageSource, MessageDeliveryAware { - - public final static String FILE_INFO_PROPERTY = "file.info"; - - protected final Log logger = LogFactory.getLog(this.getClass()); - - private final Backlog backlog; - - - public AbstractDirectorySource() { - this(null); - } - - public AbstractDirectorySource(Comparator comparator) { - this.backlog = comparator == null ? new Backlog() : new Backlog(comparator); - } - - - protected Backlog getBacklog() { - return this.backlog; - } - - public final Message receive() { - try { - refreshSnapshotAndMarkProcessing(this.backlog); - if (!getBacklog().isEmpty()) { - return buildNextMessage(); - } - return null; - } - catch (Exception e) { - throw new MessagingException("Error while polling for messages.", e); - } - } - - protected void refreshSnapshotAndMarkProcessing(Backlog backlog) throws IOException { - List snapshot = new ArrayList(); - this.populateSnapshot(snapshot); - backlog.processSnapshot(snapshot); - } - - /** - * Hook point for implementors to create the next message that should be - * 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 buildNextMessage() throws IOException { - return MessageBuilder.withPayload(retrieveNextPayload()).build(); - } - - public void onSend(Message message) { - if (logger.isDebugEnabled()) { - logger.debug(message + " processed successfully. Files will be removed from backlog"); - } - this.backlog.processed(); - } - - public void onFailure(Message failedMessage, Throwable exception) { - if (this.logger.isWarnEnabled()) { - this.logger.warn("Failure notification received by [" + this.getClass().getSimpleName() + "] for message: " - + failedMessage + ". Selected files will be moved back to the backlog.", exception); - } - this.backlog.processingFailed(); - } - - /** - * Constructs the snapshot by iterating files. - */ - protected abstract void populateSnapshot(List snapshot) throws IOException; - - /** - * Returns the next file, based on the backlog data. - */ - protected abstract T retrieveNextPayload() throws IOException; - - protected final void filesProcessed() { - this.backlog.processed(); - } - -} diff --git a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/Backlog.java b/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/Backlog.java deleted file mode 100644 index c1a0050339..0000000000 --- a/org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/Backlog.java +++ /dev/null @@ -1,227 +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.Collection; -import java.util.Collections; -import java.util.Comparator; -import java.util.HashSet; -import java.util.List; -import java.util.Set; -import java.util.concurrent.PriorityBlockingQueue; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.util.Assert; - -/** - * Keeps track of a backlog in a threadsafe, stateful manner. - * - * @author Marius Bogoevici - * @author Iwein Fuld - */ -public class Backlog> { - - private static final int INITIAL_QUEUE_CAPACITY = 5; - - private final Log logger = LogFactory.getLog(this.getClass()); - - /* - * Backlog, doneProcessing and currentlyProcessing should be in consistent - * state together. To do that access to them is synchronized on this and - * atomic operations have been defined on this class. It is not a problem if - * items exist in more than one of these collections at the same time, but the - * item should not be removed from one collection before it is added to the - * next. - */ - private final PriorityBlockingQueue backlog; - - // @GuardedBy(this) - private Set doneProcessing = new HashSet(); - - // @GuardedBy(this) - private Set currentlyProcessing = new HashSet(); - - /* - * This is the storage for backlog that is being processed by a specific - * thread. - */ - private ThreadLocal> processingBuffer = new ThreadLocal>() { - @Override - protected List initialValue() { - return new ArrayList(); - } - }; - - /** - * Constructs a Backlog around a naturally ordered - * {@link PriorityBlockingQueue}. - */ - public Backlog() { - this.backlog = new PriorityBlockingQueue(); - } - - /** - * Constructs a backlog around a {@link PriorityBlockingQueue} that is - * created with the supplied comparator. For natural ordering use - * {@link #Backlog()}. - * @param comparator - */ - public Backlog(Comparator comparator) { - this.backlog = new PriorityBlockingQueue(INITIAL_QUEUE_CAPACITY, comparator); - } - - public void processSnapshot(List currentSnapshot) { - /* - * clear the threadLocal backlog. When the thread processes a new - * snapshot it is done with the previous message. If there are still - * messages in the processing buffer something is wrong because they - * were not processed, nor raised as failed. - */ - Assert.isTrue(processingBuffer.get().isEmpty(), "Processing buffer not emptied before poll."); - /* - * compute everything that is new to the backlog preventing side effect - * on currentSnapshot. - */ - Collection newInCurrentSnapshot = new ArrayList(currentSnapshot); - synchronized (this) { - newInCurrentSnapshot.removeAll(backlog); - newInCurrentSnapshot.removeAll(currentlyProcessing); - newInCurrentSnapshot.removeAll(doneProcessing); - - backlog.retainAll(currentSnapshot); - doneProcessing.retainAll(currentSnapshot); - backlog.addAll(newInCurrentSnapshot); - } - } - - /** - * When a source is using a backlog in a single threaded context it can use - * this method instead of using - * {@link #prepareForProcessing(int)}. In a threaded scenario - * this method can be used to mark a subset of the processing buffer as - * processed. Calling this method in a threaded scenario without using - * {@link #prepareForProcessing(int)} will manipulate the - * backlog directly. In threaded scenarios a call to - * {@link #selectForProcessing(int)} followed by a call to - * {@link #processed()} is recommended. - * @param items the items that have been processed - */ - public synchronized void fileProcessed(T... items) { - for (T item : items) { - if (item != null) { - if (logger.isDebugEnabled()) { - logger.debug("Removing item '" + item + "' from the undo buffer. It has been processed."); - } - this.doneProcessing.add(item); - this.backlog.remove(item); - this.processingBuffer.get().remove(item); - } - } - } - - /** - * Moves items from the backlog to a thread local processing buffer, - * reserving them for this thread. It is the responsibility of the reserving - * thread to process the items and provide feedback to the backlog through - * {@link #processed()} or {@link #processingFailed()}. - * @param maxBatchSize if -1 prepare the whole backlog. - */ - public void prepareForProcessing(int maxBatchSize) { - List processingBuffer = this.processingBuffer.get(); - /* - * It is important to properly lock the access to backlog and - * currentlyProcessing in this case, because the removal from the - * backlog happens before addition to currently processed. If another - * thread accesses this type of state it might duplicate messages from - * the source into the backlog. - */ - synchronized (this) { - if (maxBatchSize == -1) { - this.backlog.drainTo(processingBuffer); - } - else { - this.backlog.drainTo(processingBuffer, maxBatchSize); - } - currentlyProcessing.addAll(processingBuffer); - } - if (logger.isDebugEnabled()) { - logger.debug("Preparing " + processingBuffer + " for processing"); - } - } - - /** - * Convenience method that returns the items prepared for processing - * immediately. - * @param maxBatchSize - * @return prepared processing buffer - */ - public List selectForProcessing(int maxBatchSize) { - prepareForProcessing(maxBatchSize); - return getProcessingBuffer(); - } - - /** - * Marks all from the processing buffer as done. Use only in combination - * with {@link #prepareForProcessing(int)} - */ - public void processed() { - if (logger.isDebugEnabled()) { - logger.debug("Moving processing buffer " + processingBuffer.get() + " to doneProcessing"); - } - synchronized (this) { - this.doneProcessing.addAll(this.processingBuffer.get()); - currentlyProcessing.removeAll(processingBuffer.get()); - } - this.processingBuffer.get().clear(); - } - - /** - * Marks all from the processing buffer as not done. Use only in combination - * with {@link #prepareForProcessing(int)} - */ - public void processingFailed() { - if (logger.isDebugEnabled()) { - logger.debug("Moving processing buffer " + processingBuffer.get() - + " back to backlog. Processing has failed"); - } - List processing = this.processingBuffer.get(); - synchronized (this) { - this.backlog.addAll(processing); - currentlyProcessing.removeAll(processing); - } - processing.clear(); - } - - public List 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 -