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

INT-1095 Aggregator now properly handles the resolution of replyChannel by name.

Browse Source
This commit is contained in:
Mark Fisher
2010-04-25 16:06:41 +00:00
parent a6aff09329
commit a70ad276db
5 changed files with 111 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
org.springframework.integration/src/main/java/org/springframework/integration/aggregator/CorrelatingMessageHandler.java
View File

@@ -84,8 +84,6 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
private volatile MessageChannel discardChannel = new NullChannel();
private volatile ChannelResolver channelResolver;
private final IdTracker tracker = new IdTracker();
private final BlockingQueue<DelayedKey> keysInBuffer = new DelayQueue<DelayedKey>();
@@ -153,7 +151,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
}
public void setChannelResolver(ChannelResolver channelResolver) {
this.channelResolver = channelResolver;
super.setChannelResolver(channelResolver);
}
public void setDiscardChannel(MessageChannel discardChannel) {
@@ -192,7 +190,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
logger.debug("Completing group with correlationKey [" + correlationKey + "]");
}
outputProcessor.processAndSend(group, channelTemplate,
this.resolveReplyChannel(message, this.outputChannel, this.channelResolver));
this.resolveReplyChannel(message, this.outputChannel));
}
}
else {
@@ -292,8 +290,7 @@ public class CorrelatingMessageHandler extends AbstractMessageHandler implements
MessageGroup group = new MessageGroup(all, completionStrategy, key, deleteOrTrackCallback());
if (all.size() > 0) {
// last chance for normal completion
MessageChannel outputChannel = resolveReplyChannel(
all.get(0), this.outputChannel, this.channelResolver);
MessageChannel outputChannel = resolveReplyChannel(all.get(0), this.outputChannel);
boolean processed = false;
if (group.isComplete()) {
outputProcessor.processAndSend(group, channelTemplate, outputChannel);

4
org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java
View File

@@ -74,8 +74,8 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
protected abstract void handleMessageInternal(Message<?> message) throws Exception;
protected final MessageChannel resolveReplyChannel(Message<?> requestMessage,
MessageChannel defaultOutputChannel,
ChannelResolver channelResolver) {
MessageChannel defaultOutputChannel) {
ChannelResolver channelResolver = this.getChannelResolver();
MessageChannel replyChannel = defaultOutputChannel;
if (replyChannel == null) {
Object replyChannelHeader = requestMessage.getHeaders().getReplyChannel();

2
org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractReplyProducingMessageHandler.java
View File

@@ -99,7 +99,7 @@ public abstract class AbstractReplyProducingMessageHandler extends AbstractMessa
}
return;
}
MessageChannel replyChannel = resolveReplyChannel(message, this.outputChannel, this.getChannelResolver());
MessageChannel replyChannel = resolveReplyChannel(message, this.outputChannel);
MessageHeaders requestHeaders = message.getHeaders();
this.handleResult(result, requestHeaders, replyChannel);
}

18
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorReplyChannelTests-context.xml Normal file
View File

@@ -0,0 +1,18 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:beans="http://www.springframework.org/schema/beans"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<channel id="input"/>
<channel id="output">
<queue capacity="1"/>
</channel>
<splitter id="splitter" input-channel="input" output-channel="aggregatorChannel"/>
<aggregator id="aggregator" input-channel="aggregatorChannel"/>
</beans:beans>

87
org.springframework.integration/src/test/java/org/springframework/integration/aggregator/AggregatorReplyChannelTests.java Normal file
View File

@@ -0,0 +1,87 @@
/*
* Copyright 2002-2010 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.aggregator;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.util.ArrayList;
import java.util.List;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.channel.PollableChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Mark Fisher
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class AggregatorReplyChannelTests {
@Autowired
private volatile MessageChannel input;
@Autowired
private volatile PollableChannel output;
private final List<String> list = new ArrayList<String>();
@Before
public void setupList() {
list.add("foo");
list.add("bar");
}
@Test
public void replyChannelHeader() {
this.verifyReply(MessageBuilder.withPayload(list).setReplyChannel(output).build());
}
@Test // INT-1095
public void replyChannelNameHeader() {
this.verifyReply(MessageBuilder.withPayload(list).setReplyChannelName("output").build());
}
@SuppressWarnings("unchecked")
private void verifyReply(Message<?> message) {
assertNull(output.receive(0));
input.send(message);
Message<?> result = output.receive(0);
assertNotNull(result);
assertTrue(result.getPayload() instanceof List);
List<?> resultList = (List<?>) result.getPayload();
assertEquals(2, resultList.size());
assertTrue(resultList.contains("foo"));
assertTrue(resultList.contains("bar"));
}
}