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

ChannelPurger now accepts more than one MessageChannel (with a varargs constructor).

Browse Source
This commit is contained in:
Mark Fisher
2008-03-04 18:54:20 +00:00
parent e3a018fa74
commit 0d8dc5ba81
2 changed files with 105 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
spring-integration-core/src/main/java/org/springframework/integration/channel/ChannelPurger.java
View File

@@ -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 <em>not</em> match the
* provided {@link MessageSelector} will be removed from the channel. If no
* {@link MessageSelector} is provided, then <em>all</em> messages will be
* A utility class for purging {@link Message Messages} from one or more
* {@link MessageChannel MessageChannels}. Any message that does <em>not</em>
* match the provided {@link MessageSelector} will be removed from the channel.
* If no {@link MessageSelector} is provided, then <em>all</em> messages will be
* cleared from the channel.
* <p>
* 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<Message<?>> purge() {
if (this.selector == null) {
return this.channel.clear();
List<Message<?>> purgedMessages = new ArrayList<Message<?>>();
for (MessageChannel channel : this.channels) {
List<Message<?>> results = (this.selector == null) ?
channel.clear() : channel.purge(this.selector);
if (results != null) {
purgedMessages.addAll(results);
}
}
return this.channel.purge(this.selector);
return purgedMessages;
}
}

87
spring-integration-core/src/test/java/org/springframework/integration/channel/ChannelPurgerTests.java
View File

@@ -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<Message<?>> 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<Message<?>> 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<Message<?>> 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<Message<?>> 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<Message<?>> 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<Message<?>> 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);
}
}