Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Removed FTP module

Browse Source
This commit is contained in:
Mark Fisher
2008-11-03 15:51:22 +00:00
parent 3b30cbd4f5
commit a1f6d4a468
28 changed files with 0 additions and 2235 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
org.springframework.integration.ftp/.classpath
View File

@@ -1,21 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<classpath>
<classpathentry kind="src" path="src/main/java"/>
<classpathentry kind="src" output="target/test-classes" path="src/test/java"/>
<classpathentry kind="src" path="src/main/resources"/>
<classpathentry kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/>
<classpathentry kind="var" path="IVY_CACHE/org.apache.commons/com.springsource.org.apache.commons.logging/1.1.1/com.springsource.org.apache.commons.logging-1.1.1.jar" sourcepath="/IVY_CACHE/org.apache.commons/com.springsource.org.apache.commons.logging/1.1.1/com.springsource.org.apache.commons.logging-sources-1.1.1.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.apache.commons/com.springsource.org.apache.commons.net/1.4.1/com.springsource.org.apache.commons.net-1.4.1.jar" sourcepath="/IVY_CACHE/org.apache.commons/com.springsource.org.apache.commons.net/1.4.1/com.springsource.org.apache.commons.net-sources-1.4.1.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.apache.oro/com.springsource.org.apache.oro/2.0.8/com.springsource.org.apache.oro-2.0.8.jar" sourcepath="/IVY_CACHE/org.apache.oro/com.springsource.org.apache.oro/2.0.8/com.springsource.org.apache.oro-sources-2.0.8.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.junit/com.springsource.org.junit/4.4.0/com.springsource.org.junit-4.4.0.jar" sourcepath="/IVY_CACHE/org.junit/com.springsource.org.junit/4.4.0/com.springsource.org.junit-sources-4.4.0.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.springframework/org.springframework.beans/2.5.6/org.springframework.beans-2.5.6.jar" sourcepath="/IVY_CACHE/org.springframework/org.springframework.beans/2.5.6/org.springframework.beans-sources-2.5.6.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.springframework/org.springframework.context/2.5.6/org.springframework.context-2.5.6.jar" sourcepath="/IVY_CACHE/org.springframework/org.springframework.context/2.5.6/org.springframework.context-sources-2.5.6.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.springframework/org.springframework.context.support/2.5.6/org.springframework.context.support-2.5.6.jar" sourcepath="/IVY_CACHE/org.springframework/org.springframework.context.support/2.5.6/org.springframework.context.support-sources-2.5.6.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.springframework/org.springframework.core/2.5.6/org.springframework.core-2.5.6.jar" sourcepath="/IVY_CACHE/org.springframework/org.springframework.core/2.5.6/org.springframework.core-sources-2.5.6.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.springframework/org.springframework.test/2.5.6/org.springframework.test-2.5.6.jar" sourcepath="/IVY_CACHE/org.springframework/org.springframework.test/2.5.6/org.springframework.test-sources-2.5.6.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.easymock/com.springsource.org.easymock/2.3.0/com.springsource.org.easymock-2.3.0.jar" sourcepath="/IVY_CACHE/org.easymock/com.springsource.org.easymock/2.3.0/com.springsource.org.easymock-sources-2.3.0.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.easymock/com.springsource.org.easymock.classextension/2.3.0/com.springsource.org.easymock.classextension-2.3.0.jar" sourcepath="/IVY_CACHE/org.easymock/com.springsource.org.easymock.classextension/2.3.0/com.springsource.org.easymock.classextension-sources-2.3.0.jar"/>
<classpathentry kind="var" path="IVY_CACHE/net.sourceforge.cglib/com.springsource.net.sf.cglib/2.1.3/com.springsource.net.sf.cglib-2.1.3.jar" sourcepath="/IVY_CACHE/net.sourceforge.cglib/com.springsource.net.sf.cglib/2.1.3/com.springsource.net.sf.cglib-sources-2.1.3.jar"/>
<classpathentry combineaccessrules="false" kind="src" path="/org.springframework.integration"/>
<classpathentry kind="output" path="target/classes"/>
</classpath>

17
org.springframework.integration.ftp/.project
View File

