diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java b/org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java index cf00c94823..b17e3bc47c 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/filter/MessageFilter.java @@ -17,8 +17,10 @@ package org.springframework.integration.filter; import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageChannel; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.ReplyMessageHolder; +import org.springframework.integration.message.MessageDeliveryException; import org.springframework.integration.message.MessageRejectedException; import org.springframework.integration.selector.MessageSelector; import org.springframework.util.Assert; @@ -39,6 +41,8 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler { private volatile boolean throwExceptionOnRejection; + private volatile MessageChannel discardChannel; + /** * Create a MessageFilter that will delegate to the given @@ -54,19 +58,45 @@ public class MessageFilter extends AbstractReplyProducingMessageHandler { * Specify whether this filter should throw a * {@link MessageRejectedException} when its selector does not accept a * Message. The default value is false meaning that rejected - * Messages will be quietly dropped. + * Messages will be quietly dropped or sent to the discard channel if + * available. Typically this value would not be true when + * a discard channel is provided, but if so, it will still apply + * (in such a case, the Message will be sent to the discard channel, + * and then the exception will be thrown). + * @see #setDiscardChannel(MessageChannel) */ public void setThrowExceptionOnRejection(boolean throwExceptionOnRejection) { this.throwExceptionOnRejection = throwExceptionOnRejection; } + /** + * Specify a channel where rejected Messages should be sent. If the discard + * channel is null (the default), rejected Messages will be dropped. However, + * the 'throwExceptionOnRejection' flag determines whether rejected Messages + * trigger an exception. That value is evaluated regardless of the presence + * of a discard channel. + * @see #setThrowExceptionOnRejection(boolean) + */ + public void setDiscardChannel(MessageChannel discardChannel) { + this.discardChannel = discardChannel; + } + @Override protected void handleRequestMessage(Message message, ReplyMessageHolder replyHolder) { if (this.selector.accept(message)) { replyHolder.set(message); } - else if (this.throwExceptionOnRejection) { - throw new MessageRejectedException(message); + else { + if (this.discardChannel != null) { + boolean discarded = this.sendReplyMessage(message, this.discardChannel); + if (!discarded) { + throw new MessageDeliveryException(message, + "failed to send rejected Message to the discard channel"); + } + } + if (this.throwExceptionOnRejection) { + throw new MessageRejectedException(message); + } } } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/filter/MessageFilterTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/filter/MessageFilterTests.java index a0a886fd8d..e76486c5d8 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/filter/MessageFilterTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/filter/MessageFilterTests.java @@ -130,4 +130,57 @@ public class MessageFilterTests { assertTrue(inputChannel.send(message)); } + @Test + public void filterDiscardsMessage() { + DirectChannel inputChannel = new DirectChannel(); + QueueChannel outputChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); + MessageFilter filter = new MessageFilter(new MessageSelector() { + public boolean accept(Message message) { + return false; + } + }); + filter.setOutputChannel(outputChannel); + filter.setDiscardChannel(discardChannel); + EventDrivenConsumer endpoint = new EventDrivenConsumer(inputChannel, filter); + endpoint.start(); + Message message = new StringMessage("test"); + assertTrue(inputChannel.send(message)); + Message reply = discardChannel.receive(0); + assertNotNull(reply); + assertEquals(message, reply); + assertNull(outputChannel.receive(0)); + } + + @Test(expected = MessageRejectedException.class) + public void filterDiscardsMessageAndThrowsException() throws Exception { + DirectChannel inputChannel = new DirectChannel(); + QueueChannel outputChannel = new QueueChannel(); + QueueChannel discardChannel = new QueueChannel(); + MessageFilter filter = new MessageFilter(new MessageSelector() { + public boolean accept(Message message) { + return false; + } + }); + filter.setOutputChannel(outputChannel); + filter.setDiscardChannel(discardChannel); + filter.setThrowExceptionOnRejection(true); + EventDrivenConsumer endpoint = new EventDrivenConsumer(inputChannel, filter); + endpoint.start(); + Message message = new StringMessage("test"); + try { + assertTrue(inputChannel.send(message)); + } + catch (Exception e) { + throw e; + } + finally { + Message reply = discardChannel.receive(0); + assertNotNull(reply); + assertEquals(message, reply); + assertNull(outputChannel.receive(0)); + } + + } + }