diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java index 60e39265c2..b3ddd0a804 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2013 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. @@ -32,11 +32,12 @@ import org.springframework.util.Assert; * The capacity must be a positive integer value. For a zero-capacity version * based upon a {@link java.util.concurrent.SynchronousQueue}, consider the * {@link RendezvousChannel}. - * + * * @author Mark Fisher * @author Oleg Zhurakousky + * @author Gary Russell */ -public class QueueChannel extends AbstractPollableChannel { +public class QueueChannel extends AbstractPollableChannel implements QueueChannelOperations { private final BlockingQueue> queue; @@ -67,6 +68,7 @@ public class QueueChannel extends AbstractPollableChannel { } + @Override protected boolean doSend(Message message, long timeout) { Assert.notNull(message, "'message' must not be null"); try { @@ -85,6 +87,7 @@ public class QueueChannel extends AbstractPollableChannel { } } + @Override protected Message doReceive(long timeout) { try { if (timeout > 0) { @@ -131,9 +134,9 @@ public class QueueChannel extends AbstractPollableChannel { public int getQueueSize() { return this.queue.size(); } - + public int getRemainingCapacity() { return this.queue.remainingCapacity(); } - + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannelOperations.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannelOperations.java new file mode 100644 index 0000000000..dad36c15a2 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannelOperations.java @@ -0,0 +1,52 @@ +/* + * Copyright 2002-2013 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.channel; + +import java.util.List; + +import org.springframework.integration.Message; +import org.springframework.integration.core.MessageSelector; + +/** + * Operations available on a channel that has queuing semantics. + * + * @author Gary Russell + * @since 3.0 + * + */ +public interface QueueChannelOperations { + + /** + * 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); + + /** + * Return the current number of queued {@link Message Messages} in this channel. + */ + int getQueueSize(); + + /** + * Return the remaining capacity of this channel. + */ + int getRemainingCapacity(); + +} diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PriorityChannelTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PriorityChannelTests.java index 6823cb7fd9..d4524d4a63 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PriorityChannelTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/PriorityChannelTests.java @@ -1,11 +1,11 @@ /* - * Copyright 2002-2010 the original author or authors. - * + * Copyright 2002-2013 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. You may obtain a copy of the License at - * + * * http://www.apache.org/licenses/LICENSE-2.0 - * + * * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the * specific language governing permissions and limitations under the License. @@ -23,11 +23,14 @@ import javax.management.ObjectName; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.integration.channel.QueueChannelOperations; +import org.springframework.integration.core.PollableChannel; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Dave Syer + * @author Gary Russell * @since 2.0 */ @ContextConfiguration @@ -36,7 +39,10 @@ public class PriorityChannelTests { @Autowired private MBeanServer server; - + + @Autowired + private PollableChannel testChannel; + @Test public void testHandlerMBeanRegistration() throws Exception { Set names = server.queryNames(new ObjectName("test.PriorityChannel:type=MessageHandler,*"), null); @@ -49,6 +55,7 @@ public class PriorityChannelTests { Set names = server.queryNames(new ObjectName("test.PriorityChannel:type=MessageChannel,name=testChannel,*"), null); assertEquals(1, names.size()); assertEquals(0, server.getAttribute(names.iterator().next(), "QueueSize")); + assertEquals(0, ((QueueChannelOperations) testChannel).getQueueSize()); } public static class Source {