From bc9864ffebc8d7106778e0e28195752dba06e8e2 Mon Sep 17 00:00:00 2001
From: Mark Fisher <markfisher@vmware.com>
Date: Thu, 10 Apr 2008 23:40:46 +0000
Subject: [PATCH] SynchronousChannel receives a Message from its source rather
 than an Object.

---
 .../adapter/PollingSourceAdapter.java         | 25 +++++++++++++++----
 .../dispatcher/SynchronousChannel.java        |  6 ++---
 2 files changed, 22 insertions(+), 9 deletions(-)

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 f6488f5cce..6aba0a2ec5 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
@@ -16,6 +16,8 @@
 
 package org.springframework.integration.adapter;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.Executors;
 
 import org.springframework.context.Lifecycle;
@@ -132,6 +134,20 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
 		this.running = false;
 	}
 
+	public List<Message<T>> poll(int limit) {
+		List<Message<T>> results = new ArrayList<Message<T>>();
+		int count = 0;
+		while (count < limit) {
+			Message<T> message = this.source.poll();
+			if (message == null) {
+				break;
+			}
+			results.add(message);
+			count++;
+		}
+		return results;
+	}
+
 	public int processMessages() {
 		if (!this.isRunning()) {
 			if (logger.isDebugEnabled()) {
@@ -140,12 +156,11 @@ public class PollingSourceAdapter<T> extends AbstractSourceAdapter<T> implements
 			return 0;
 		}
 		int messagesProcessed = 0;
-		int limit = this.maxMessagesPerTask;
-		while (messagesProcessed < limit) {
-			Message<T> result = this.source.poll();
-			if (result != null && this.sendToChannel(result)) {
+		List<Message<T>> messages = this.poll(this.maxMessagesPerTask);
+		for (Message<T> message : messages) {
+			if (this.sendToChannel(message)) {
 				messagesProcessed++;
-				this.onSend(result);
+				this.onSend(message);
 			}
 			else {
 				return messagesProcessed;
diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
index 668f000f5e..c8de742864 100644
--- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
+++ b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
@@ -27,7 +27,6 @@ import org.springframework.integration.channel.AbstractMessageChannel;
 import org.springframework.integration.channel.DispatcherPolicy;
 import org.springframework.integration.handler.MessageHandler;
 import org.springframework.integration.message.Message;
-import org.springframework.integration.message.SimplePayloadMessageMapper;
 import org.springframework.integration.message.selector.MessageSelector;
 
 /**
@@ -84,10 +83,9 @@ public class SynchronousChannel extends AbstractMessageChannel {
 	@Override
 	protected Message<?> doReceive(long timeout) {
 		if (this.source != null) {
-			Object result = this.source.poll();
+			Message<?> result = this.source.poll();
 			if (result != null) {
-				return (result instanceof Message<?>) ? (Message<?>) result :
-						new SimplePayloadMessageMapper<Object>().toMessage(result);
+				return result;
 			}
 		}
 		return messageHolder.get().poll();