INT-4174: Expose Channel Subscriber Count

JIRA: https://jira.spring.io/browse/INT-4174

I considered just using the channel counter but I suppose there is a (small) possibility
that someone might possibly use a custom dispatcher and subscribe to it directly.

(cherry picked from commit bd68eff)

spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java

@@ -21,6 +21,7 @@ import java.util.concurrent.atomic.AtomicInteger;
 import org.springframework.integration.MessageDispatchingException;
 import org.springframework.integration.dispatcher.AbstractDispatcher;
 import org.springframework.integration.dispatcher.MessageDispatcher;
+import org.springframework.integration.support.management.SubscribableChannelManagement;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageDeliveryException;
@@ -37,19 +38,30 @@ import org.springframework.util.Assert;
  * @author Gary Russell
  */
 public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
-		implements SubscribableChannel {
+		implements SubscribableChannel, SubscribableChannelManagement {
 
 	private final AtomicInteger handlerCounter = new AtomicInteger();
 
+	@Override
+	public int getSubscriberCount() {
+		MessageDispatcher dispatcher = getRequiredDispatcher();
+		if (dispatcher instanceof AbstractDispatcher) {
+			return ((AbstractDispatcher) dispatcher).getHandlerCount();
+		}
+		return this.handlerCounter.get();
+	}
+
+	@Override
 	public boolean subscribe(MessageHandler handler) {
-		MessageDispatcher dispatcher = this.getRequiredDispatcher();
+		MessageDispatcher dispatcher = getRequiredDispatcher();
 		boolean added = dispatcher.addHandler(handler);
 		this.adjustCounterIfNecessary(dispatcher, added ? 1 : 0);
 		return added;
 	}
 
+	@Override
 	public boolean unsubscribe(MessageHandler handle) {
-		MessageDispatcher dispatcher = this.getRequiredDispatcher();
+		MessageDispatcher dispatcher = getRequiredDispatcher();
 		boolean removed = dispatcher.removeHandler(handle);
 		this.adjustCounterIfNecessary(dispatcher, removed ? -1 : 0);
 		return removed;
@@ -74,7 +86,7 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
 	@Override
 	protected boolean doSend(Message<?> message, long timeout) {
 		try {
-			return this.getRequiredDispatcher().dispatch(message);
+			return getRequiredDispatcher().dispatch(message);
 		}
 		catch (MessageDispatchingException e) {
 			String description = e.getMessage() + " for channel '" + this.getFullChannelName() + "'.";
@@ -83,7 +95,7 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
 	}
 
 	private MessageDispatcher getRequiredDispatcher() {
-		MessageDispatcher dispatcher = this.getDispatcher();
+		MessageDispatcher dispatcher = getDispatcher();
 		Assert.state(dispatcher != null, "'dispatcher' must not be null");
 		return dispatcher;
 	}

spring-integration-core/src/main/java/org/springframework/integration/support/management/PollableChannelManagement.java

@@ -20,6 +20,8 @@ import org.springframework.jmx.export.annotation.ManagedMetric;
 import org.springframework.jmx.support.MetricType;
 
 /**
+ * Metrics for pollable channels.
+ *
  * @author Gary Russell
  * @since 4.2
  *

spring-integration-core/src/main/java/org/springframework/integration/support/management/SubscribableChannelManagement.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2016 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.support.management;
+
+import org.springframework.jmx.export.annotation.ManagedMetric;
+import org.springframework.jmx.support.MetricType;
+
+/**
+ * Metrics for subscribable channels.
+ *
+ * @author Gary Russell
+ * @since 4.3.6
+ *
+ */
+public interface SubscribableChannelManagement {
+
+	/**
+	 * The number of message handlers currently subscribed to this channel.
+	 * @return the number of subscribers.
+	 */
+	@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Subscriber Count")
+	int getSubscriberCount();
+
+}

spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java

@@ -102,15 +102,20 @@ public class P2pChannelTests {
 
 		MessageHandler handler1 = mock(MessageHandler.class);
 		channel.subscribe(handler1);
+		assertEquals(1, channel.getSubscriberCount());
 		assertEquals(String.format(log, 1), logs.remove(0));
 		MessageHandler handler2 = mock(MessageHandler.class);
 		channel.subscribe(handler2);
+		assertEquals(2, channel.getSubscriberCount());
 		assertEquals(String.format(log, 2), logs.remove(0));
 		channel.unsubscribe(handler1);
+		assertEquals(1, channel.getSubscriberCount());
 		assertEquals(String.format(log, 1), logs.remove(0));
 		channel.unsubscribe(handler1);
+		assertEquals(1, channel.getSubscriberCount());
 		assertEquals(0, logs.size());
 		channel.unsubscribe(handler2);
+		assertEquals(0, channel.getSubscriberCount());
 		assertEquals(String.format(log, 0), logs.remove(0));
 		verify(logger, times(4)).info(Mockito.anyString());
 	}

spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java

@@ -120,6 +120,7 @@ public class MBeanAttributeFilterTests {
 				"MeanSendRate",
 				"MinSendDuration",
 				"StandardDeviationSendDuration",
+				"SubscriberCount",
 				"TimeSinceLastSend"));
 
 		adapterNot.stop();