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

Committing the initial version of the FTP source adapter (INT-153).

Browse Source
This commit is contained in:
Mark Fisher
2008-04-01 01:38:51 +00:00
parent c55756ab55
commit 6e845339ad
12 changed files with 686 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-adapters/.classpath
View File

@@ -9,7 +9,9 @@
<classpathentry kind="var" path="IVY_CACHE/javax.servlet/servlet-api/servlet-api-2.4.jar" sourcepath="/IVY_CACHE/javax.servlet/servlet-api/servlet-api-sources-2.4.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.aopalliance/aopalliance/aopalliance-1.0.jar" sourcepath="IVY_CACHE/org.aopalliance/aopalliance/aopalliance-sources-1.0.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.apache/commons-logging/commons-logging-1.1.jar" sourcepath="/IVY_CACHE/org.apache/commons-logging/commons-logging-sources-1.1.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.apache.commons/commons-net/commons-net-1.4.1.jar" sourcepath="/IVY_CACHE/org.apache.commons/commons-net/commons-net-sources-1.4.1.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.apache/commons-pool/commons-pool-1.3.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.apache.oro/oro/oro-2.0.8.jar" sourcepath="/IVY_CACHE/org.apache.oro/oro/oro-sources-2.0.8.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.junit/junit/junit-4.4.jar" sourcepath="/IVY_CACHE/org.junit/junit/junit-sources-4.4.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.springframework/spring-aop/spring-aop-2.5.2.jar" sourcepath="/IVY_CACHE/org.springframework/spring-aop/spring-aop-sources-2.5.2.jar"/>
<classpathentry kind="var" path="IVY_CACHE/org.springframework/spring-beans/spring-beans-2.5.2.jar" sourcepath="/IVY_CACHE/org.springframework/spring-beans/spring-beans-sources-2.5.2.jar"/>

2
spring-integration-adapters/ivy.xml
View File

@@ -24,6 +24,8 @@
<dependency org="javax.jms" name="jms" rev="1.1" conf="compile->default"/>
<dependency org="javax.mail" name="mail" rev="1.4" conf="compile->default"/>
<dependency org="javax.servlet" name="servlet-api" rev="2.4" conf="compile->default"/>
<dependency org="org.apache.commons" name="commons-net" rev="1.4.1" conf="compile->default"/>
<dependency org="org.apache.oro" name="oro" rev="2.0.8" conf="compile->default"/>
<dependency org="org.junit" name="junit" rev="4.4" conf="test->default"/>
<dependency org="org.springframework.integration" name="spring-integration-core" rev="latest.integration" conf="compile"/>
<dependency org="org.springframework" name="spring-context" rev="2.5.2" conf="compile->default"/>

11
spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers
View File

@@ -1,9 +1,10 @@
file-source=org.springframework.integration.adapter.file.config.FileSourceAdapterParser
file-target=org.springframework.integration.adapter.file.config.FileTargetAdapterParser
jms-source=org.springframework.integration.adapter.jms.config.JmsSourceAdapterParser
jms-target=org.springframework.integration.adapter.jms.config.JmsTargetAdapterParser
rmi-source=org.springframework.integration.adapter.rmi.config.RmiSourceAdapterParser
rmi-target=org.springframework.integration.adapter.rmi.config.RmiTargetAdapterParser
ftp-source=org.springframework.integration.adapter.ftp.config.FtpSourceAdapterParser
httpinvoker-source=org.springframework.integration.adapter.httpinvoker.config.HttpInvokerSourceAdapterParser
httpinvoker-target=org.springframework.integration.adapter.httpinvoker.config.HttpInvokerTargetAdapterParser
mail-target=org.springframework.integration.adapter.mail.config.MailTargetAdapterParser
jms-source=org.springframework.integration.adapter.jms.config.JmsSourceAdapterParser
jms-target=org.springframework.integration.adapter.jms.config.JmsTargetAdapterParser
mail-target=org.springframework.integration.adapter.mail.config.MailTargetAdapterParser
rmi-source=org.springframework.integration.adapter.rmi.config.RmiSourceAdapterParser
rmi-target=org.springframework.integration.adapter.rmi.config.RmiTargetAdapterParser

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

