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

Migrated FTP Channel Adapter code from the 'org.springframework.integration.adapter' module to the new 'org.springframework.integration.ftp' module.

Browse Source
This commit is contained in:
Mark Fisher
2008-09-22 14:25:08 +00:00
parent 5260e8b1ac
commit a97b767d56
33 changed files with 363 additions and 483 deletions
Download Patch File Download Diff File Expand all files Collapse all files

57
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd
View File

@@ -16,55 +16,6 @@
]]></xsd:documentation>
</xsd:annotation>
<xsd:element name="file-source">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a file-based source channel adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="directory" type="xsd:string" use="required"/>
<xsd:attribute name="type" type="fileSourceType" use="optional"/>
<xsd:attribute name="message-creator" type="xsd:string" use="optional"/>
<xsd:attribute name="file-filter" type="xsd:string"/>
<xsd:attribute name="filename-filter" type="xsd:string"/>
<xsd:attribute name="filename-pattern" type="xsd:string"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="file-target">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines a file-based target.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="directory" type="xsd:string" use="required"/>
<xsd:attribute name="name-generator" type="xsd:string" use="optional"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="ftp-source">
<xsd:complexType>
<xsd:annotation>
<xsd:documentation>
Defines an ftp-receiving target channel adapter.
</xsd:documentation>
</xsd:annotation>
<xsd:attribute name="id" type="xsd:string"/>
<xsd:attribute name="username" type="xsd:string" use="optional"/>
<xsd:attribute name="password" type="xsd:string" use="optional"/>
<xsd:attribute name="host" type="xsd:string" use="required"/>
<xsd:attribute name="port" type="xsd:int" use="optional"/>
<xsd:attribute name="local-working-directory" type="xsd:string" use="required"/>
<xsd:attribute name="remote-working-directory" type="xsd:string" use="optional"/>
<xsd:attribute name="type" type="fileSourceType" use="optional"/>
<xsd:attribute name="message-creator" type="xsd:string" use="optional"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="mail-target">
<xsd:complexType>
<xsd:annotation>
@@ -108,12 +59,4 @@
</xsd:complexType>
</xsd:element>
<xsd:simpleType name="fileSourceType">
<xsd:restriction base="xsd:string">
<xsd:enumeration value="text"/>
<xsd:enumeration value="binary"/>
<xsd:enumeration value="file"/>
</xsd:restriction>
</xsd:simpleType>
</xsd:schema>

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

@@ -1,137 +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.adapter.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.message.Message;
import org.springframework.integration.message.MessageCreator;
import org.springframework.integration.message.MessageDeliveryAware;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.PollableSource;
import org.springframework.util.Assert;
/**
* Base class for implementing a PollableSource that creates messages from files
* in a directory, either local or remote.
*
* @author Marius Bogoevici
* @author Iwein Fuld
*/
public abstract class AbstractDirectorySource<T> implements PollableSource<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;
private final MessageCreator<T, T> messageCreator;
public AbstractDirectorySource(MessageCreator<T, T> messageCreator) {
this(messageCreator, null);
}
public AbstractDirectorySource(MessageCreator<T, T> messageCreator, Comparator<FileSnapshot> comparator) {
this.backlog = comparator == null ? new Backlog<FileSnapshot>() : new Backlog<FileSnapshot>(comparator);
Assert.notNull(messageCreator, "The MessageCreator must not be null");
this.messageCreator = messageCreator;
}
protected Backlog<FileSnapshot> getBacklog() {
return this.backlog;
}
public MessageCreator<T, T> getMessageCreator() {
return this.messageCreator;
}
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 (like
* FileSource). In cases where retrieval could be expensive because of
* network latency, a batched approach could be implemented here. See
* FtpSource for an example.
*
* @return the next message containing (part of) the unprocessed content of
* the directory
* @throws IOException
*/
protected Message<T> buildNextMessage() throws IOException {
return this.messageCreator.createMessage(retrieveNextPayload());
}
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.
*
* @param snapshot
* @throws IOException
*/
protected abstract void populateSnapshot(List<FileSnapshot> snapshot) throws IOException;
/**
* Returns the next file, based on the backlog data.
*
* @return
* @throws IOException
*/
protected abstract T retrieveNextPayload() throws IOException;
protected final void filesProcessed() {
this.backlog.processed();
}
}

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

