diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java index d14e896e75..f7cba58170 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/RecipientListRouter.java @@ -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 channels; + private final MessageChannelTemplate channelTemplate = new MessageChannelTemplate(); + public void setChannels(List 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 false meaning + * that an Exception will be thrown whenever a send fails. To override + * this and suppress Exceptions, set the value to true. + */ + 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 false meaning that sequence headers will + * not be applied. If planning to use an Aggregator downstream + * with the default correlation and completion strategies, you should set + * this flag to true. + */ + 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 determineTargetChannels(Message message) { - return this.channels; + protected void handleMessageInternal(Message message) throws Exception { + List channelList = new ArrayList(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); + } + } } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java index da9dccbf3e..bfc62efa85 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/router/RecipientListRouterTests.java @@ -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 channels = new ArrayList(); @@ -48,11 +50,11 @@ public class RecipientListRouterTests { RecipientListRouter router = new RecipientListRouter(); router.setChannels(channels); router.afterPropertiesSet(); - Message message = new StringMessage("test"); - Collection resolved = router.determineTargetChannels(message); - assertEquals(2, resolved.size()); - assertTrue(resolved.contains(channel1)); - assertTrue(resolved.contains(channel2)); + Collection channelList = (Collection) + 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 channels = new ArrayList(); + channels.add(channelA); + channels.add(channelB); + channels.add(channelC); + router.setChannels(channels); + channelA.send(new StringMessage("blocker")); + Message 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 channels = new ArrayList(); + channels.add(channelA); + channels.add(channelB); + channels.add(channelC); + router.setChannels(channels); + channelB.send(new StringMessage("blocker")); + Message 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 channels = new ArrayList(); + channels.add(channelA); + channels.add(channelB); + channels.add(channelC); + router.setChannels(channels); + channelC.send(new StringMessage("blocker")); + Message 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 channels = new ArrayList(); + channels.add(channelA); + channels.add(channelB); + channels.add(channelC); + router.setChannels(channels); + channelA.send(new StringMessage("blocker")); + Message 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 channels = new ArrayList(); + channels.add(channelA); + channels.add(channelB); + channels.add(channelC); + router.setChannels(channels); + channelB.send(new StringMessage("blocker")); + Message 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 channels = new ArrayList(); + channels.add(channelA); + channels.add(channelB); + channels.add(channelC); + router.setChannels(channels); + channelC.send(new StringMessage("blocker")); + Message 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 channels = new ArrayList(); + channels.add(channelA); + channels.add(channelB); + router.setChannels(channels); + Message 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 channels = new ArrayList(); + channels.add(channelA); + channels.add(channelB); + router.setChannels(channels); + Message 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();