@@ -43,6 +43,26 @@
</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="channel" type="xsd:string" use="required"/>
<xsd:attribute name="period" type="xsd:long" use="required"/>
<xsd:attribute name="text-based" type="xsd:boolean" use="optional"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="jms-source">
<xsd:complexType>
<xsd:annotation>

63
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java Normal file
View File

@@ -0,0 +1,63 @@
/*
* 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.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
/**
* Tracks changes in the context. This implementation is thread-safe as it
* allows to synchronously process a new directory structure.
*
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class DirectoryContentManager {
private Map<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
private final Map<String, FileInfo> backlog = new HashMap<String, FileInfo>();
public synchronized void processSnapshot(Map<String, FileInfo> remoteSnapshot) {
Iterator<Map.Entry<String, FileInfo>> iter = this.backlog.entrySet().iterator();
while (iter.hasNext()) {
String fileName = iter.next().getKey();
if (!remoteSnapshot.containsKey(fileName)) {
iter.remove();
}
}
for (String fileName : remoteSnapshot.keySet()) {
if (!this.snapshot.containsKey(fileName)
|| (!this.snapshot.get(fileName).equals(remoteSnapshot.get(fileName)))) {
this.backlog.put(fileName, remoteSnapshot.get(fileName));
}
}
this.snapshot = new HashMap<String, FileInfo>(remoteSnapshot);
}
public synchronized void fileProcessed(String fileName) {
this.backlog.remove(fileName);
}
public Map<String, FileInfo> getBacklog() {
return Collections.unmodifiableMap(this.backlog);
}
}

69
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FileInfo.java Normal file
View File

@@ -0,0 +1,69 @@
/*
* 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;
/**
* Information about a file in a directory.
*
* @author Marius Bogoevici
*/
public class FileInfo {
private final String fileName;
private final long modificationTimestamp;
private final long size;
public FileInfo(String fileName, long modificationTimestamp, long size) {
this.fileName = fileName;
this.modificationTimestamp = modificationTimestamp;
this.size = size;
}
public String getFileName() {
return fileName;
}
public long getModificationTimestamp() {
return modificationTimestamp;
}
public long getSize() {
return size;
}
@Override
public boolean equals(Object other) {
if (other == null || !(other instanceof FileInfo)) {
return false;
}
FileInfo otherInfo = (FileInfo) other;
return this.getSize() == otherInfo.getSize()
&& this.getModificationTimestamp() == otherInfo.getModificationTimestamp()
&& this.fileName.equals(otherInfo.getFileName());
}
@Override
public int hashCode() {
return (fileName == null ? 0 : fileName.hashCode()) ^ new Long(modificationTimestamp).hashCode()
^ new Long(size).hashCode();
}
}

185
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java Normal file
View File