@@ -1,228 +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.adapter.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 Mark Fisher
* @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();
}
}

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

@@ -1,36 +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.adapter.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;
}

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

@@ -1,43 +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.adapter.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);
}

101
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/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.adapter.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<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());
}
}

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

@@ -1,122 +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.adapter.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.integration.message.DefaultMessageCreator;
import org.springframework.integration.message.MessageCreator;
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(new DefaultMessageCreator<List<File>>(), clientPool);
}
public FtpSource(MessageCreator<List<File>, List<File>> messageCreator, FTPClientPool clientPool) {
super(messageCreator);
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) {
// some awkwardness here because the 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);
}
}
}

76
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/FtpTarget.java
View File

@@ -1,76 +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.adapter.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.message.Message;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageMapper;
import org.springframework.integration.message.MessageTarget;
import org.springframework.util.Assert;
/**
* Target adapter for sending files to an FTP server.
*
* @author Iwein Fuld
*/
public class FtpTarget implements MessageTarget {
private final MessageMapper<?, File> messageMapper;
private final FTPClientPool ftpClientPool;
public FtpTarget(MessageMapper<?, File> messageMapper, FTPClientPool ftpClientPool) {
Assert.notNull(messageMapper, "messageMapper must not be null");
Assert.notNull(ftpClientPool, "ftpClientPool must not be null");
this.ftpClientPool = ftpClientPool;
this.messageMapper = messageMapper;
}
public boolean send(Message message) {
boolean sent = false;
File file = this.messageMapper.mapMessage(message);
if (file != null && file.exists()) {
FTPClient client = null;
try {
FileInputStream fileInputStream = new FileInputStream(file);
client = this.ftpClientPool.getClient();
sent = client.storeFile(file.getName(), fileInputStream);
fileInputStream.close();
}
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);
}
}
return sent;
}
}

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

@@ -1,195 +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.adapter.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.message.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<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 DefaultFactory();
private final Log log = LogFactory.getLog(this.getClass());
private volatile String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY;
// setters
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;
}
public QueuedFTPClientPool() {
this(DEFAULT_POOL_SIZE);
}
/**
* @param maxPoolSize the maximum size of the pool
*/
public QueuedFTPClientPool(int maxPoolSize) {
pool = new ArrayBlockingQueue<FTPClient>(maxPoolSize);
}
/**
* Returns an active FTPClient connected to the configured server. When no
* clients are available in the queue a new client is created with the
* factory.
*
* 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 DefaultFactory 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;
}
}
}

45
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/AbstractDirectorySourceParser.java
View File

@@ -1,45 +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.adapter.ftp.config;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Base class for directory-based sources.
*
* @author Marius Bogoevici
*/
public abstract class AbstractDirectorySourceParser extends AbstractSimpleBeanDefinitionParser {
public static final String MESSAGE_CREATOR_REFERENCE_ATTRIBUTE = "message-creator";
@Override
protected boolean isEligibleAttribute(String attributeName) {
return !MESSAGE_CREATOR_REFERENCE_ATTRIBUTE.equals(attributeName) && super.isEligibleAttribute(attributeName);
}
@Override
protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) {
String messageCreatorReference = element.getAttribute(MESSAGE_CREATOR_REFERENCE_ATTRIBUTE);
if (StringUtils.hasText(messageCreatorReference)) {
beanDefinition.addConstructorArgReference(messageCreatorReference);
}
}
}

77
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceParser.java
View File

