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.

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

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<Message<?>> clear();
+
+	/**
+	 * Remove any {@link Message Messages} that are not accepted by the provided selector.
+	 */
+	List<Message<?>> 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();
+
+}

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<ObjectName> names = server.queryNames(new ObjectName("test.PriorityChannel:type=MessageHandler,*"), null);
@@ -49,6 +55,7 @@ public class PriorityChannelTests {
 		Set<ObjectName> 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 {