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.

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<Applica
 
 	public void onApplicationEvent(ApplicationEvent event) {
 		if (CollectionUtils.isEmpty(this.eventTypes)) {
-			this.sendToChannel(event);
+			this.sendMessage(event);
 			return;
 		}
 		for (Class<? extends ApplicationEvent> eventType : this.eventTypes) {
 			if (eventType.isAssignableFrom(event.getClass())) {
-				this.sendToChannel(event);
+				this.sendMessage(event);
 				return;
 			}
 		}
 	}
 
+	private void sendMessage(ApplicationEvent event) {
+		this.sendToChannel(new GenericMessage<ApplicationEvent>(event));
+	}
+
 }

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<T> extends AbstractMessageMapper<T, Fil
 			if (payload == null) {
 				return null;
 			}
-			Message<T> message = new GenericMessage<T>(this.getIdGenerator().generateId(), payload);
+			Message<T> message = new GenericMessage<T>(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());

spring-integration-adapters/src/main/java/org/springframework/integration/adapter/file/FileSource.java

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

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<File> {
+public class FileSourceAdapter extends PollingSourceAdapter<Object> implements PollableSource<Object> {
 
-	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<?, File> mapper = this.getMessageMapper();
-		if (mapper != null && (mapper instanceof AbstractFileMapper<?>)) {
-			((AbstractFileMapper<?>) mapper).setBackupDirectory(backupDirectory);
+	public Message<Object> 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;
 	}
 
 }

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.

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

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<String, FileInfo> previousSnapshot = new HashMap<String, FileInfo>();
 
 	private final Map<String, FileInfo> backlog = new HashMap<String, FileInfo>();
@@ -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<String, FileInfo> getBacklog() {

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<File> implements PollableSource<File> {
+public class FtpSourceAdapter extends PollingSourceAdapter<Object> implements PollableSource<Object> {
 
 	private final static String DEFAULT_HOST = "localhost";
 
@@ -67,6 +70,8 @@ public class FtpSourceAdapter extends PollingSourceAdapter<File> 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<File> 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<Object> 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<Object> poll() {
 		try {
 			this.establishConnection();
 			FTPFile[] fileList = this.client.listFiles();
@@ -146,7 +157,7 @@ public class FtpSourceAdapter extends PollingSourceAdapter<File> 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 {

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<Object> poll() {
+		return new GenericMessage<Object>(this.getJmsTemplate().receiveAndConvert());
 	}
 
 }

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<byte[]> {
 		this.shouldTruncate = shouldTruncate;
 	}
 
-	public byte[] poll() {
+	public Message<byte[]> poll() {
 		try {
 			byte[] bytes;
 			int bytesRead = 0;
@@ -80,12 +82,12 @@ public class ByteStreamSource implements PollableSource<byte[]> {
 				return null;
 			}
 			if (!this.shouldTruncate) {
-				return bytes;
+				return new GenericMessage<byte[]>(bytes);
 			}
 			else {
 				byte[] result = new byte[bytesRead];
 				System.arraycopy(bytes, 0, result, 0, result.length);
-				return result;
+				return new GenericMessage<byte[]>(result);
 			}
 		}
 		catch (IOException e) {

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<String> {
 	}
 
 
-	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) {

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

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.

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileSourceAdapterParserTests.xml

@@ -0,0 +1,21 @@
+<?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"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/context
+			http://www.springframework.org/schema/context/spring-context-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:file-source id="adapter" directory="${java.io.tmpdir}" channel="testChannel" poll-period="1234"/>
+
+	<context:property-placeholder/>
+
+</beans>

spring-integration-adapters/src/test/java/org/springframework/integration/adapter/file/config/fileTargetAdapterParserTests.xml

@@ -2,8 +2,11 @@
 <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"
+	xmlns:context="http://www.springframework.org/schema/context"
 	xsi:schemaLocation="http://www.springframework.org/schema/beans
 			http://www.springframework.org/schema/beans/spring-beans-2.5.xsd
+			http://www.springframework.org/schema/context
+			http://www.springframework.org/schema/context/spring-context-2.5.xsd
 			http://www.springframework.org/schema/integration
 			http://www.springframework.org/schema/integration/spring-integration-1.0.xsd">
 
@@ -13,4 +16,6 @@
 
 	<si:file-target id="adapter" directory="${java.io.tmpdir}" channel="testChannel"/>
 
+	<context:property-placeholder/>
+
 </beans>

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

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<T> implements SourceAdapter, Initial
 
 	private MessageChannel channel;
 
-	private MessageMapper<?,T> mapper = new SimplePayloadMessageMapper<T>();
-
 	private long sendTimeout = -1;
 
 	private volatile boolean initialized = false;
@@ -58,15 +54,6 @@ public abstract class AbstractSourceAdapter<T> implements SourceAdapter, Initial
 		this.sendTimeout = sendTimeout;
 	}
 
-	public void setMessageMapper(MessageMapper<?,T> mapper) {
-		Assert.notNull(mapper, "'mapper' must not be null");
-		this.mapper = mapper;
-	}
-
-	protected MessageMapper<?,T> 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<T> implements SourceAdapter, Initial
 	protected void initialize() {		
 	}
 
-	protected boolean sendToChannel(T object) {
+	protected boolean sendToChannel(Message<T> 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);
 		}

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<T> implements PollableSource<Object>, Initiali
 		this.invoker.setMethodValidator(new MessageReceivingMethodValidator());
 	}
 
-	public Object poll() {
+	public Message<Object> poll() {
 		if (this.invoker == null) {
 			this.afterPropertiesSet();
 		}
-		return this.invoker.invokeMethod(new Object[] {});
+		return new GenericMessage<Object>(this.invoker.invokeMethod(new Object[] {}));
 	}
 
 

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> {
 
-	T poll();
+	Message<T> poll();
 
 }

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<T> extends AbstractSourceAdapter<T> implements
 		int messagesProcessed = 0;
 		int limit = this.maxMessagesPerTask;
 		while (messagesProcessed < limit) {
-			T result = this.source.poll();
+			Message<T> result = this.source.poll();
 			if (result != null && this.sendToChannel(result)) {
 				messagesProcessed++;
 				this.onSend(result);
@@ -154,11 +155,11 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
 	}
 
 	/**
-	 * Callback method invoked after an item is sent to the channel.
+	 * Callback method invoked after a message is sent to the channel.
 	 * <p>
 	 * Subclasses may override. The default implementation does nothing.
 	 */
-	protected void onSend(T sentItem) {
+	protected void onSend(Message<T> sentMessage) {
 	}
 
 

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<TestBean> source = new MethodInvokingSource<TestBean>();
 		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)

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<String> poll() {
 			if (count.get() >= limit) {
 				return null;
 			}
-			return message + "." + count.incrementAndGet();
+			return new GenericMessage<String>(message + "." + count.incrementAndGet());
 		}
 	}
 

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<Object> poll() {
+		public Message<Object> poll() {
 			latch.countDown();
 			throw new RuntimeException("intentional test failure");
 		}

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<String>() {
-			public String poll() {
-				return "foo";
+			public Message<String> poll() {
+				return new StringMessage("foo");
 			}
 		});
 		Message<?> message = channel.receive();
@@ -175,7 +175,7 @@ public class SynchronousChannelTests {
 	}
 
 
-	private static class MessageReturningTestSource implements PollableSource<StringMessage> {
+	private static class MessageReturningTestSource implements PollableSource<String> {
 
 		private final String messageText;