From e822bfd908a210ee37a67b2d9f1264a43d07999f Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 17 Jan 2013 15:38:14 -0500 Subject: [PATCH] INT-2823 Add QueueChannelOperations Operations on QueueChannel (clear, getQueueSize etc) are not available when the IntegrationMBeanExporter is configured because they are not declared on the PollableChannel interface. Add an additional interface QueueChannelOperations. The proxy proxies all interfaces so users can cast the proxy to QueueChannelOperations to use its methods. --- .../integration/channel/QueueChannel.java | 13 +++-- .../channel/QueueChannelOperations.java | 52 +++++++++++++++++++ .../jmx/config/PriorityChannelTests.java | 17 ++++-- 3 files changed, 72 insertions(+), 10 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannelOperations.java 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 {