@@ -0,0 +1,185 @@
/*
* 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.Collection;
import java.util.HashMap;
import java.util.LinkedList;
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.FTPFile;
import org.springframework.integration.adapter.PollableSource;
import org.springframework.integration.adapter.PollingSourceAdapter;
import org.springframework.integration.adapter.file.ByteArrayFileMapper;
import org.springframework.integration.adapter.file.TextFileMapper;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
/**
* A source adapter for receiving files via FTP.
*
* @author Marius Bogoevici
* @author Mark Fisher
*/
public class FtpSourceAdapter extends PollingSourceAdapter<File> implements PollableSource<File> {
private final static String DEFAULT_HOST = "localhost";
private final static int DEFAULT_PORT = 21;
private final static String DEFAULT_REMOTE_WORKING_DIRECTORY = "/";
private final Log logger = LogFactory.getLog(this.getClass());
private volatile String username;
private volatile String password;
private volatile String host = DEFAULT_HOST;
private volatile int port = DEFAULT_PORT;
private volatile String remoteWorkingDirectory = DEFAULT_REMOTE_WORKING_DIRECTORY;
private volatile File localWorkingDirectory;
private volatile boolean textBased = true;
private final DirectoryContentManager directoryContentManager = new DirectoryContentManager();
private final FTPClient client = new FTPClient();
public void setHost(String host) {
this.host = host;
}
public void setPort(int port) {
this.port = port;
}
public void setUsername(String username) {
this.username = username;
}
public void setPassword(String password) {
this.password = password;
}
public void setRemoteWorkingDirectory(String remoteWorkingDirectory) {
Assert.hasText(remoteWorkingDirectory, "'remoteWorkingDirectory' is required");
this.remoteWorkingDirectory = remoteWorkingDirectory;
}
public void setLocalWorkingDirectory(File localWorkingDirectory) {
Assert.notNull(localWorkingDirectory, "'localWorkingDirectory' must not be null");
this.localWorkingDirectory = localWorkingDirectory;
}
public boolean isTextBased() {
return textBased;
}
public void setTextBased(boolean textBased) {
this.textBased = textBased;
}
@Override
protected void initialize() {
this.setSource(this);
if (this.isTextBased()) {
this.setMessageMapper(new TextFileMapper(this.localWorkingDirectory));
}
else {
this.setMessageMapper(new ByteArrayFileMapper(this.localWorkingDirectory));
}
}
@Override
protected void onSend(File file) {
this.directoryContentManager.fileProcessed(file.getName());
}
public final Collection<File> poll(int limit) {
try {
LinkedList<File> localFileList = new LinkedList<File>();
this.client.connect(this.host, this.port);
if (!StringUtils.hasText(this.username)) {
throw new MessageHandlingException("username is required");
}
if (!this.client.login(this.username, this.password)) {
throw new MessageHandlingException("Login failed. Please check the username and password.");
}
if (logger.isDebugEnabled()) {
logger.debug("login successful");
}
this.client.setFileType(FTP.IMAGE_FILE_TYPE);
if (!this.remoteWorkingDirectory.equals(this.client.printWorkingDirectory())
&& !this.client.changeWorkingDirectory(this.remoteWorkingDirectory)) {
throw new MessageHandlingException("Could not change directory to '" +
remoteWorkingDirectory + "'. Please check the path.");
}
if (logger.isDebugEnabled()) {
logger.debug("working directory is: " + this.client.printWorkingDirectory());
}
FTPFile[] fileList = this.client.listFiles();
HashMap<String, FileInfo> snapshot = new HashMap<String, FileInfo>();
for (FTPFile ftpFile : fileList) {
FileInfo fileInfo = new FileInfo(ftpFile.getName(), ftpFile.getTimestamp().getTimeInMillis(),
ftpFile.getSize());
snapshot.put(ftpFile.getName(), fileInfo);
}
this.directoryContentManager.processSnapshot(snapshot);
for (String fileName : this.directoryContentManager.getBacklog().keySet()) {
File file = new File(this.localWorkingDirectory, fileName);
if (file.exists()) {
file.delete();
}
FileOutputStream fileOutputStream = new FileOutputStream(file);
this.client.retrieveFile(fileName, fileOutputStream);
fileOutputStream.close();
localFileList.add(file);
if (limit >= localFileList.size()) {
break;
}
}
return localFileList;
}
catch (Exception e) {
try {
if (this.client.isConnected()) {
this.client.disconnect();
}
}
catch (IOException ioe) {
throw new MessageHandlingException("Error when disconnecting from ftp.", ioe);
}
throw new MessageHandlingException("Error while polling for messages.", e);
}
}
}

62
spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParser.java Normal file
View File

@@ -0,0 +1,62 @@
/*
* 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.beans.factory.xml.AbstractSimpleBeanDefinitionParser;
import org.springframework.core.Conventions;
import org.springframework.integration.adapter.ftp.FtpSourceAdapter;
import org.springframework.util.StringUtils;
/**
* Parser for the &lt;ftp-source/&gt; element.
*
* @author Marius Bogoevici
*/
public class FtpSourceAdapterParser extends AbstractSimpleBeanDefinitionParser {
private static final String CHANNEL_ATTRIBUTE = "channel";
protected Class<?> getBeanClass(Element element) {
return FtpSourceAdapter.class;
}
protected boolean shouldGenerateId() {
return false;
}
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected boolean isEligibleAttribute(String attributeName) {
return !CHANNEL_ATTRIBUTE.equals(attributeName) && super.isEligibleAttribute(attributeName);
}
protected void postProcess(BeanDefinitionBuilder beanDefinition, Element element) {
String channelRef = element.getAttribute(CHANNEL_ATTRIBUTE);
if (StringUtils.hasText(channelRef)) {
beanDefinition.addPropertyReference(
Conventions.attributeNameToPropertyName(CHANNEL_ATTRIBUTE), channelRef);
}
}
}

