From 96ff547c8c06c9e465270485890974ccfbdfdebb Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Mon, 28 Nov 2016 12:23:50 -0500 Subject: [PATCH] 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 --- .../channel/AbstractSubscribableChannel.java | 34 ++++++++--------- .../dispatcher/AbstractDispatcher.java | 5 +-- .../dispatcher/MessageDispatcher.java | 24 +++++++++++- .../dispatcher/UnicastingDispatcher.java | 2 +- .../management/PollableChannelManagement.java | 2 + .../SubscribableChannelManagement.java | 38 +++++++++++++++++++ .../integration/channel/P2pChannelTests.java | 23 +++-------- .../jmx/MBeanAttributeFilterTests.java | 1 + .../jpa/test/entity/StudentDomain.java | 4 +- .../src/test/resources/H2-CreateTables.sql | 2 - 10 files changed, 90 insertions(+), 45 deletions(-) create mode 100644 spring-integration-core/src/main/java/org/springframework/integration/support/management/SubscribableChannelManagement.java diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java index 180a0be054..36c2d32dd4 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractSubscribableChannel.java +++ b/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; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java index 5aafb907da..54c9246642 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/AbstractDispatcher.java +++ b/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(); } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java index fa6717594e..209c181b05 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/MessageDispatcher.java +++ b/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(); + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java b/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java index 32801cbccc..76f5486da3 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/dispatcher/UnicastingDispatcher.java +++ b/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; diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/PollableChannelManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/PollableChannelManagement.java index 84e027eda5..015f8246f8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/PollableChannelManagement.java +++ b/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 * diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/SubscribableChannelManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/SubscribableChannelManagement.java new file mode 100644 index 0000000000..5c18cc1f72 --- /dev/null +++ b/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(); + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java b/spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java index 5eb31f2482..3b68231902 100644 --- a/spring-integration-core/src/test/java/org/springframework/integration/channel/P2pChannelTests.java +++ b/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()); } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java index 202f9c4016..540019a341 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests.java +++ b/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(); diff --git a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/test/entity/StudentDomain.java b/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/test/entity/StudentDomain.java index e5e25ecdbb..2613d924ec 100644 --- a/spring-integration-jpa/src/test/java/org/springframework/integration/jpa/test/entity/StudentDomain.java +++ b/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") diff --git a/spring-integration-jpa/src/test/resources/H2-CreateTables.sql b/spring-integration-jpa/src/test/resources/H2-CreateTables.sql index ae8736fe45..0ab0a7d01a 100644 --- a/spring-integration-jpa/src/test/resources/H2-CreateTables.sql +++ b/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));