From 069e7672e319fbfa1d7a6f8b7b77e5185c43b02a Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Thu, 10 Apr 2008 21:40:41 +0000 Subject: [PATCH] PollableSource now returns a Message rather than an Object from poll(). Now PollableSource implementations may use a MessageMapper, but the "source adapter" no longer does. --- .../event/ApplicationEventSourceAdapter.java | 11 ++- .../adapter/file/AbstractFileMapper.java | 5 +- .../integration/adapter/file/FileSource.java | 77 --------------- .../adapter/file/FileSourceAdapter.java | 99 ++++++++++++++----- .../adapter/file/TextFileMapper.java | 2 +- .../file/config/FileSourceAdapterParser.java | 8 +- .../adapter/ftp/DirectoryContentManager.java | 18 +++- .../adapter/ftp/FtpSourceAdapter.java | 25 +++-- .../adapter/jms/JmsPollableSource.java | 6 +- .../adapter/stream/ByteStreamSource.java | 8 +- .../adapter/stream/CharacterStreamSource.java | 6 +- .../config/FileSourceAdapterParserTests.java | 50 ++++++++++ .../config/FileTargetAdapterParserTests.java | 2 +- .../config/fileSourceAdapterParserTests.xml | 21 ++++ .../config/fileTargetAdapterParserTests.xml | 5 + .../CharacterStreamSourceAdapterTests.java | 1 + .../adapter/AbstractSourceAdapter.java | 30 +----- .../adapter/MethodInvokingSource.java | 6 +- .../integration/adapter/PollableSource.java | 6 +- .../adapter/PollingSourceAdapter.java | 7 +- .../adapter/MethodInvokingSourceTests.java | 6 +- .../adapter/PollingSourceAdapterTests.java | 9 +- .../integration/bus/MessageBusTests.java | 3 +- .../dispatcher/SynchronousChannelTests.java | 6 +- 24 files changed, 242 insertions(+), 175 deletions(-) delete mode 100644 spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java create mode 100644 spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParserTests.java create mode 100644 spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileSourceAdapterParserTests.xml diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java index 78167c383a..11ca0f269e 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/event/ApplicationEventSourceAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -22,6 +22,7 @@ import java.util.List; import org.springframework.context.ApplicationEvent; import org.springframework.context.ApplicationListener; import org.springframework.integration.adapter.AbstractSourceAdapter; +import org.springframework.integration.message.GenericMessage; import org.springframework.util.Assert; import org.springframework.util.CollectionUtils; @@ -49,15 +50,19 @@ public class ApplicationEventSourceAdapter extends AbstractSourceAdapter eventType : this.eventTypes) { if (eventType.isAssignableFrom(event.getClass())) { - this.sendToChannel(event); + this.sendMessage(event); return; } } } + private void sendMessage(ApplicationEvent event) { + this.sendToChannel(new GenericMessage(event)); + } + } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/AbstractFileMapper.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/AbstractFileMapper.java index e04d836752..23b9435d6a 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/AbstractFileMapper.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/AbstractFileMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -77,7 +77,8 @@ public abstract class AbstractFileMapper extends AbstractMessageMapper message = new GenericMessage(this.getIdGenerator().generateId(), payload); + Message message = new GenericMessage(payload); + message.getHeader().setProperty(FileNameGenerator.FILENAME_PROPERTY_KEY, file.getName()); if (this.backupDirectory != null) { FileWriter writer = new FileWriter(this.backupDirectory.getAbsolutePath() + File.separator + file.getName()); diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java deleted file mode 100644 index 95add15912..0000000000 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java +++ /dev/null @@ -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.file; - -import java.io.File; -import java.io.FileFilter; -import java.io.FilenameFilter; - -import org.springframework.integration.adapter.PollableSource; -import org.springframework.integration.message.MessagingException; -import org.springframework.util.Assert; - -/** - * A messaging source that polls a directory to retrieve files. - * - * @author Mark Fisher - */ -public class FileSource implements PollableSource { - - private File directory; - - private FileFilter fileFilter; - - private FilenameFilter filenameFilter; - - - public FileSource(File directory) { - Assert.notNull("directory must not be null"); - this.directory = directory; - } - - public void setFileFilter(FileFilter fileFilter) { - this.fileFilter = fileFilter; - } - - public void setFilenameFilter(FilenameFilter filenameFilter) { - this.filenameFilter = filenameFilter; - } - - public File poll() { - File[] files = null; - if (this.fileFilter != null) { - files = this.directory.listFiles(this.fileFilter); - } - else if (this.filenameFilter != null) { - files = this.directory.listFiles(this.filenameFilter); - } - else { - files = this.directory.listFiles(); - } - if (files == null) { - throw new MessagingException("Problem occurred while polling for files. " + - "Is '" + directory.getAbsolutePath() + "' a directory?"); - } - for (int i = 0; i < files.length; i++) { - if (files[i].isFile()) { - return files[i]; - } - } - return null; - } - -} diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java index adf36b0827..4232a5b74d 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSourceAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -17,49 +17,96 @@ package org.springframework.integration.adapter.file; import java.io.File; +import java.io.FileFilter; +import java.io.FilenameFilter; +import org.springframework.integration.adapter.PollableSource; import org.springframework.integration.adapter.PollingSourceAdapter; -import org.springframework.integration.channel.MessageChannel; -import org.springframework.integration.message.MessageMapper; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessagingException; import org.springframework.util.Assert; /** - * Channel adapter for polling a directory and creating messages from its files. + * A messaging source that polls a directory to retrieve files. * * @author Mark Fisher */ -public class FileSourceAdapter extends PollingSourceAdapter { +public class FileSourceAdapter extends PollingSourceAdapter implements PollableSource { - public FileSourceAdapter(File directory, MessageChannel channel, int period) { - this(directory, channel, period, true); + private final File directory; + + private volatile boolean textBased = true; + + private volatile AbstractFileMapper mapper; + + private volatile FileNameGenerator fileNameGenerator; + + private volatile FileFilter fileFilter; + + private volatile FilenameFilter filenameFilter; + + + public FileSourceAdapter(File directory) { + Assert.notNull(directory, "directory must not be null"); + this.directory = directory; } - public FileSourceAdapter(File directory, MessageChannel channel, int period, boolean isTextBased) { - super(new FileSource(directory)); - this.setChannel(channel); - this.setPeriod(period); - if (isTextBased) { - this.setMessageMapper(new TextFileMapper(directory)); - } - else { - this.setMessageMapper(new ByteArrayFileMapper(directory)); - } + + public boolean isTextBased() { + return this.textBased; + } + + public void setTextBased(boolean textBased) { + this.textBased = textBased; + } + + public void setFileFilter(FileFilter fileFilter) { + this.fileFilter = fileFilter; + } + + public void setFilenameFilter(FilenameFilter filenameFilter) { + this.filenameFilter = filenameFilter; } public void setFileNameGenerator(FileNameGenerator fileNameGenerator) { - Assert.notNull(fileNameGenerator, "'fileNameGenerator' must not be null"); - MessageMapper mapper = this.getMessageMapper(); - if (mapper instanceof AbstractFileMapper) { - ((AbstractFileMapper) mapper).setFileNameGenerator(fileNameGenerator); + this.fileNameGenerator = fileNameGenerator; + } + + @Override + protected void initialize() { + this.setSource(this); + if (this.isTextBased()) { + this.mapper = new TextFileMapper(this.directory); + } + else { + this.mapper = new ByteArrayFileMapper(this.directory); + } + if (this.fileNameGenerator != null) { + this.mapper.setFileNameGenerator(this.fileNameGenerator); } } - public void setBackupDirectory(File backupDirectory) { - Assert.notNull(backupDirectory, "'backupDirectory' must not be null"); - MessageMapper mapper = this.getMessageMapper(); - if (mapper != null && (mapper instanceof AbstractFileMapper)) { - ((AbstractFileMapper) mapper).setBackupDirectory(backupDirectory); + public Message poll() { + File[] files = null; + if (this.fileFilter != null) { + files = this.directory.listFiles(this.fileFilter); } + else if (this.filenameFilter != null) { + files = this.directory.listFiles(this.filenameFilter); + } + else { + files = this.directory.listFiles(); + } + if (files == null) { + throw new MessagingException("Problem occurred while polling for files. " + + "Is '" + directory.getAbsolutePath() + "' a directory?"); + } + for (int i = 0; i < files.length; i++) { + if (files[i].isFile()) { + return this.mapper.toMessage(files[i]); + } + } + return null; } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/TextFileMapper.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/TextFileMapper.java index d1661f21f5..31e95e05cc 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/TextFileMapper.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/TextFileMapper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java index 262f679eed..91c3a65cae 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. @@ -45,9 +45,9 @@ public class FileSourceAdapterParser extends AbstractSingleBeanDefinitionParser String directory = element.getAttribute("directory"); String channel = element.getAttribute("channel"); String pollPeriod = element.getAttribute("poll-period"); - builder.addConstructorArg(directory); - builder.addConstructorArgReference(channel); - builder.addConstructorArg(pollPeriod); + builder.addConstructorArgValue(directory); + builder.addPropertyReference("channel", channel); + builder.addPropertyValue("period", pollPeriod); } } 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 index cf0419d3ba..f1965cd87e 100644 --- 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 @@ -21,6 +21,9 @@ import java.util.HashMap; import java.util.Iterator; import java.util.Map; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + /** * Tracks changes in a directory. This implementation is thread-safe as it * allows to synchronously process a new directory structure. @@ -30,6 +33,8 @@ import java.util.Map; */ public class DirectoryContentManager { + private final Log logger = LogFactory.getLog(this.getClass()); + private Map previousSnapshot = new HashMap(); private final Map backlog = new HashMap(); @@ -40,12 +45,18 @@ public class DirectoryContentManager { while (iter.hasNext()) { String fileName = iter.next().getKey(); if (!currentSnapshot.containsKey(fileName)) { + if (logger.isDebugEnabled()) { + logger.debug("Removing file '" + fileName + "' from backlog. It no longer exists in remote directory."); + } iter.remove(); } } for (String fileName : currentSnapshot.keySet()) { if (!this.previousSnapshot.containsKey(fileName) || (!this.previousSnapshot.get(fileName).equals(currentSnapshot.get(fileName)))) { + if (logger.isDebugEnabled()) { + logger.debug("Adding new or modified file '" + fileName + "' to backlog."); + } this.backlog.put(fileName, currentSnapshot.get(fileName)); } } @@ -53,7 +64,12 @@ public class DirectoryContentManager { } public synchronized void fileProcessed(String fileName) { - this.backlog.remove(fileName); + if (fileName != null) { + if (logger.isDebugEnabled()) { + logger.debug("Removing file '" + fileName + "' from the backlog. It has been processed."); + } + this.backlog.remove(fileName); + } } public Map getBacklog() { 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 index a731b43b41..af1c9efdfe 100644 --- 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 @@ -31,7 +31,10 @@ 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.FileNameGenerator; import org.springframework.integration.adapter.file.TextFileMapper; +import org.springframework.integration.message.Message; +import org.springframework.integration.message.MessageMapper; import org.springframework.integration.message.MessagingException; import org.springframework.util.Assert; import org.springframework.util.StringUtils; @@ -42,7 +45,7 @@ import org.springframework.util.StringUtils; * @author Marius Bogoevici * @author Mark Fisher */ -public class FtpSourceAdapter extends PollingSourceAdapter implements PollableSource { +public class FtpSourceAdapter extends PollingSourceAdapter implements PollableSource { private final static String DEFAULT_HOST = "localhost"; @@ -67,6 +70,8 @@ public class FtpSourceAdapter extends PollingSourceAdapter implements Poll private volatile boolean textBased = true; + private volatile MessageMapper mapper; + private final DirectoryContentManager directoryContentManager = new DirectoryContentManager(); private final FTPClient client = new FTPClient(); @@ -110,20 +115,26 @@ public class FtpSourceAdapter extends PollingSourceAdapter implements Poll protected void initialize() { this.setSource(this); if (this.isTextBased()) { - this.setMessageMapper(new TextFileMapper(this.localWorkingDirectory)); + this.mapper = new TextFileMapper(this.localWorkingDirectory); } else { - this.setMessageMapper(new ByteArrayFileMapper(this.localWorkingDirectory)); + this.mapper = new ByteArrayFileMapper(this.localWorkingDirectory); } } @Override - protected void onSend(File file) { - this.directoryContentManager.fileProcessed(file.getName()); + protected void onSend(Message message) { + String filename = message.getHeader().getProperty(FileNameGenerator.FILENAME_PROPERTY_KEY); + if (StringUtils.hasText(filename)) { + this.directoryContentManager.fileProcessed(filename); + } + else if (this.logger.isWarnEnabled()) { + logger.warn("No filename in Message header, cannot send notification of processing."); + } } - public final File poll() { + public final Message poll() { try { this.establishConnection(); FTPFile[] fileList = this.client.listFiles(); @@ -146,7 +157,7 @@ public class FtpSourceAdapter extends PollingSourceAdapter implements Poll FileOutputStream fileOutputStream = new FileOutputStream(file); this.client.retrieveFile(fileName, fileOutputStream); fileOutputStream.close(); - return file; + return this.mapper.toMessage(file); } catch (Exception e) { try { diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java index 6f8cc4d83b..7edc2ba92d 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/jms/JmsPollableSource.java @@ -20,6 +20,8 @@ import javax.jms.ConnectionFactory; import javax.jms.Destination; import org.springframework.integration.adapter.PollableSource; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; import org.springframework.jms.core.JmsTemplate; /** @@ -49,8 +51,8 @@ public class JmsPollableSource extends AbstractJmsTemplateBasedAdapter implement } - public Object poll() { - return this.getJmsTemplate().receiveAndConvert(); + public Message poll() { + return new GenericMessage(this.getJmsTemplate().receiveAndConvert()); } } diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java index e15ad84d99..ad8d5403a5 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/ByteStreamSource.java @@ -21,6 +21,8 @@ import java.io.IOException; import java.io.InputStream; import org.springframework.integration.adapter.PollableSource; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; import org.springframework.integration.message.MessagingException; /** @@ -65,7 +67,7 @@ public class ByteStreamSource implements PollableSource { this.shouldTruncate = shouldTruncate; } - public byte[] poll() { + public Message poll() { try { byte[] bytes; int bytesRead = 0; @@ -80,12 +82,12 @@ public class ByteStreamSource implements PollableSource { return null; } if (!this.shouldTruncate) { - return bytes; + return new GenericMessage(bytes); } else { byte[] result = new byte[bytesRead]; System.arraycopy(bytes, 0, result, 0, result.length); - return result; + return new GenericMessage(result); } } catch (IOException e) { diff --git a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java index 8f38e9d6f4..3c51415c13 100644 --- a/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java +++ b/spring-integration-adapters/src/main/java/org/springframework/integration/adapter/stream/CharacterStreamSource.java @@ -22,6 +22,7 @@ import java.io.Reader; import org.springframework.integration.adapter.PollableSource; import org.springframework.integration.message.MessagingException; +import org.springframework.integration.message.StringMessage; import org.springframework.util.Assert; /** @@ -55,13 +56,14 @@ public class CharacterStreamSource implements PollableSource { } - public String poll() { + public StringMessage poll() { try { synchronized (this.monitor) { if (!this.reader.ready()) { return null; } - return this.reader.readLine(); + String line = this.reader.readLine(); + return (line != null) ? new StringMessage(line) : null; } } catch (IOException e) { diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParserTests.java new file mode 100644 index 0000000000..f4a1461d8c --- /dev/null +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileSourceAdapterParserTests.java @@ -0,0 +1,50 @@ +/* + * 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.file.config; + +import static org.junit.Assert.assertEquals; + +import java.io.File; + +import org.junit.Test; + +import org.springframework.beans.DirectFieldAccessor; +import org.springframework.context.ApplicationContext; +import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.adapter.file.FileSourceAdapter; +import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.scheduling.PollingSchedule; + +/** + * @author Mark Fisher + */ +public class FileSourceAdapterParserTests { + + @Test + public void testFileSourceAdapterParser() { + ApplicationContext context = new ClassPathXmlApplicationContext("fileSourceAdapterParserTests.xml", this.getClass()); + FileSourceAdapter adapter = (FileSourceAdapter) context.getBean("adapter"); + DirectFieldAccessor accessor = new DirectFieldAccessor(adapter); + PollingSchedule schedule = (PollingSchedule) accessor.getPropertyValue("schedule"); + assertEquals(1234, schedule.getPeriod()); + File directory = (File) accessor.getPropertyValue("directory"); + assertEquals(System.getProperty("java.io.tmpdir"), directory.getAbsolutePath()); + MessageChannel channel = (MessageChannel) context.getBean("testChannel"); + assertEquals(channel, accessor.getPropertyValue("channel")); + } + +} diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileTargetAdapterParserTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileTargetAdapterParserTests.java index 7f427ef72b..feb9137f9d 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileTargetAdapterParserTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/FileTargetAdapterParserTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2007 the original author or authors. + * 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. diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileSourceAdapterParserTests.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileSourceAdapterParserTests.xml new file mode 100644 index 0000000000..37eec4f644 --- /dev/null +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileSourceAdapterParserTests.xml @@ -0,0 +1,21 @@ + + + + + + + + + + + + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileTargetAdapterParserTests.xml b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileTargetAdapterParserTests.xml index fdd81139df..88a4ae1cac 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileTargetAdapterParserTests.xml +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileTargetAdapterParserTests.xml @@ -2,8 +2,11 @@ @@ -13,4 +16,6 @@ + + diff --git a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java index 3b59f277d5..b8809ba03f 100644 --- a/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java +++ b/spring-integration-adapters/src/test/java/org/springframework/integration/adapter/stream/CharacterStreamSourceAdapterTests.java @@ -94,6 +94,7 @@ public class CharacterStreamSourceAdapterTests { MessageChannel channel = new SimpleChannel(); CharacterStreamSourceAdapter adapter = new CharacterStreamSourceAdapter(reader); adapter.setChannel(channel); + adapter.setInitialDelay(5000); adapter.setMaxMessagesPerTask(5); adapter.start(); int count = adapter.processMessages(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java index ae633b3ef3..10a77a18af 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/AbstractSourceAdapter.java @@ -23,8 +23,6 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.MessageChannel; import org.springframework.integration.message.Message; -import org.springframework.integration.message.MessageMapper; -import org.springframework.integration.message.SimplePayloadMessageMapper; import org.springframework.util.Assert; /** @@ -38,8 +36,6 @@ public abstract class AbstractSourceAdapter implements SourceAdapter, Initial private MessageChannel channel; - private MessageMapper mapper = new SimplePayloadMessageMapper(); - private long sendTimeout = -1; private volatile boolean initialized = false; @@ -58,15 +54,6 @@ public abstract class AbstractSourceAdapter implements SourceAdapter, Initial this.sendTimeout = sendTimeout; } - public void setMessageMapper(MessageMapper mapper) { - Assert.notNull(mapper, "'mapper' must not be null"); - this.mapper = mapper; - } - - protected MessageMapper getMessageMapper() { - return this.mapper; - } - public final void afterPropertiesSet() { if (this.channel == null) { throw new ConfigurationException("'channel' is required"); @@ -85,29 +72,16 @@ public abstract class AbstractSourceAdapter implements SourceAdapter, Initial protected void initialize() { } - protected boolean sendToChannel(T object) { + protected boolean sendToChannel(Message message) { if (!this.initialized) { this.afterPropertiesSet(); } - if (object == null) { + if (message == null) { if (logger.isDebugEnabled()) { logger.debug("adapter attempted to send a null object"); } return false; } - Message message = null; - if (object instanceof Message) { - message = (Message) object; - } - else { - message = this.mapper.toMessage(object); - } - if (message == null) { - if (logger.isWarnEnabled()) { - logger.warn("unable to create Message from source object: " + object); - } - return false; - } if (this.sendTimeout < 0) { return this.channel.send(message); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java index c7571b89ca..7a1e70f91f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/MethodInvokingSource.java @@ -21,6 +21,8 @@ import java.lang.reflect.Method; import org.springframework.beans.factory.InitializingBean; import org.springframework.integration.ConfigurationException; import org.springframework.integration.handler.HandlerMethodInvoker; +import org.springframework.integration.message.GenericMessage; +import org.springframework.integration.message.Message; import org.springframework.integration.util.MethodValidator; import org.springframework.util.Assert; @@ -54,11 +56,11 @@ public class MethodInvokingSource implements PollableSource, Initiali this.invoker.setMethodValidator(new MessageReceivingMethodValidator()); } - public Object poll() { + public Message poll() { if (this.invoker == null) { this.afterPropertiesSet(); } - return this.invoker.invokeMethod(new Object[] {}); + return new GenericMessage(this.invoker.invokeMethod(new Object[] {})); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java index 492e0e4023..520b0c37d0 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/adapter/PollableSource.java @@ -16,13 +16,15 @@ package org.springframework.integration.adapter; +import org.springframework.integration.message.Message; + /** - * Interface for any external data source that can be polled. + * Interface for any external message source that can be polled. * * @author Mark Fisher */ public interface PollableSource { - T poll(); + Message poll(); } 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 c1bdf03320..f6488f5cce 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 @@ -21,6 +21,7 @@ import java.util.concurrent.Executors; import org.springframework.context.Lifecycle; import org.springframework.integration.ConfigurationException; import org.springframework.integration.channel.MessageChannel; +import org.springframework.integration.message.Message; import org.springframework.integration.message.MessageMapper; import org.springframework.integration.scheduling.MessagingTask; import org.springframework.integration.scheduling.MessagingTaskScheduler; @@ -141,7 +142,7 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements int messagesProcessed = 0; int limit = this.maxMessagesPerTask; while (messagesProcessed < limit) { - T result = this.source.poll(); + Message result = this.source.poll(); if (result != null && this.sendToChannel(result)) { messagesProcessed++; this.onSend(result); @@ -154,11 +155,11 @@ public class PollingSourceAdapter extends AbstractSourceAdapter implements } /** - * Callback method invoked after an item is sent to the channel. + * Callback method invoked after a message is sent to the channel. *

* Subclasses may override. The default implementation does nothing. */ - protected void onSend(T sentItem) { + protected void onSend(Message sentMessage) { } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java index aa09aebe6e..8f6a826af3 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/MethodInvokingSourceTests.java @@ -21,6 +21,7 @@ import static org.junit.Assert.assertNotNull; import org.junit.Test; +import org.springframework.integration.message.Message; import org.springframework.integration.message.MessagingException; /** @@ -33,9 +34,10 @@ public class MethodInvokingSourceTests { MethodInvokingSource source = new MethodInvokingSource(); source.setObject(new TestBean()); source.setMethod("validMethod"); - Object result = source.poll(); + Message result = source.poll(); assertNotNull(result); - assertEquals("valid", result); + assertNotNull(result.getPayload()); + assertEquals("valid", result.getPayload()); } @Test(expected=MessagingException.class) diff --git a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java index 819ba1b135..96dcac2d8a 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/adapter/PollingSourceAdapterTests.java @@ -25,6 +25,7 @@ import java.util.concurrent.atomic.AtomicInteger; import org.junit.Test; import org.springframework.integration.channel.SimpleChannel; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.message.Message; /** @@ -40,7 +41,7 @@ public class PollingSourceAdapterTests { adapter.setChannel(channel); adapter.setPeriod(100); adapter.start(); - Message message = channel.receive(); + Message message = channel.receive(1000); assertNotNull("message should not be null", message); assertEquals("testing.1", message.getPayload()); } @@ -57,7 +58,7 @@ public class PollingSourceAdapterTests { adapter.processMessages(); adapter.processMessages(); adapter.stop(); - Message message1 = channel.receive(); + Message message1 = channel.receive(1000); assertNotNull("message should not be null", message1); assertEquals("testing.1", message1.getPayload()); Message message2 = channel.receive(0); @@ -111,11 +112,11 @@ public class PollingSourceAdapterTests { this.count.set(0); } - public String poll() { + public Message poll() { if (count.get() >= limit) { return null; } - return message + "." + count.incrementAndGet(); + return new GenericMessage(message + "." + count.incrementAndGet()); } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java index 670d755081..3609bf84d9 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/bus/MessageBusTests.java @@ -21,7 +21,6 @@ import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; -import java.util.Collection; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -269,7 +268,7 @@ public class MessageBusTests { this.latch = latch; } - public Collection poll() { + public Message poll() { latch.countDown(); throw new RuntimeException("intentional test failure"); } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java index feb3065db8..3d5519e2bf 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/dispatcher/SynchronousChannelTests.java @@ -97,8 +97,8 @@ public class SynchronousChannelTests { @Test public void testReceive() { SynchronousChannel channel = new SynchronousChannel(new PollableSource() { - public String poll() { - return "foo"; + public Message poll() { + return new StringMessage("foo"); } }); Message message = channel.receive(); @@ -175,7 +175,7 @@ public class SynchronousChannelTests { } - private static class MessageReturningTestSource implements PollableSource { + private static class MessageReturningTestSource implements PollableSource { private final String messageText;