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

SynchronousChannel receives a Message from its source rather than an Object.

Browse Source
This commit is contained in:
Mark Fisher
2008-04-10 23:40:46 +00:00
parent 069e7672e3
commit bc9864ffeb
2 changed files with 22 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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;

6
spring-integration-core/src/main/java/org/springframework/integration/dispatcher/SynchronousChannel.java
View File

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