@@ -1,17 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<projectDescription>
<name>org.springframework.integration.ftp</name>
<comment></comment>
<projects>
</projects>
<buildSpec>
<buildCommand>
<name>org.eclipse.jdt.core.javabuilder</name>
<arguments>
</arguments>
</buildCommand>
</buildSpec>
<natures>
<nature>org.eclipse.jdt.core.javanature</nature>
</natures>
</projectDescription>

8
org.springframework.integration.ftp/build.xml
View File

@@ -1,8 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<project name="org.springframework.integration.ftp">
<property file="${basedir}/../build.properties"/>
<import file="${basedir}/../build-spring-integration/package-bundle.xml"/>
<import file="${basedir}/../spring-build/standard/default.xml"/>
</project>

33
org.springframework.integration.ftp/ivy.xml
View File

@@ -1,33 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<?xml-stylesheet type="text/xsl" href="http://ivyrep.jayasoft.org/ivy-doc.xsl"?>
<ivy-module
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:noNamespaceSchemaLocation="http://incubator.apache.org/ivy/schemas/ivy.xsd"
version="1.3">
<info organisation="org.springframework.integration" module="${ant.project.name}">
<license name="Apache 2.0" url="http://www.apache.org/licenses/LICENSE-2.0"/>
<ivyauthor name="Mark Fisher"/>
</info>
<configurations>
<include file="${spring.build.dir}/common/default-ivy-configurations.xml"/>
</configurations>
<publications>
<artifact name="${ant.project.name}"/>
<artifact name="${ant.project.name}-sources" type="src" ext="jar"/>
</publications>
<dependencies>
<dependency org="net.sourceforge.cglib" name="com.springsource.net.sf.cglib" rev="2.1.3" conf="test->runtime"/>
<dependency org="org.apache.commons" name="com.springsource.org.apache.commons.net" rev="1.4.1" conf="optional->runtime"/>
<dependency org="org.apache.oro" name="com.springsource.org.apache.oro" rev="2.0.8" conf="optional->runtime"/>
<dependency org="org.easymock" name="com.springsource.org.easymock" rev="2.3.0" conf="test->runtime"/>
<dependency org="org.easymock" name="com.springsource.org.easymock.classextension" rev="2.3.0" conf="test->runtime"/>
<dependency org="org.junit" name="com.springsource.org.junit" rev="4.4.0" conf="test->runtime"/>
<dependency org="org.springframework" name="org.springframework.context.support" rev="2.5.6" conf="compile->runtime"/>
<dependency org="org.springframework.integration" name="org.springframework.integration" rev="latest.integration" conf="compile->compile"/>
</dependencies>
</ivy-module>

124
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/AbstractDirectorySource.java
View File

@@ -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<T> implements MessageSource<T>, MessageDeliveryAware<T> {
public final static String FILE_INFO_PROPERTY = "file.info";
protected final Log logger = LogFactory.getLog(this.getClass());
private final Backlog<FileSnapshot> backlog;
public AbstractDirectorySource() {
this(null);
}
public AbstractDirectorySource(Comparator<FileSnapshot> comparator) {
this.backlog = comparator == null ? new Backlog<FileSnapshot>() : new Backlog<FileSnapshot>(comparator);
}
protected Backlog<FileSnapshot> getBacklog() {
return this.backlog;
}
public final Message<T> 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<FileSnapshot> backlog) throws IOException {
List<FileSnapshot> snapshot = new ArrayList<FileSnapshot>();
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<T> buildNextMessage() throws IOException {
return MessageBuilder.withPayload(retrieveNextPayload()).build();
}
public void onSend(Message<T> message) {
if (logger.isDebugEnabled()) {
logger.debug(message + " processed successfully. Files will be removed from backlog");
}
this.backlog.processed();
}
public void onFailure(Message<T> 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<FileSnapshot> 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();
}
}

227
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/Backlog.java
View File