@@ -1,77 +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.adapter.ftp.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.integration.adapter.ftp.FtpSource;
import org.springframework.integration.adapter.ftp.QueuedFTPClientPool;
/**
* Parser for the &lt;ftp-source/&gt; element.
*
* @author Mark Fisher
* @author Marius Bogoevici
* @author Iwein Fuld
*/
public class FtpSourceParser extends AbstractDirectorySourceParser {
private static final String POOL_ATTRIBUTE_USER = "username";
private static final String POOL_ATTRIBUTE_PASS = "password";
private static final String POOL_ATTRIBUTE_HOST = "host";
private static final String POOL_ATTRIBUTE_PORT = "port";
private static final String POOL_ATTRIBUTE_REMOTEDIR = "remote-working-directory";
@Override
protected Class<?> getBeanClass(Element element) {
return FtpSource.class;
}
@Override
protected boolean isEligibleAttribute(String attributeName) {
return !POOL_ATTRIBUTE_HOST.equals(attributeName)
&& !POOL_ATTRIBUTE_PASS.equals(attributeName)
&& !POOL_ATTRIBUTE_PORT.equals(attributeName)
&& !POOL_ATTRIBUTE_USER.equals(attributeName)
&& !POOL_ATTRIBUTE_REMOTEDIR.equals(attributeName)
&& super.isEligibleAttribute(attributeName);
}
@Override
protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) {
super.postProcess(beanDefinition, element);
String user = element.getAttribute(POOL_ATTRIBUTE_USER);
String pass = element.getAttribute(POOL_ATTRIBUTE_PASS);
String host = element.getAttribute(POOL_ATTRIBUTE_HOST);
String port = element.getAttribute(POOL_ATTRIBUTE_PORT);
String remoteWorkingDirectory = element.getAttribute(POOL_ATTRIBUTE_REMOTEDIR);
QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool();
queuedFTPClientPool.setUsername(user);
queuedFTPClientPool.setPassword(pass);
queuedFTPClientPool.setHost(host);
queuedFTPClientPool.setPort(Integer.parseInt(port));
queuedFTPClientPool.setRemoteWorkingDirectory(remoteWorkingDirectory);
beanDefinition.addConstructorArgValue(queuedFTPClientPool);
}
}

73
org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/ftp/config/FtpTargetParser.java
View File

@@ -1,73 +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.adapter.ftp.config;
import java.io.File;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.integration.adapter.ftp.FtpTarget;
import org.springframework.integration.adapter.ftp.QueuedFTPClientPool;
import org.springframework.integration.message.DefaultMessageMapper;
import org.w3c.dom.Element;
public class FtpTargetParser extends AbstractSimpleBeanDefinitionParser {
private static final String POOL_ATTRIBUTE_USER = "username";
private static final String POOL_ATTRIBUTE_PASS = "password";
private static final String POOL_ATTRIBUTE_HOST = "host";
private static final String POOL_ATTRIBUTE_PORT = "port";
private static final String POOL_ATTRIBUTE_REMOTEDIR = "remote-working-directory";
@Override
protected Class<FtpTarget> getBeanClass(Element element) {
return FtpTarget.class;
}
@Override
protected boolean isEligibleAttribute(String attributeName) {
return !POOL_ATTRIBUTE_HOST.equals(attributeName) && !POOL_ATTRIBUTE_PASS.equals(attributeName)
&& !POOL_ATTRIBUTE_PORT.equals(attributeName) && !POOL_ATTRIBUTE_USER.equals(attributeName)
&& !POOL_ATTRIBUTE_REMOTEDIR.equals(attributeName) && super.isEligibleAttribute(attributeName);
}
@Override
protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) {
super.postProcess(beanDefinition, element);
QueuedFTPClientPool queuedFTPClientPool = constructFTPClientPool(element);
beanDefinition.addConstructorArgValue(queuedFTPClientPool);
beanDefinition.addConstructorArgValue(new DefaultMessageMapper<File>());
}
private QueuedFTPClientPool constructFTPClientPool(Element element) {
String user = element.getAttribute(POOL_ATTRIBUTE_USER);
String pass = element.getAttribute(POOL_ATTRIBUTE_PASS);
String host = element.getAttribute(POOL_ATTRIBUTE_HOST);
String port = element.getAttribute(POOL_ATTRIBUTE_PORT);
String remoteWorkingDirectory = element.getAttribute(POOL_ATTRIBUTE_REMOTEDIR);
QueuedFTPClientPool queuedFTPClientPool = new QueuedFTPClientPool();
queuedFTPClientPool.setUsername(user);
queuedFTPClientPool.setPassword(pass);
queuedFTPClientPool.setHost(host);
queuedFTPClientPool.setPort(Integer.parseInt(port));
queuedFTPClientPool.setRemoteWorkingDirectory(remoteWorkingDirectory);
return queuedFTPClientPool;
}
}

