From 0d8dc5ba811ce02cb385ec3fdd2a3e68dd41719e Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Tue, 4 Mar 2008 18:54:20 +0000 Subject: [PATCH] ChannelPurger now accepts more than one MessageChannel (with a varargs constructor). --- .../integration/channel/ChannelPurger.java | 37 +++++--- .../channel/ChannelPurgerTests.java | 87 +++++++++++++++++-- 2 files changed, 105 insertions(+), 19 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java index a72b8b89cc..a04254515b 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java @@ -16,16 +16,18 @@ package org.springframework.integration.channel; +import java.util.ArrayList; import java.util.List; import org.springframework.integration.message.Message; import org.springframework.integration.message.selector.MessageSelector; +import org.springframework.util.Assert; /** - * A utility class for purging {@link Message Messages} from a - * {@link MessageChannel}. Any message that does not match the - * provided {@link MessageSelector} will be removed from the channel. If no - * {@link MessageSelector} is provided, then all messages will be + * A utility class for purging {@link Message Messages} from one or more + * {@link MessageChannel MessageChannels}. Any message that does not + * match the provided {@link MessageSelector} will be removed from the channel. + * If no {@link MessageSelector} is provided, then all messages will be * cleared from the channel. *

* Note that the {@link #purge()} method operates on a snapshot of the messages @@ -39,26 +41,35 @@ import org.springframework.integration.message.selector.MessageSelector; */ public class ChannelPurger { - private MessageChannel channel; + private final MessageChannel[] channels; - private MessageSelector selector; + private final MessageSelector selector; - public ChannelPurger(MessageChannel channel) { - this.channel = channel; + public ChannelPurger(MessageChannel ... channels) { + this(null, channels); } - public ChannelPurger(MessageChannel channel, MessageSelector selector) { - this(channel); + public ChannelPurger(MessageSelector selector, MessageChannel ... channels) { + Assert.notEmpty(channels, "at least one channel is required"); + if (channels.length == 1) { + Assert.notNull(channels[0], "channel must not be null"); + } this.selector = selector; + this.channels = channels; } public final List> purge() { - if (this.selector == null) { - return this.channel.clear(); + List> purgedMessages = new ArrayList>(); + for (MessageChannel channel : this.channels) { + List> results = (this.selector == null) ? + channel.clear() : channel.purge(this.selector); + if (results != null) { + purgedMessages.addAll(results); + } } - return this.channel.purge(this.selector); + return purgedMessages; } } diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java index be14820a0b..61df56fd68 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java @@ -51,11 +51,11 @@ public class ChannelPurgerTests { channel.send(new StringMessage("test1")); channel.send(new StringMessage("test2")); channel.send(new StringMessage("test3")); - ChannelPurger purger = new ChannelPurger(channel, new MessageSelector() { + ChannelPurger purger = new ChannelPurger(new MessageSelector() { public boolean accept(Message message) { return false; } - }); + }, channel); List> purgedMessages = purger.purge(); assertEquals(3, purgedMessages.size()); assertNull(channel.receive(0)); @@ -67,11 +67,11 @@ public class ChannelPurgerTests { channel.send(new StringMessage("test1")); channel.send(new StringMessage("test2")); channel.send(new StringMessage("test3")); - ChannelPurger purger = new ChannelPurger(channel, new MessageSelector() { + ChannelPurger purger = new ChannelPurger(new MessageSelector() { public boolean accept(Message message) { return true; } - }); + }, channel); List> purgedMessages = purger.purge(); assertEquals(0, purgedMessages.size()); assertNotNull(channel.receive(0)); @@ -85,11 +85,11 @@ public class ChannelPurgerTests { channel.send(new StringMessage("test1")); channel.send(new StringMessage("test2")); channel.send(new StringMessage("test3")); - ChannelPurger purger = new ChannelPurger(channel, new MessageSelector() { + ChannelPurger purger = new ChannelPurger(new MessageSelector() { public boolean accept(Message message) { return (message.getPayload().equals("test2")); } - }); + }, channel); List> purgedMessages = purger.purge(); assertEquals(2, purgedMessages.size()); Message message = channel.receive(0); @@ -98,4 +98,79 @@ public class ChannelPurgerTests { assertNull(channel.receive(0)); } + @Test + public void testMultipleChannelsWithNoSelector() { + MessageChannel channel1 = new SimpleChannel(); + MessageChannel channel2 = new SimpleChannel(); + channel1.send(new StringMessage("test1")); + channel1.send(new StringMessage("test2")); + channel2.send(new StringMessage("test1")); + channel2.send(new StringMessage("test2")); + ChannelPurger purger = new ChannelPurger(channel1, channel2); + List> purgedMessages = purger.purge(); + assertEquals(4, purgedMessages.size()); + assertNull(channel1.receive(0)); + assertNull(channel2.receive(0)); + } + + @Test + public void testMultipleChannelsWithSelector() { + MessageChannel channel1 = new SimpleChannel(); + MessageChannel channel2 = new SimpleChannel(); + channel1.send(new StringMessage("test1")); + channel1.send(new StringMessage("test2")); + channel1.send(new StringMessage("test3")); + channel2.send(new StringMessage("test1")); + channel2.send(new StringMessage("test2")); + channel2.send(new StringMessage("test3")); + ChannelPurger purger = new ChannelPurger(new MessageSelector() { + public boolean accept(Message message) { + return (message.getPayload().equals("test2")); + } + }, channel1, channel2); + List> purgedMessages = purger.purge(); + assertEquals(4, purgedMessages.size()); + Message message1 = channel1.receive(0); + assertNotNull(message1); + assertEquals("test2", message1.getPayload()); + assertNull(channel1.receive(0)); + Message message2 = channel2.receive(0); + assertNotNull(message2); + assertEquals("test2", message2.getPayload()); + assertNull(channel2.receive(0)); + } + + @Test + public void testPurgeNoneWithSelectorAndMultipleChannels() { + MessageChannel channel1 = new SimpleChannel(); + MessageChannel channel2 = new SimpleChannel(); + channel1.send(new StringMessage("test1")); + channel1.send(new StringMessage("test2")); + channel2.send(new StringMessage("test1")); + channel2.send(new StringMessage("test2")); + ChannelPurger purger = new ChannelPurger(new MessageSelector() { + public boolean accept(Message message) { + return true; + } + }, channel1, channel2); + List> purgedMessages = purger.purge(); + assertEquals(0, purgedMessages.size()); + assertNotNull(channel1.receive(0)); + assertNotNull(channel1.receive(0)); + assertNotNull(channel2.receive(0)); + assertNotNull(channel2.receive(0)); + } + + @Test(expected=IllegalArgumentException.class) + public void testNullChannel() { + MessageChannel channel = null; + new ChannelPurger(channel); + } + + @Test(expected=IllegalArgumentException.class) + public void testEmptyChannelArray() { + MessageChannel[] channels = new MessageChannel[0]; + new ChannelPurger(channels); + } + }