@@ -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<T extends Comparable<T>> {
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<T> backlog;
// @GuardedBy(this)
private Set<T> doneProcessing = new HashSet<T>();
// @GuardedBy(this)
private Set<T> currentlyProcessing = new HashSet<T>();
/*
* This is the storage for backlog that is being processed by a specific
* thread.
*/
private ThreadLocal<List<T>> processingBuffer = new ThreadLocal<List<T>>() {
@Override
protected List<T> initialValue() {
return new ArrayList<T>();
}
};
/**
* Constructs a Backlog around a naturally ordered
* {@link PriorityBlockingQueue}.
*/
public Backlog() {
this.backlog = new PriorityBlockingQueue<T>();
}
/**
* 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<? super T> comparator) {
this.backlog = new PriorityBlockingQueue<T>(INITIAL_QUEUE_CAPACITY, comparator);
}
public void processSnapshot(List<T> 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<T> newInCurrentSnapshot = new ArrayList<T>(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
* <code>{@link #prepareForProcessing(int)}</code>. 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
* <code>{@link #prepareForProcessing(int)}</code> 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<T> 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<T> selectForProcessing(int maxBatchSize) {
prepareForProcessing(maxBatchSize);
return getProcessingBuffer();
}
/**
* Marks all from the processing buffer as done. Use only in combination
* with <code>{@link #prepareForProcessing(int)}</code>
*/
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 <code>{@link #prepareForProcessing(int)}</code>
*/
public void processingFailed() {
if (logger.isDebugEnabled()) {
logger.debug("Moving processing buffer " + processingBuffer.get()
+ " back to backlog. Processing has failed");
}
List<T> processing = this.processingBuffer.get();
synchronized (this) {
this.backlog.addAll(processing);
currentlyProcessing.removeAll(processing);
}
processing.clear();
}
public List<T> 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 <code>true</code> if both the thread local processing buffer and
* the backlog are empty.
*/
public boolean isEmpty() {
return this.backlog.isEmpty() && this.processingBuffer.get().isEmpty();
}
}

38
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FTPClientFactory.java
View File

@@ -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;
}

44
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FTPClientPool.java
View File

@@ -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.
* <P>
* The caller should NOT disconnect the client before calling this method.
* <p>
* 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);
}

101
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FileSnapshot.java
View File