1
org.springframework.integration.adapter/src/main/resources/META-INF/spring-integration.parsers
View File

@@ -1,4 +1,3 @@
ftp-source=org.springframework.integration.adapter.ftp.config.FtpSourceParser
mail-target=org.springframework.integration.adapter.mail.config.MailTargetParser
polling-mail-source=org.springframework.integration.adapter.mail.config.PollingMailSourceParser
imap-idle-mail-source=org.springframework.integration.adapter.mail.config.SubscribableImapIdleMailSourceParser

168
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/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.adapter.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.adapter.ftp.Backlog;
import org.springframework.integration.adapter.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));
}
}

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

@@ -1,189 +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.adapter.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;
@SuppressWarnings("unchecked")
public class ConcurrentBacklogTests {
@Test(timeout = 1000)
public void simultaniousPreparation() 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;
}
}

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

@@ -1,282 +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.adapter.ftp;
import static org.easymock.EasyMock.eq;
import static org.easymock.EasyMock.expect;
import static org.easymock.EasyMock.getCurrentArguments;
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.easymock.IAnswer;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageCreator;
/**
* @author Iwein Fuld
*/
@SuppressWarnings("unchecked")
public class FtpSourceTests {
private MessageCreator<List<File>, List<File>> messageCreator = createMock(MessageCreator.class);
private FTPClient ftpClient = createMock(FTPClient.class);
private FTPFile ftpFile = createMock(FTPFile.class);
private FTPClientPool ftpClientPool = createNiceMock(FTPClientPool.class);
@Before
public void liberalPool() throws Exception {
expect(ftpClientPool.getClient()).andReturn(ftpClient).anyTimes();
}
private Object[] globalMocks = new Object[] { messageCreator, ftpClient, ftpFile, ftpClientPool };
private FtpSource ftpSource;
private Long size = 100l;
@Before
public void initializeFtpSource() {
ftpSource = new FtpSource(messageCreator, 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);
// create message
expect(messageCreator.createMessage(isA(List.class))).andReturn(
new GenericMessage(Arrays.asList(new File("test1"))));
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);
// create message
List<File> files = Arrays.asList(new File("test1"), new File("test2"));
expect(messageCreator.createMessage(isA(List.class))).andReturn(new GenericMessage(files));
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
// expect(ftpClient.isConnected()).andReturn(true);
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);
// create message
List<File> files = Arrays.asList(new File("test1"), new File("test2"));
expect(messageCreator.createMessage(isA(List.class))).andReturn(new GenericMessage(files)).times(2);
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);
// create message
expect(messageCreator.createMessage(isA(List.class))).andAnswer(new IAnswer<Message<List<File>>>() {
public Message<List<File>> answer() throws Throwable {
return new GenericMessage(getCurrentArguments()[0]);
}
}).times(2);
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);
// create message
expect(messageCreator.createMessage(isA(List.class))).andAnswer(new IAnswer<Message<List<File>>>() {
public Message<List<File>> answer() throws Throwable {
return new GenericMessage(getCurrentArguments()[0]);
}
}).times(3);
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);
// create message
expect(messageCreator.createMessage(isA(List.class))).andReturn(
new GenericMessage(Arrays.asList(new File("test1")))).times(2);
replay(globalMocks);
Message<List<File>> received = ftpSource.receive();
ftpSource.onFailure(received, new Exception("just a test"));
assertEquals(received, ftpSource.receive());
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();
}
}
}

95
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/FtpTargetTest.java
View File