177
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/DirectoryContentManagerTests.java Normal file
View File

@@ -0,0 +1,177 @@
/*
* 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.HashMap;
import java.util.Map;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
/**
* @author Marius Bogoevici
*/
public class DirectoryContentManagerTests {
private DirectoryContentManager directoryContentManager;
@Before
public void setUp() {
directoryContentManager = new DirectoryContentManager();
}
@Test
public void testInitialization() {
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertEquals(3, directoryContentManager.getBacklog().size());
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("a.txt"));
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("b.txt"));
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("c.txt"));
}
@Test
public void testFullProcessingInOneStep() {
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
directoryContentManager.processSnapshot(remoteSnapshot);
directoryContentManager.fileProcessed("a.txt");
directoryContentManager.fileProcessed("b.txt");
directoryContentManager.fileProcessed("c.txt");
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
}
@Test
public void testFullProcessingInTwoSteps() {
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
directoryContentManager.processSnapshot(remoteSnapshot);
directoryContentManager.fileProcessed("a.txt");
directoryContentManager.fileProcessed("b.txt");
Assert.assertEquals(1, directoryContentManager.getBacklog().size());
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("c.txt"));
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, directoryContentManager.getBacklog().size());
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("c.txt"));
directoryContentManager.fileProcessed("c.txt");
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
}
@Test
public void testOneFileChangedSize() {
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
directoryContentManager.processSnapshot(remoteSnapshot);
directoryContentManager.fileProcessed("a.txt");
directoryContentManager.fileProcessed("b.txt");
Assert.assertEquals(1, directoryContentManager.getBacklog().size());
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("c.txt"));
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, directoryContentManager.getBacklog().size());
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("c.txt"));
directoryContentManager.fileProcessed("c.txt");
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
directoryContentManager.processSnapshot(remoteSnapshot);
remoteSnapshot.put("c.txt", new FileInfo("c.txt", 1001, 112));
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, directoryContentManager.getBacklog().size());
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("c.txt"));
}
@Test
public void testOneFileChangedDate() {
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
directoryContentManager.processSnapshot(remoteSnapshot);
directoryContentManager.fileProcessed("a.txt");
directoryContentManager.fileProcessed("b.txt");
Assert.assertEquals(1, directoryContentManager.getBacklog().size());
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("c.txt"));
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, directoryContentManager.getBacklog().size());
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("c.txt"));
directoryContentManager.fileProcessed("c.txt");
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
directoryContentManager.processSnapshot(remoteSnapshot);
remoteSnapshot.put("c.txt", new FileInfo("c.txt", 1011, 102));
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, directoryContentManager.getBacklog().size());
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("c.txt"));
}
@Test
public void testOneFileAdded() {
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
directoryContentManager.processSnapshot(remoteSnapshot);
directoryContentManager.fileProcessed("a.txt");
directoryContentManager.fileProcessed("b.txt");
directoryContentManager.fileProcessed("c.txt");
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
directoryContentManager.processSnapshot(remoteSnapshot);
remoteSnapshot.put("d.txt", new FileInfo("d.txt", 1003, 103));
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertEquals(1, directoryContentManager.getBacklog().size());
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("d.txt"));
}
@Test
public void testOneFileRemoved() {
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
directoryContentManager.processSnapshot(remoteSnapshot);
directoryContentManager.fileProcessed("a.txt");
directoryContentManager.fileProcessed("b.txt");
directoryContentManager.fileProcessed("c.txt");
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
directoryContentManager.processSnapshot(remoteSnapshot);
remoteSnapshot.remove("c.txt");
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
}
@Test
public void testOneFileRemovedBeforeBeingProcessedInTheNextStep() {
Assert.assertTrue(directoryContentManager.getBacklog().isEmpty());
Map<String, FileInfo> remoteSnapshot = generateInitialSnapshot();
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertTrue(directoryContentManager.getBacklog().containsKey("c.txt"));
remoteSnapshot.remove("c.txt");
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertEquals(2, directoryContentManager.getBacklog().size());
directoryContentManager.processSnapshot(remoteSnapshot);
Assert.assertEquals(2, directoryContentManager.getBacklog().size());
}
private static Map<String, FileInfo> generateInitialSnapshot() {
Map<String, FileInfo> remoteSnapshot = new HashMap<String, FileInfo>();
remoteSnapshot.put("a.txt", new FileInfo("a.txt", 1000, 100));
remoteSnapshot.put("b.txt", new FileInfo("b.txt", 1001, 101));
remoteSnapshot.put("c.txt", new FileInfo("c.txt", 1002, 102));
return remoteSnapshot;
}
}

