diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelPurger.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelPurger.java index 96e7a94da8..b6e661c020 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelPurger.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/ChannelPurger.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2010 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. @@ -25,7 +25,7 @@ import org.springframework.util.Assert; /** * A utility class for purging {@link Message Messages} from one or more - * {@link PollableChannel PollableChannels}. Any message that does not + * {@link QueueChannel QueueChannels}. 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. @@ -41,16 +41,16 @@ import org.springframework.util.Assert; */ public class ChannelPurger { - private final PollableChannel[] channels; + private final QueueChannel[] channels; private final MessageSelector selector; - public ChannelPurger(PollableChannel ... channels) { + public ChannelPurger(QueueChannel ... channels) { this(null, channels); } - public ChannelPurger(MessageSelector selector, PollableChannel ... channels) { + public ChannelPurger(MessageSelector selector, QueueChannel ... channels) { Assert.notEmpty(channels, "at least one channel is required"); if (channels.length == 1) { Assert.notNull(channels[0], "channel must not be null"); @@ -62,7 +62,7 @@ public class ChannelPurger { public final List> purge() { List> purgedMessages = new ArrayList>(); - for (PollableChannel channel : this.channels) { + for (QueueChannel channel : this.channels) { List> results = (this.selector == null) ? channel.clear() : channel.purge(this.selector); if (results != null) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java index e724daa671..6acc464d54 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/PollableChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2010 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,11 +16,8 @@ package org.springframework.integration.channel; -import java.util.List; - import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageChannel; -import org.springframework.integration.selector.MessageSelector; /** * @author Mark Fisher @@ -46,14 +43,4 @@ public interface PollableChannel extends MessageChannel { */ Message receive(long timeout); - /** - * Remove all {@link Message Messages} from this channel. - */ - List> clear(); - - /** - * Remove any {@link Message Messages} that are not accepted by the provided selector. - */ - List> purge(MessageSelector selector); - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/QueueChannel.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/QueueChannel.java index 9f73ca3a3b..8edcd3c7f0 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/QueueChannel.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/QueueChannel.java @@ -101,12 +101,18 @@ public class QueueChannel extends AbstractPollableChannel { } } + /** + * Remove all {@link Message Messages} from this channel. + */ public List> clear() { List> clearedMessages = new ArrayList>(); this.queue.drainTo(clearedMessages); return clearedMessages; } + /** + * Remove any {@link Message Messages} that are not accepted by the provided selector. + */ public List> purge(MessageSelector selector) { if (selector == null) { return this.clear(); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/AggregationResendTest.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/AggregationResendTest.java index 8d8f8691eb..881acadada 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/AggregationResendTest.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/scenarios/AggregationResendTest.java @@ -16,20 +16,23 @@ package org.springframework.integration.aggregator.scenarios; +import java.util.ArrayList; +import java.util.List; + import junit.framework.Assert; + +import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; + import org.springframework.beans.factory.annotation.Autowired; import org.springframework.integration.channel.DirectChannel; -import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.core.Message; import org.springframework.integration.message.MessageBuilder; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; -import java.util.ArrayList; -import java.util.List; - /** * Tests courtesy of Sean Crotty (INT-1093) * @@ -46,26 +49,31 @@ public class AggregationResendTest { DirectChannel input_for_aggregator_without_explicit_timeout; @Autowired - PollableChannel reply; + QueueChannel reply; + - @Test /** - * We expect to get back only one Message from the aggregator. We set no - * explicit timeout value on the aggregator. But apparently is automatically - * times out after 60 seconds. So What we'll see is that we get one aggregate - * Message back immediately. Then we'll get another 3 after the 60 seconds. + * We expect to get back only one Message from the aggregator. We set an + * explicit timeout value of 1 second on the aggregator. What we'll see is + * that we get one aggregate Message back immediately. + * + *

We should not get another 3 after the 1 second. */ + @Test public void testAggregatorWithoutExplicitTimeoutReturnsOnlyOneMessage() throws Exception { sendMessage(input_for_aggregator_with_explicit_timeout, 2000); } - @Test /** - * We expect to get back only one Message from the aggregator. We set an - * explicit timeout value of 1 second on the aggregator. What we'll see is - * that we get one aggregate Message back immediately. Then we'll get another - * 3 after the 1 second. + * We expect to get back only one Message from the aggregator. We set no + * explicit timeout value on the aggregator, but it automatically times out + * after 60 seconds. What we'll see is that we get one aggregate Message back + * immediately. + * + *

We should not get another 3 after the 60 seconds. */ + @Test + @Ignore // disabling from normal testing, should be the same behavior whether explicit or default public void testAggregatorWithTimeoutReturnsOnlyOneMessage() throws Exception { sendMessage(input_for_aggregator_without_explicit_timeout, 62000); }