@@ -1,95 +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.adapter.ftp;
import static org.easymock.classextension.EasyMock.*;
import static org.junit.Assert.*;
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.message.Message;
import org.springframework.integration.message.MessageMapper;
/**
*
* @author Iwein Fuld
*
*/
@SuppressWarnings("unchecked")
public class FtpTargetTest {
private FtpTarget ftpTarget;
// Mocks and initialization
private MessageMapper<?, File> messageMapper = createMock(MessageMapper.class);
private Message message = createMock(Message.class);
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);
@Before
public void liberalPool() throws Exception {
expect(ftpClientPool.getClient()).andReturn(ftpClient).anyTimes();
}
/*
* 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[] { messageMapper, message, ftpClient, ftpClientPool };
@Before
public void intitializeSubject() {
this.ftpTarget = new FtpTarget(messageMapper, ftpClientPool);
}
// Tests
@Test
public void send() throws Exception {
expect(messageMapper.mapMessage(message)).andReturn(File.createTempFile("test", ".tmp"));
expect(ftpClient.storeFile(isA(String.class), isA(FileInputStream.class))).andReturn(true);
replay(allMocks);
boolean sent = ftpTarget.send(message);
assertTrue(sent);
verify(allMocks);
}
@Test
public void sendFailed_negative() throws Exception {
expect(messageMapper.mapMessage(message)).andReturn(File.createTempFile("test", ".tmp"));
expect(ftpClient.storeFile(isA(String.class), isA(FileInputStream.class))).andReturn(false);
replay(allMocks);
boolean sent = ftpTarget.send(message);
assertFalse(sent);
verify(allMocks);
}
@Test
public void sendFailed_() throws Exception {
}
}

110
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/QueuedFTPClientPoolTest.java
View File

@@ -1,110 +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.adapter.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;
/**
* @author Iwein Fuld
*/
public class QueuedFTPClientPoolTest {
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;
}
}

34
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/CustomMessageCreator.java
View File

@@ -1,34 +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.adapter.ftp.config;
import java.io.File;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageCreator;
/**
* @author Marius Bogoevici
*/
public class CustomMessageCreator implements MessageCreator<File, String>{
public Message<String> createMessage(File object) {
return new GenericMessage<String> (object.getAbsolutePath());
}
}

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

@@ -1,86 +0,0 @@
package org.springframework.integration.adapter.ftp.config;
import static org.junit.Assert.assertTrue;
import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.adapter.ftp.FtpSource;
import org.springframework.integration.adapter.ftp.QueuedFTPClientPool;
import org.springframework.integration.channel.ChannelRegistry;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.config.MessageBusParser;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageCreator;
/**
* @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 FtpSource ftpSource;
private MessageCreator<List<File>, List<File>> messageCreator = new MessageCreator<List<File>, List<File>>() {
public Message<List<File>> createMessage(List<File> object) {
return new GenericMessage<List<File>>(object);
}
};
private static File localWorkDir;
@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(messageCreator, 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());
}
@Test public void withChannelAdapter() {
ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceWithChannelAdapter.xml", this.getClass());
ChannelRegistry channelRegistry = (ChannelRegistry) context.getBean(MessageBusParser.MESSAGE_BUS_BEAN_NAME);
PollableChannel input = (PollableChannel) channelRegistry.lookupChannel("output");
List<File> files = new ArrayList<File>();
files.add((File) input.receive().getPayload());
files.add((File) input.receive().getPayload());
assertTrue(files.containsAll(Arrays.asList(new File("file1"), new File("file2"))));
}
}

71
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceParserTests.java
View File

@@ -1,71 +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.adapter.ftp.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.File;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.adapter.ftp.FtpSource;
import org.springframework.integration.message.DefaultMessageCreator;
/**
* @author Mark Fisher
* @author Marius Bogoevici
* @author Iwein Fuld
*/
public class FtpSourceParserTests {
@Test
public void testFtpSourceAdapterParser() {
ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass());
FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceDefault");
DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(ftpSource);
DirectFieldAccessor accessor = new DirectFieldAccessor(sourceAccessor.getPropertyValue("clientPool"));
assertEquals("testHost", accessor.getPropertyValue("host"));
assertEquals(2121, accessor.getPropertyValue("port"));
assertEquals(new File("/local"), sourceAccessor.getPropertyValue("localWorkingDirectory"));
assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory"));
assertEquals("testUser", accessor.getPropertyValue("username"));
assertEquals("testPassword", accessor.getPropertyValue("password"));
Object messageCreator = sourceAccessor.getPropertyValue("messageCreator");
assertTrue(messageCreator instanceof DefaultMessageCreator);
}
@Test
public void testFtpSourceCustomType() {
ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceParserTests.xml", this.getClass());
FtpSource ftpSource = (FtpSource) context.getBean("ftpSourceCustom");
DirectFieldAccessor sourceAccessor = new DirectFieldAccessor(ftpSource);
DirectFieldAccessor accessor = new DirectFieldAccessor(sourceAccessor.getPropertyValue("clientPool"));
assertEquals("testHost", accessor.getPropertyValue("host"));
assertEquals(2121, accessor.getPropertyValue("port"));
assertEquals(new File("/local"), sourceAccessor.getPropertyValue("localWorkingDirectory"));
assertEquals("/remote", accessor.getPropertyValue("remoteWorkingDirectory"));
assertEquals("testUser", accessor.getPropertyValue("username"));
assertEquals("testPassword", accessor.getPropertyValue("password"));
Object messageCreator = sourceAccessor.getPropertyValue("messageCreator");
assertTrue(messageCreator instanceof CustomMessageCreator);
}
}