43
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParserTests.java Normal file
View File

@@ -0,0 +1,43 @@
/*
* 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 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.FtpSourceAdapter;
import org.springframework.integration.scheduling.PollingSchedule;
/**
* @author Marius Bogoevici
*/
public class FtpSourceAdapterParserTests {
@Test
public void testFtpSourceAdapterParser() {
ApplicationContext context = new ClassPathXmlApplicationContext("ftpSourceAdapterParserTests.xml", this.getClass());
FtpSourceAdapter ftpAdapter = (FtpSourceAdapter) context.getBean("ftpAdapter");
DirectFieldAccessor ftpPollingAdapterAccessor = new DirectFieldAccessor(ftpAdapter);
assertEquals(context.getBean("testChannel"), ftpPollingAdapterAccessor.getPropertyValue("channel"));
assertEquals(12345L, ((PollingSchedule) ftpPollingAdapterAccessor.getPropertyValue("schedule")).getPeriod());
}
}

22
spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceAdapterParserTests.xml Normal file
View File

@@ -0,0 +1,22 @@
<?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:message-bus/>
<si:channel id="testChannel"/>
<si:ftp-source id="ftpAdapter" channel="testChannel"
period="12345"
host="localhost"
port="2121"
local-working-directory="${java.io.tmpdir}/spring-integration-samples/input"
remote-working-directory="/remote"
username="myUser"
password="myPassword"/>
</beans>

40
spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java
View File

@@ -20,6 +20,7 @@ import java.util.Collection;
import java.util.concurrent.Executors;
import org.springframework.context.Lifecycle;
import org.springframework.integration.MessagingConfigurationException;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageMapper;
@@ -40,25 +41,38 @@ import org.springframework.util.Assert;
*/
public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements MessagingTaskSchedulerAware, Lifecycle {
private PollableSource<T> source;
private volatile PollableSource<T> source;
private PollingSchedule schedule = new PollingSchedule(1000);
private volatile PollingSchedule schedule = new PollingSchedule(1000);
private MessagingTaskScheduler scheduler;
private volatile MessagingTaskScheduler scheduler;
private int maxMessagesPerTask = 1;
private volatile int maxMessagesPerTask = 1;
private volatile boolean starting;
private volatile boolean running;
/**
* Create a new adapter for the given source.
*/
public PollingSourceAdapter(PollableSource<T> source) {
this.setSource(source);
}
/**
* No-arg constructor for providing source after construction.
*/
public PollingSourceAdapter() {
}
public void setSource(PollableSource<T> source) {
Assert.notNull(source, "'source' must not be null");
this.source = source;
}
public void setInitialDelay(long intialDelay) {
Assert.isTrue(intialDelay >= 0, "'intialDelay' must not be negative");
this.schedule.setInitialDelay(intialDelay);
@@ -86,6 +100,13 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
return this.running;
}
@Override
protected void initialize() {
if (this.source == null) {
throw new MessagingConfigurationException("source must not be null");
}
}
public void start() {
if (this.isRunning() || this.starting) {
return;
@@ -129,12 +150,21 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
for (T next : results) {
if (this.sendToChannel(next)) {
messagesProcessed++;
this.onSend(next);
}
}
}
return messagesProcessed;
}
/**
* Callback method invoked after an item is sent to the channel.
* <p>
* Subclasses may override. The default implementation does nothing.
*/
protected void onSend(T sentItem) {
}
private class PollingSourceAdapterTask implements MessagingTask {