diff --git a/spring-integration-adapters/.classpath b/spring-integration-adapters/.classpath index 1f743c678d..38dd0ea8e0 100644 --- a/spring-integration-adapters/.classpath +++ b/spring-integration-adapters/.classpath @@ -9,7 +9,9 @@ + + diff --git a/spring-integration-adapters/ivy.xml b/spring-integration-adapters/ivy.xml index 902635a383..befea0f5bb 100644 --- a/spring-integration-adapters/ivy.xml +++ b/spring-integration-adapters/ivy.xml @@ -24,6 +24,8 @@ + + diff --git a/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers b/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers index bb6b29b210..40bc9b3c16 100644 --- a/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers +++ b/spring-integration-adapters/src/main/java/META-INF/spring-integration.parsers @@ -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 \ No newline at end of file +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 \ No newline at end of file diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd index 7f09130a6e..8bb81bd4eb 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/config/spring-integration-adapters-1.0.xsd @@ -43,6 +43,26 @@ + + + + + Defines an ftp-receiving target channel adapter. + + + + + + + + + + + + + + + diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java new file mode 100644 index 0000000000..52e30c951a --- /dev/null +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/DirectoryContentManager.java @@ -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 snapshot = new HashMap(); + + private final Map backlog = new HashMap(); + + + public synchronized void processSnapshot(Map remoteSnapshot) { + Iterator> 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(remoteSnapshot); + } + + public synchronized void fileProcessed(String fileName) { + this.backlog.remove(fileName); + } + + public Map getBacklog() { + return Collections.unmodifiableMap(this.backlog); + } + +} diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FileInfo.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FileInfo.java new file mode 100644 index 0000000000..3e7fd8de20 --- /dev/null +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FileInfo.java @@ -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(); + } + +} diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java new file mode 100644 index 0000000000..cde6a2f789 --- /dev/null +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/FtpSourceAdapter.java @@ -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 implements PollableSource { + + 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 poll(int limit) { + try { + LinkedList localFileList = new LinkedList(); + 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 snapshot = new HashMap(); + 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); + } + } + +} diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParser.java new file mode 100644 index 0000000000..c7cfe13c8b --- /dev/null +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParser.java @@ -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 <ftp-source/> 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); + } + } + +} diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/DirectoryContentManagerTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/DirectoryContentManagerTests.java new file mode 100644 index 0000000000..c3063ca24b --- /dev/null +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/DirectoryContentManagerTests.java @@ -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 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 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 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 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 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 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 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 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 generateInitialSnapshot() { + Map remoteSnapshot = new HashMap(); + 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; + } + +} diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParserTests.java new file mode 100644 index 0000000000..dfec16c294 --- /dev/null +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/FtpSourceAdapterParserTests.java @@ -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()); + } + +} diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceAdapterParserTests.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceAdapterParserTests.xml new file mode 100644 index 0000000000..beaae3e5a9 --- /dev/null +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/ftp/config/ftpSourceAdapterParserTests.xml @@ -0,0 +1,22 @@ + + + + + + + + + diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java index de1cfa8420..75d861e44c 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollingSourceAdapter.java @@ -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 extends AbstractSourceAdapter implements MessagingTaskSchedulerAware, Lifecycle { - private PollableSource source; + private volatile PollableSource 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 source) { + this.setSource(source); + } + + /** + * No-arg constructor for providing source after construction. + */ + public PollingSourceAdapter() { + } + + + public void setSource(PollableSource 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 extends AbstractSourceAdapter 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 extends AbstractSourceAdapter 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. + *

+ * Subclasses may override. The default implementation does nothing. + */ + protected void onSend(T sentItem) { + } + private class PollingSourceAdapterTask implements MessagingTask {