65
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/FtpTargetIntegrationTest.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.adapter.ftp.config;
import static org.junit.Assert.assertTrue;
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.adapter.ftp.FtpTarget;
import org.springframework.integration.adapter.ftp.QueuedFTPClientPool;
import org.springframework.integration.message.DefaultMessageMapper;
import org.springframework.integration.message.GenericMessage;
/**
* @author Iwein Fuld
*/
@Ignore
public class FtpTargetIntegrationTest {
private FtpTarget ftpTarget;
@Before
public void initFtpTarget() {
QueuedFTPClientPool clientPool = new QueuedFTPClientPool();
clientPool.setHost("localhost");
clientPool.setUsername("ftp-user");
clientPool.setPassword("kaas");
clientPool.setRemoteWorkingDirectory("ftp-test");
ftpTarget = new FtpTarget(new DefaultMessageMapper<File>(), clientPool);
}
@Test
public void send() throws Exception {
File file = File.createTempFile("test", "");
assertTrue(ftpTarget.send(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();
}
}
}

57
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceParserTests.xml
View File

@@ -1,57 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:si="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">
<si:ftp-source id="ftpSourceDefault"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"/>
<si:ftp-source id="ftpSourceText"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"
/>
<si:ftp-source id="ftpSourceBinary"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"
/>
<si:ftp-source id="ftpSourceFile"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"
/>
<si:ftp-source id="ftpSourceCustom"
host="testHost"
port="2121"
local-working-directory="/local"
remote-working-directory="/remote"
username="testUser"
password="testPassword"
message-creator="customMessageCreator"/>
<bean id="customMessageCreator"
class="org.springframework.integration.adapter.ftp.config.CustomMessageCreator"/>
</beans>

36
org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceWithChannelAdapter.xml
View File

@@ -1,36 +0,0 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:si="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:p="http://www.springframework.org/schema/p"
xsi:schemaLocation="
http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
<si:message-bus />
<bean id="ftpSource"
class="org.springframework.integration.adapter.ftp.FtpSource">
<constructor-arg>
<bean
class="org.springframework.integration.message.DefaultMessageCreator" />
</constructor-arg>
<constructor-arg>
<bean
class="org.springframework.integration.adapter.ftp.QueuedFTPClientPool"
p:host="localhost" p:password="kaas" p:username="ftp-user"
p:remoteWorkingDirectory="ftp-test" />
</constructor-arg>
</bean>
<si:channel-adapter id="input" source="ftpSource" />
<bean id="splitter"
class="org.springframework.integration.adapter.ftp.config.CollectionSplitter" />
<si:splitter output-channel="output" input-channel="input"
ref="splitter" method="split" />
<si:channel id="output" />
</beans>