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..ffdfec73b6 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 @@ -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; } 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 894a3fb909..246ffc0e46 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 @@ -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()); } 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();