@@ -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.
* <p>
* 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<FileSnapshot> {
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;
}
/**
* <p>
* Be careful to note that the file that this snapshot refers to might have
* changed. In particular: <code>
* snapshot.getModificationTimestamp() != snapshot.getFile().lastModified()
* </code> will evalutate to <code>true</code> 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());
}
}

78
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSendingMessageHandler.java
View File

@@ -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);
}
}
}
}

114
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/FtpSource.java
View File

@@ -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<List<File>> {
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<FileSnapshot> directoryContentManager) throws IOException {
List<FileSnapshot> snapshot = new ArrayList<FileSnapshot>();
populateSnapshot(snapshot);
directoryContentManager.processSnapshot(snapshot);
directoryContentManager.prepareForProcessing(maxFilesPerMessage);
}
@Override
protected void populateSnapshot(List<FileSnapshot> 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<File> retrieveNextPayload() throws IOException {
FTPClient client = this.clientPool.getClient();
try {
List<File> files = new ArrayList<File>();
List<FileSnapshot> 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);
}
}
}

198
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/QueuedFTPClientPool.java
View File

@@ -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.
* <p>
* 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<FTPClient> 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<FTPClient>(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.
* <p>
* It is possible that released clients are disconnected by the remote
* server (@see {@link FTPClient#sendNoOp()}. In this case getClient is
* called recursively to obtain a client that is still alive. For this
* 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;
}
}
}

58
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParser.java
View File

@@ -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 &lt;inbound-channel-adapter/&gt; 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());
}
}

32
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpNamespaceHandler.java
View File

@@ -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());
}
}

54
org.springframework.integration.ftp/src/main/java/org/springframework/integration/ftp/config/FtpOutboundChannelAdapterParser.java
View File

@@ -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 &lt;outbound-channel-adapter/&gt; 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();
}
}

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

@@ -1,37 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<xsd:schema xmlns="http://www.springframework.org/schema/integration/ftp"
xmlns:xsd="http://www.w3.org/2001/XMLSchema"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:tool="http://www.springframework.org/schema/tool"
targetNamespace="http://www.springframework.org/schema/integration/ftp"
elementFormDefault="qualified"
attributeFormDefault="unqualified">
<xsd:import namespace="http://www.springframework.org/schema/beans"/>
<xsd:import namespace="http://www.springframework.org/schema/tool"/>
<xsd:annotation>
<xsd:documentation><![CDATA[
Defines the configuration elements for Spring Integration's FTP Channel Adapters.
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="inbound-channel-adapter">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines an inbound FTP-polling Channel Adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="channel" type="xsd:string"/>
<xsd:attribute name="username" type="xsd:string" use="optional"/>
<xsd:attribute name="password" type="xsd:string" use="optional"/>
<xsd:attribute name="host" type="xsd:string" use="required"/>
<xsd:attribute name="port" type="xsd:int" use="optional"/>
<xsd:attribute name="local-working-directory" type="xsd:string" use="required"/>
<xsd:attribute name="remote-working-directory" type="xsd:string" use="optional"/>
</xsd:complexType>
</xsd:element>
</xsd:schema>

1
org.springframework.integration.ftp/src/main/resources/META-INF/spring.handlers
View File

@@ -1 +0,0 @@
http\://www.springframework.org/schema/integration/ftp=org.springframework.integration.ftp.config.FtpNamespaceHandler

1
org.springframework.integration.ftp/src/main/resources/META-INF/spring.schemas
View File

@@ -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

168
org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/BacklogTests.java
View File

@@ -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<FileSnapshot> backlog;
private ArrayList<FileSnapshot> remoteSnapshot;
private FileSnapshot[] process;
@Before
public void setUp() {
backlog = new Backlog<FileSnapshot>();
process = new FileSnapshot[5];
}
@Test
public void testInitialization() {
Assert.assertTrue(backlog.isEmpty());
backlog.processSnapshot(remoteSnapshot);
PriorityBlockingQueue<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) 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<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) 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<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) 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<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) 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<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) 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<FileSnapshot> queue = (PriorityBlockingQueue<FileSnapshot>) 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<FileSnapshot>();
remoteSnapshot.add(new FileSnapshot("a.txt", 1000, 100));
remoteSnapshot.add(new FileSnapshot("b.txt", 1001, 101));
remoteSnapshot.add(new FileSnapshot("c.txt", 1002, 102));
}
}

190
org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/ConcurrentBacklogTests.java
View File

@@ -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<String> 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<String> 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<String> 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;
}
}

92
org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSendingMessageHandlerTests.java
View File

@@ -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>(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>(File.createTempFile("test", ".tmp"));
expect(ftpClient.storeFile(isA(String.class), isA(FileInputStream.class))).andReturn(false);
replay(allMocks);
handler.handleMessage(message);
verify(allMocks);
}
}

253
org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/FtpSourceTests.java
View File

@@ -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<List<File>> received = ftpSource.receive();
ftpSource.onSend(received);
verify(globalMocks);
}
private FTPFile[] mockedFTPFilesNamed(String... names) {
List<FTPFile> files = new ArrayList<FTPFile>();
// 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<File> files = Arrays.asList(new File("test1"), new File("test2"));
replay(globalMocks);
Message receivedFiles = ftpSource.receive();
ftpSource.onSend(receivedFiles);
Message<List<File>> 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<File> 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<List<File>> receivedFiles1 = ftpSource.receive();
ftpSource.onSend(receivedFiles1);
Message<List<File>> receivedFiles2 = ftpSource.receive();
ftpSource.onSend(receivedFiles2);
verify(globalMocks);
List<File> allReceived = new ArrayList<File>(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<List<File>> 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<List<File>> 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();
}
}
}

113
org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/QueuedFTPClientPoolTests.java
View File

@@ -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<FTPClient> fromPool = new ArrayList<FTPClient>();
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;
}
}

53
org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpInboundChannelAdapterParserTests.java
View File

@@ -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"));
}
}

65
org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSendingMessageConsumerIntegrationTests.java
View File

@@ -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>(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();
}
}
}

78
org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/FtpSourceIntegrationTests.java
View File

@@ -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<List<File>> received = ftpSource.receive();
assertTrue(received.getPayload().iterator().next().exists());
}
}

24
org.springframework.integration.ftp/src/test/java/org/springframework/integration/ftp/config/ftpInboundChannelAdapterParserTests.xml
View File

@@ -1,24 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration/ftp"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:integration="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd
http://www.springframework.org/schema/integration/ftp
http://www.springframework.org/schema/integration/ftp/spring-integration-ftp-1.0.xsd">
<integration:channel id="channel"/>
<inbound-channel-adapter id="adapter"
channel="channel"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"/>
</beans:beans>

13
org.springframework.integration.ftp/template.mf
View File

@@ -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