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.

Polishing -- ** master only **

Eliminate conditional logic in AbstractMessageChannel by adding getHandlerCount() to
MessageDispatcher.

Do not cherry pick.

* Also fix JPA tests model for proper database `sequence` to avoid race condition when the test data is inserted with lower `id` than expected predefined data has

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

@@ -16,11 +16,9 @@
 
 package org.springframework.integration.channel;
 
-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 +35,24 @@ 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() {
+		return getRequiredDispatcher().getHandlerCount();
+	}
 
+	@Override
 	public boolean subscribe(MessageHandler handler) {
-		MessageDispatcher dispatcher = this.getRequiredDispatcher();
+		MessageDispatcher dispatcher = getRequiredDispatcher();
 		boolean added = dispatcher.addHandler(handler);
-		this.adjustCounterIfNecessary(dispatcher, added ? 1 : 0);
+		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;
@@ -57,16 +60,9 @@ public abstract class AbstractSubscribableChannel extends AbstractMessageChannel
 
 	private void adjustCounterIfNecessary(MessageDispatcher dispatcher, int delta) {
 		if (delta != 0) {
-			int counter = 0;
-			if (dispatcher instanceof AbstractDispatcher) {
-				counter = ((AbstractDispatcher) dispatcher).getHandlerCount();
-			}
-			else {
-				// some other dispatcher - hand-roll the counter
-				counter = this.handlerCounter.addAndGet(delta);
-			}
 			if (logger.isInfoEnabled()) {
-				logger.info("Channel '" + this.getFullChannelName() + "' has " + counter + " subscriber(s).");
+				logger.info("Channel '" + this.getFullChannelName() + "' has " + dispatcher.getHandlerCount()
+						+ " subscriber(s).");
 			}
 		}
 	}
@@ -74,7 +70,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 +79,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/dispatcher/AbstractDispatcher.java

@@ -140,10 +140,9 @@ public abstract class AbstractDispatcher implements MessageDispatcher {
 		return this.getClass().getSimpleName() + " with handlers: " + this.handlers.toString();
 	}
 
-	/**
-	 * @return The current number of handlers
-	 */
+	@Override
 	public int getHandlerCount() {
 		return this.handlers.size();
 	}
+
 }

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2008 the original author or authors.
+ * Copyright 2002-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.
@@ -23,13 +23,35 @@ import org.springframework.messaging.MessageHandler;
  * Strategy interface for dispatching messages to handlers.
  *
  * @author Mark Fisher
+ * @author Gary Russell
  */
 public interface MessageDispatcher {
 
+	/**
+	 * Add a message handler.
+	 * @param handler the handler.
+	 * @return true if successfully added.
+	 */
 	boolean addHandler(MessageHandler handler);
 
+	/**
+	 * Remove a message handler.
+	 * @param handler the handler.
+	 * @return true of successfully removed.
+	 */
 	boolean removeHandler(MessageHandler handler);
 
+	/**
+	 * Dispatch the message.
+	 * @param message the message.
+	 * @return true if dispatched.
+	 */
 	boolean dispatch(Message<?> message);
 
+	/**
+	 * Return the current handler count.
+	 * @return the handler count.
+	 */
+	int getHandlerCount();
+
 }

spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java

@@ -129,7 +129,7 @@ public class UnicastingDispatcher extends AbstractDispatcher {
 	}
 
 	private boolean doDispatch(Message<?> message) {
-		if (this.tryOptimizedDispatch(message)) {
+		if (tryOptimizedDispatch(message)) {
 			return true;
 		}
 		boolean success = false;

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

@@ -31,7 +31,6 @@ import org.apache.commons.logging.Log;
 import org.junit.Test;
 import org.mockito.Mockito;
 
-import org.springframework.integration.dispatcher.MessageDispatcher;
 import org.springframework.messaging.MessageHandler;
 import org.springframework.util.ReflectionUtils;
 
@@ -50,23 +49,6 @@ public class P2pChannelTests {
 		verifySubscriptions(channel);
 	}
 
-	@Test
-	public void testCustomChannelLoggingWithMoreThenOneSubscriberNotAbstractDispatcher() {
-		final MessageDispatcher mockDispatcher = mock(MessageDispatcher.class);
-		when(mockDispatcher.addHandler(Mockito.any(MessageHandler.class))).thenReturn(true);
-		when(mockDispatcher.removeHandler(Mockito.any(MessageHandler.class))).thenReturn(true).thenReturn(false).thenReturn(true);
-
-		final AbstractSubscribableChannel channel = new AbstractSubscribableChannel() {
-			@Override
-			protected MessageDispatcher getDispatcher() {
-				return mockDispatcher;
-			}
-		};
-		channel.setBeanName("customChannel");
-
-		verifySubscriptions(channel);
-	}
-
 	/**
 	 * @param channel
 	 */
@@ -91,15 +73,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();

spring-integration-jpa/src/test/java/org/springframework/integration/jpa/test/entity/StudentDomain.java

@@ -26,6 +26,7 @@ import javax.persistence.Id;
 import javax.persistence.NamedNativeQuery;
 import javax.persistence.NamedQueries;
 import javax.persistence.NamedQuery;
+import javax.persistence.SequenceGenerator;
 import javax.persistence.Table;
 import javax.persistence.Temporal;
 import javax.persistence.TemporalType;
@@ -48,11 +49,12 @@ import javax.persistence.TemporalType;
 	@NamedQuery(name = "updateStudent", query = "update Student s set s.lastName = :lastName, s.lastUpdated = :lastUpdated where s.rollNumber in (select max(a.rollNumber) from Student a)")
 })
 @NamedNativeQuery(resultClass = StudentDomain.class, name = "updateStudentNativeQuery", query = "update Student s set s.lastName = :lastName, lastUpdated = :lastUpdated where s.rollNumber in (select max(a.rollNumber) from Student a)")
+@SequenceGenerator(name = "student_sequence", initialValue = 1004, allocationSize = 1)
 public class StudentDomain {
 
 	@Id
 	@Column(name = "rollNumber")
-	@GeneratedValue(strategy = GenerationType.AUTO)
+	@GeneratedValue(generator = "student_sequence")
 	private Long rollNumber;
 
 	@Column(name = "firstName")

spring-integration-jpa/src/test/resources/H2-CreateTables.sql

@@ -1,7 +1,5 @@
 drop table StudentReadStatus;
 drop table Student;
-DROP TABLE OPENJPA_SEQUENCE_TABLE;
 
-CREATE TABLE OPENJPA_SEQUENCE_TABLE (ID TINYINT NOT NULL, SEQUENCE_VALUE BIGINT, PRIMARY KEY (ID));
 CREATE TABLE Student (rollNumber BIGINT generated by default as identity, dateOfBirth DATE, firstName VARCHAR(255), gender VARCHAR(255), lastName VARCHAR(255), lastUpdated TIMESTAMP, PRIMARY KEY (rollNumber));
 CREATE TABLE StudentReadStatus (rollNumber INTEGER NOT NULL, readAt TIMESTAMP, PRIMARY KEY (rollNumber));