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

INT-679, INT-693 RecipientListRouter now supports a boolean 'applySequence' flag (false by default) and a boolean 'ignoreSendFailures' flag (also false by default).

Browse Source
This commit is contained in:
Mark Fisher
2009-06-29 22:39:49 +00:00
parent 914dfed1ef
commit b1e8c23d88
2 changed files with 314 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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.
@@ -16,12 +16,18 @@
package org.springframework.integration.router;
import java.util.Collection;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.channel.MessageChannelTemplate;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.core.MessageHeaders;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.util.Assert;
/**
@@ -32,22 +38,74 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
*/
public class RecipientListRouter extends AbstractMessageRouter implements InitializingBean {
public class RecipientListRouter extends AbstractMessageHandler implements InitializingBean {
private volatile boolean ignoreSendFailures;
private volatile boolean applySequence;
private volatile List<MessageChannel> channels;
private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate();
public void setChannels(List<MessageChannel> channels) {
this.channels = channels;
}
/**
* Set the timeout for sending a message to the resolved channel. By
* default, there is no timeout, meaning the send will block indefinitely.
*/
public void setTimeout(long timeout) {
this.channelTemplate.setSendTimeout(timeout);
}
/**
* Specify whether send failures for one or more of the recipients
* should be ignored. By default this is <code>false</code> meaning
* that an Exception will be thrown whenever a send fails. To override
* this and suppress Exceptions, set the value to <code>true</code>.
*/
public void setIgnoreSendFailures(boolean ignoreSendFailures) {
this.ignoreSendFailures = ignoreSendFailures;
}
/**
* Specify whether to apply the sequence number and size headers to the
* messages prior to sending to the recipient channels. By default, this
* value is <code>false</code> meaning that sequence headers will
* <em>not</em> be applied. If planning to use an Aggregator downstream
* with the default correlation and completion strategies, you should set
* this flag to <code>true</code>.
*/
public void setApplySequence(boolean applySequence) {
this.applySequence = applySequence;
}
public void afterPropertiesSet() {
Assert.notEmpty(this.channels, "a non-empty channel list is required");
}
@Override
protected Collection<MessageChannel> determineTargetChannels(Message<?> message) {
return this.channels;
protected void handleMessageInternal(Message<?> message) throws Exception {
List<MessageChannel> channelList = new ArrayList<MessageChannel>(this.channels);
int sequenceSize = channelList.size();
int sequenceNumber = 1;
for (MessageChannel channel : channelList) {
final Message<?> messageToSend = (!this.applySequence) ? message
: MessageBuilder.fromMessage(message)
.setSequenceNumber(sequenceNumber++)
.setSequenceSize(sequenceSize)
.setCorrelationId(message.getHeaders().getId())
.setHeader(MessageHeaders.ID, UUID.randomUUID())
.build();
boolean sent = this.channelTemplate.send(messageToSend, channel);
if (!sent && !this.ignoreSendFailures) {
throw new MessageDeliveryException(message,
"RecipientListRouter failed to send to channel: " + channel);
}
}
}
}

259
org.springframework.integration/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2008 the original author or authors.
* Copyright 2002-2009 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.
@@ -27,10 +27,11 @@ import java.util.Collections;
import java.util.List;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.Message;
import org.springframework.integration.core.MessageChannel;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.StringMessage;
/**
@@ -39,7 +40,8 @@ import org.springframework.integration.message.StringMessage;
public class RecipientListRouterTests {
@Test
public void resolveWithChannelList() {
@SuppressWarnings("unchecked")
public void channelConfig() {
QueueChannel channel1 = new QueueChannel();
QueueChannel channel2 = new QueueChannel();
List<MessageChannel> channels = new ArrayList<MessageChannel>();
@@ -48,11 +50,11 @@ public class RecipientListRouterTests {
RecipientListRouter router = new RecipientListRouter();
router.setChannels(channels);
router.afterPropertiesSet();
Message<String> message = new StringMessage("test");
Collection<MessageChannel> resolved = router.determineTargetChannels(message);
assertEquals(2, resolved.size());
assertTrue(resolved.contains(channel1));
assertTrue(resolved.contains(channel2));
Collection<MessageChannel> channelList = (Collection<MessageChannel>)
new DirectFieldAccessor(router).getPropertyValue("channels");
assertEquals(2, channelList.size());
assertTrue(channelList.contains(channel1));
assertTrue(channelList.contains(channel2));
}
@Test
@@ -90,6 +92,247 @@ public class RecipientListRouterTests {
assertNull(result2);
}
@Test(expected = MessageDeliveryException.class)
public void sendFailureOnFirstRecipientTriggersExceptionByDefault() {
QueueChannel channelA = new QueueChannel(1);
QueueChannel channelB = new QueueChannel(1);
QueueChannel channelC = new QueueChannel(1);
channelA.setBeanName("channelA");
channelB.setBeanName("channelB");
channelC.setBeanName("channelC");
RecipientListRouter router = new RecipientListRouter();
router.setTimeout(0);
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channelA);
channels.add(channelB);
channels.add(channelC);
router.setChannels(channels);
channelA.send(new StringMessage("blocker"));
Message<String> message = new StringMessage("test");
try {
router.handleMessage(message);
}
catch (RuntimeException e) {
Message<?> result1a = channelA.receive(0);
assertNotNull(result1a);
assertEquals("blocker", result1a.getPayload());
assertNull(channelB.receive(0));
assertNull(channelC.receive(0));
throw e;
}
}
@Test(expected = MessageDeliveryException.class)
public void sendFailureOnMiddleRecipientTriggersExceptionByDefault() {
QueueChannel channelA = new QueueChannel(1);
QueueChannel channelB = new QueueChannel(1);
QueueChannel channelC = new QueueChannel(1);
channelA.setBeanName("channelA");
channelB.setBeanName("channelB");
channelC.setBeanName("channelC");
RecipientListRouter router = new RecipientListRouter();
router.setTimeout(0);
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channelA);
channels.add(channelB);
channels.add(channelC);
router.setChannels(channels);
channelB.send(new StringMessage("blocker"));
Message<String> message = new StringMessage("test");
try {
router.handleMessage(message);
}
catch (RuntimeException e) {
Message<?> result1a = channelA.receive(0);
assertNotNull(result1a);
assertEquals("test", result1a.getPayload());
Message<?> result1b = channelB.receive(0);
assertNotNull(result1b);
assertEquals("blocker", result1b.getPayload());
assertNull(channelC.receive(0));
throw e;
}
}
@Test(expected = MessageDeliveryException.class)
public void sendFailureOnLastRecipientTriggersExceptionByDefault() {
QueueChannel channelA = new QueueChannel(1);
QueueChannel channelB = new QueueChannel(1);
QueueChannel channelC = new QueueChannel(1);
channelA.setBeanName("channelA");
channelB.setBeanName("channelB");
channelC.setBeanName("channelC");
RecipientListRouter router = new RecipientListRouter();
router.setTimeout(0);
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channelA);
channels.add(channelB);
channels.add(channelC);
router.setChannels(channels);
channelC.send(new StringMessage("blocker"));
Message<String> message = new StringMessage("test");
try {
router.handleMessage(message);
}
catch (RuntimeException e) {
Message<?> result1a = channelA.receive(0);
assertNotNull(result1a);
assertEquals("test", result1a.getPayload());
Message<?> result1b = channelB.receive(0);
assertNotNull(result1b);
assertEquals("test", result1b.getPayload());
Message<?> result1c = channelC.receive(0);
assertNotNull(result1c);
assertEquals("blocker", result1c.getPayload());
throw e;
}
}
@Test
public void sendFailureOnFirstRecipientIgnored() {
QueueChannel channelA = new QueueChannel(1);
QueueChannel channelB = new QueueChannel(1);
QueueChannel channelC = new QueueChannel(1);
channelA.setBeanName("channelA");
channelB.setBeanName("channelB");
channelC.setBeanName("channelC");
RecipientListRouter router = new RecipientListRouter();
router.setIgnoreSendFailures(true);
router.setTimeout(0);
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channelA);
channels.add(channelB);
channels.add(channelC);
router.setChannels(channels);
channelA.send(new StringMessage("blocker"));
Message<String> message = new StringMessage("test");
router.handleMessage(message);
Message<?> result1a = channelA.receive(0);
Message<?> result1b = channelB.receive(0);
Message<?> result1c = channelC.receive(0);
assertNotNull(result1a);
assertNotNull(result1b);
assertNotNull(result1c);
assertEquals("blocker", result1a.getPayload());
assertEquals("test", result1b.getPayload());
assertEquals("test", result1c.getPayload());
}
@Test
public void sendFailureOnMiddleRecipientIgnored() {
QueueChannel channelA = new QueueChannel(1);
QueueChannel channelB = new QueueChannel(1);
QueueChannel channelC = new QueueChannel(1);
channelA.setBeanName("channelA");
channelB.setBeanName("channelB");
channelC.setBeanName("channelC");
RecipientListRouter router = new RecipientListRouter();
router.setIgnoreSendFailures(true);
router.setTimeout(0);
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channelA);
channels.add(channelB);
channels.add(channelC);
router.setChannels(channels);
channelB.send(new StringMessage("blocker"));
Message<String> message = new StringMessage("test");
router.handleMessage(message);
Message<?> result1a = channelA.receive(0);
Message<?> result1b = channelB.receive(0);
Message<?> result1c = channelC.receive(0);
assertNotNull(result1a);
assertNotNull(result1b);
assertNotNull(result1c);
assertEquals("test", result1a.getPayload());
assertEquals("blocker", result1b.getPayload());
assertEquals("test", result1c.getPayload());
}
@Test
public void sendFailureOnLastRecipientIgnored() {
QueueChannel channelA = new QueueChannel(1);
QueueChannel channelB = new QueueChannel(1);
QueueChannel channelC = new QueueChannel(1);
channelA.setBeanName("channelA");
channelB.setBeanName("channelB");
channelC.setBeanName("channelC");
RecipientListRouter router = new RecipientListRouter();
router.setIgnoreSendFailures(true);
router.setTimeout(0);
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channelA);
channels.add(channelB);
channels.add(channelC);
router.setChannels(channels);
channelC.send(new StringMessage("blocker"));
Message<String> message = new StringMessage("test");
router.handleMessage(message);
Message<?> result1a = channelA.receive(0);
Message<?> result1b = channelB.receive(0);
Message<?> result1c = channelC.receive(0);
assertNotNull(result1a);
assertNotNull(result1b);
assertNotNull(result1c);
assertEquals("test", result1a.getPayload());
assertEquals("test", result1b.getPayload());
assertEquals("blocker", result1c.getPayload());
}
@Test
public void applySequenceNotEnabledByDefault() {
QueueChannel channelA = new QueueChannel(1);
QueueChannel channelB = new QueueChannel(1);
channelA.setBeanName("channelA");
channelB.setBeanName("channelB");
RecipientListRouter router = new RecipientListRouter();
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channelA);
channels.add(channelB);
router.setChannels(channels);
Message<String> message = new StringMessage("test");
router.handleMessage(message);
Message<?> result1a = channelA.receive(0);
Message<?> result1b = channelB.receive(0);
assertNotNull(result1a);
assertNotNull(result1b);
assertEquals("test", result1a.getPayload());
assertEquals(0, result1a.getHeaders().getSequenceNumber().intValue());
assertEquals(0, result1a.getHeaders().getSequenceSize().intValue());
assertNull(result1a.getHeaders().getCorrelationId());
assertEquals("test", result1b.getPayload());
assertEquals(0, result1b.getHeaders().getSequenceNumber().intValue());
assertEquals(0, result1b.getHeaders().getSequenceSize().intValue());
assertNull(result1b.getHeaders().getCorrelationId());
}
@Test
public void applySequenceEnabled() {
QueueChannel channelA = new QueueChannel(1);
QueueChannel channelB = new QueueChannel(1);
channelA.setBeanName("channelA");
channelB.setBeanName("channelB");
RecipientListRouter router = new RecipientListRouter();
router.setApplySequence(true);
List<MessageChannel> channels = new ArrayList<MessageChannel>();
channels.add(channelA);
channels.add(channelB);
router.setChannels(channels);
Message<String> message = new StringMessage("test");
router.handleMessage(message);
Message<?> result1a = channelA.receive(0);
Message<?> result1b = channelB.receive(0);
assertNotNull(result1a);
assertNotNull(result1b);
assertEquals("test", result1a.getPayload());
assertEquals(1, result1a.getHeaders().getSequenceNumber().intValue());
assertEquals(2, result1a.getHeaders().getSequenceSize().intValue());
assertEquals(message.getHeaders().getId(), result1a.getHeaders().getCorrelationId());
assertEquals("test", result1b.getPayload());
assertEquals(2, result1b.getHeaders().getSequenceNumber().intValue());
assertEquals(2, result1b.getHeaders().getSequenceSize().intValue());
assertEquals(message.getHeaders().getId(), result1b.getHeaders().getCorrelationId());
}
@Test(expected = IllegalArgumentException.class)
public void nullChannelListRejected() {
RecipientListRouter router = new RecipientListRouter();