diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java index 48830e089b..fb6a6a71e1 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractMessageChannel.java @@ -26,11 +26,15 @@ import java.util.concurrent.CopyOnWriteArrayList; import org.springframework.core.OrderComparator; import org.springframework.core.convert.ConversionService; +import org.springframework.integration.channel.management.ChannelSendMetrics; +import org.springframework.integration.channel.management.MessageChannelMetrics; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.history.TrackableComponent; import org.springframework.integration.support.converter.DefaultDatatypeChannelMessageConverter; +import org.springframework.integration.support.management.Statistics; +import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageDeliveryException; @@ -38,6 +42,7 @@ import org.springframework.messaging.MessagingException; import org.springframework.messaging.converter.MessageConverter; import org.springframework.messaging.support.ChannelInterceptor; import org.springframework.util.Assert; +import org.springframework.util.StopWatch; import org.springframework.util.StringUtils; /** @@ -51,8 +56,9 @@ import org.springframework.util.StringUtils; * @author Gary Russell * @author Artem Bilan */ +@ManagedResource public abstract class AbstractMessageChannel extends IntegrationObjectSupport - implements MessageChannel, TrackableComponent, ChannelInterceptorAware { + implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics { private final ChannelInterceptorList interceptors = new ChannelInterceptorList(); @@ -66,6 +72,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport private volatile MessageConverter messageConverter; + private volatile boolean statsEnabled; + + private volatile ChannelSendMetrics channelMetrics; + @Override public String getComponentType() { return "channel"; @@ -76,6 +86,16 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport this.shouldTrack = shouldTrack; } + @Override + public void enableStats(boolean statsEnabled) { + this.statsEnabled = statsEnabled; + } + + @Override + public boolean isStatsEnabled() { + return this.statsEnabled; + } + /** * Specify the Message payload datatype(s) supported by this channel. If a * payload type does not match directly, but the 'conversionService' is @@ -192,6 +212,86 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport return this.interceptors; } + @Override + public void reset() { + this.channelMetrics.reset(); + } + + @Override + public int getSendCount() { + return this.channelMetrics.getSendCount(); + } + + @Override + public long getSendCountLong() { + return this.channelMetrics.getSendCountLong(); + } + + @Override + public int getSendErrorCount() { + return this.channelMetrics.getSendErrorCount(); + } + + @Override + public long getSendErrorCountLong() { + return this.channelMetrics.getSendErrorCountLong(); + } + + @Override + public double getTimeSinceLastSend() { + return this.channelMetrics.getTimeSinceLastSend(); + } + + @Override + public double getMeanSendRate() { + return this.channelMetrics.getMeanSendRate(); + } + + @Override + public double getMeanErrorRate() { + return this.channelMetrics.getMeanErrorRate(); + } + + @Override + public double getMeanErrorRatio() { + return this.channelMetrics.getMeanErrorRatio(); + } + + @Override + public double getMeanSendDuration() { + return this.channelMetrics.getMeanSendDuration(); + } + + @Override + public double getMinSendDuration() { + return this.channelMetrics.getMinSendDuration(); + } + + @Override + public double getMaxSendDuration() { + return this.channelMetrics.getMaxSendDuration(); + } + + @Override + public double getStandardDeviationSendDuration() { + return this.channelMetrics.getStandardDeviationSendDuration(); + } + + @Override + public Statistics getSendDuration() { + return this.channelMetrics.getSendDuration(); + } + + @Override + public Statistics getSendRate() { + return this.channelMetrics.getSendRate(); + } + + @Override + public Statistics getErrorRate() { + return this.channelMetrics.getErrorRate(); + } + @Override protected void onInit() throws Exception { super.onInit(); @@ -205,6 +305,15 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport } } } + initMetrics(); + } + + protected void initMetrics() { + setChannelMetrics(new ChannelSendMetrics(getComponentName())); + } + + protected void setChannelMetrics(ChannelSendMetrics channelMetrics) { + this.channelMetrics = channelMetrics; } /** @@ -263,6 +372,8 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport Deque interceptorStack = null; boolean sent = false; + boolean statsProcessed = false; + StopWatch timer = null; try { if (this.datatypes.length > 0) { message = this.convertPayloadIfNecessary(message); @@ -274,7 +385,14 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport return false; } } + if (this.statsEnabled) { + timer = this.channelMetrics.beforeSend(); + } sent = this.doSend(message, timeout); + if (this.statsEnabled) { + this.channelMetrics.afterSend(timer, sent); + statsProcessed = true; + } this.interceptors.postSend(message, this, sent); if (interceptorStack != null) { this.interceptors.afterSendCompletion(message, this, sent, null, interceptorStack); @@ -282,6 +400,9 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport return sent; } catch (Exception e) { + if (this.statsEnabled && !statsProcessed) { + this.channelMetrics.afterSend(timer, false); + } if (interceptorStack != null) { this.interceptors.afterSendCompletion(message, this, sent, e, interceptorStack); } @@ -293,6 +414,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport } } + protected ChannelSendMetrics getMetrics() { + return this.channelMetrics; + } + private Message convertPayloadIfNecessary(Message message) { // first pass checks if the payload type already matches any of the datatypes for (Class datatype : this.datatypes) { @@ -309,7 +434,10 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport return (Message) converted; } else { - return this.getMessageBuilderFactory().withPayload(converted).copyHeaders(message.getHeaders()).build(); + return getMessageBuilderFactory() + .withPayload(converted) + .copyHeaders(message.getHeaders()) + .build(); } } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java index ae82e61cf7..8b020b7aa7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/AbstractPollableChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -19,6 +19,8 @@ package org.springframework.integration.channel; import java.util.ArrayDeque; import java.util.Deque; +import org.springframework.integration.channel.management.ChannelReceiveMetrics; +import org.springframework.integration.channel.management.PollableChannelManagement; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; import org.springframework.messaging.support.ChannelInterceptor; @@ -28,8 +30,41 @@ import org.springframework.messaging.support.ChannelInterceptor; * * @author Mark Fisher * @author Artem Bilan + * @author Gary Russell */ -public abstract class AbstractPollableChannel extends AbstractMessageChannel implements PollableChannel { +public abstract class AbstractPollableChannel extends AbstractMessageChannel implements PollableChannel, + PollableChannelManagement { + + + @Override + protected void initMetrics() { + setChannelMetrics(new ChannelReceiveMetrics(this.getComponentName())); + } + + @Override + protected ChannelReceiveMetrics getMetrics() { + return (ChannelReceiveMetrics) super.getMetrics(); + } + + @Override + public int getReceiveCount() { + return getMetrics().getReceiveCount(); + } + + @Override + public long getReceiveCountLong() { + return getMetrics().getReceiveCountLong(); + } + + @Override + public int getReceiveErrorCount() { + return getMetrics().getReceiveErrorCount(); + } + + @Override + public long getReceiveErrorCountLong() { + return getMetrics().getReceiveErrorCountLong(); + } /** * Receive the first available message from this channel. If the channel @@ -40,7 +75,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp */ @Override public final Message receive() { - return this.receive(-1); + return receive(-1); } /** @@ -58,7 +93,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp */ @Override public final Message receive(long timeout) { - ChannelInterceptorList interceptorList = this.getInterceptors(); + ChannelInterceptorList interceptorList = getInterceptors(); Deque interceptorStack = null; try { if (interceptorList.getInterceptors().size() > 0) { @@ -69,6 +104,9 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp } } Message message = this.doReceive(timeout); + if (isStatsEnabled()) { + getMetrics().afterReceive(); + } message = interceptorList.postReceive(message, this); if (interceptorStack != null) { interceptorList.afterReceiveCompletion(message, this, null, interceptorStack); @@ -76,6 +114,9 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp return message; } catch (RuntimeException e) { + if (isStatsEnabled()) { + getMetrics().afterError(); + } if (interceptorStack != null) { interceptorList.afterReceiveCompletion(null, this, e, interceptorStack); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java index 619f507941..f5943c830f 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/NullChannel.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2015 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. @@ -19,40 +19,168 @@ package org.springframework.integration.channel; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.beans.factory.BeanNameAware; +import org.springframework.integration.channel.management.ChannelSendMetrics; +import org.springframework.integration.channel.management.MessageChannelMetrics; +import org.springframework.integration.support.context.NamedComponent; +import org.springframework.integration.support.management.Statistics; +import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; +import reactor.util.StringUtils; + /** * A channel implementation that essentially behaves like "/dev/null". * All receive() calls will return null, and all send() calls * will return true although no action is performed. * Note however that the invocations are logged at debug-level. - * + * * @author Mark Fisher + * @author Gary Russell */ -public class NullChannel implements PollableChannel { +@ManagedResource +public class NullChannel implements PollableChannel, MessageChannelMetrics, BeanNameAware, NamedComponent { private final Log logger = LogFactory.getLog(this.getClass()); + private volatile ChannelSendMetrics metrics = new ChannelSendMetrics("nullChannel"); + private volatile boolean statsEnabled; + + private String beanName; + + @Override + public void setBeanName(String beanName) { + this.beanName = beanName; + this.metrics = new ChannelSendMetrics(getComponentName()); + } + + @Override + public String getComponentName() { + return StringUtils.hasText(this.beanName) ? this.beanName: "nullChannel"; + } + + @Override + public String getComponentType() { + return "channel"; + } + + @Override + public void reset() { + this.metrics.reset(); + } + + @Override + public void enableStats(boolean statsEnabled) { + this.statsEnabled = statsEnabled; + } + + @Override + public boolean isStatsEnabled() { + return this.statsEnabled; + } + + @Override + public int getSendCount() { + return this.metrics.getSendCount(); + } + + @Override + public long getSendCountLong() { + return this.metrics.getSendCountLong(); + } + + @Override + public int getSendErrorCount() { + return this.metrics.getSendErrorCount(); + } + + @Override + public long getSendErrorCountLong() { + return this.metrics.getSendErrorCountLong(); + } + + @Override + public double getTimeSinceLastSend() { + return this.metrics.getTimeSinceLastSend(); + } + + @Override + public double getMeanSendRate() { + return this.metrics.getMeanSendRate(); + } + + @Override + public double getMeanErrorRate() { + return this.metrics.getMeanErrorRate(); + } + + @Override + public double getMeanErrorRatio() { + return this.metrics.getMeanErrorRatio(); + } + + @Override + public double getMeanSendDuration() { + return this.metrics.getMeanSendDuration(); + } + + @Override + public double getMinSendDuration() { + return this.metrics.getMinSendDuration(); + } + + @Override + public double getMaxSendDuration() { + return this.metrics.getMaxSendDuration(); + } + + @Override + public double getStandardDeviationSendDuration() { + return this.metrics.getStandardDeviationSendDuration(); + } + + @Override + public Statistics getSendDuration() { + return this.metrics.getSendDuration(); + } + + @Override + public Statistics getSendRate() { + return this.metrics.getSendRate(); + } + + @Override + public Statistics getErrorRate() { + return this.metrics.getErrorRate(); + } + + @Override public boolean send(Message message) { if (logger.isDebugEnabled()) { logger.debug("message sent to null channel: " + message); } + if (this.statsEnabled) { + this.metrics.afterSend(this.metrics.beforeSend(), true); + } return true; } + @Override public boolean send(Message message, long timeout) { return this.send(message); } + @Override public Message receive() { if (logger.isDebugEnabled()) { logger.debug("receive called on null channel"); - } + } return null; } + @Override public Message receive(long timeout) { return this.receive(); } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java index 851179f792..1a48fa4b37 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/PublishSubscribeChannel.java @@ -47,10 +47,7 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { private volatile Integer maxSubscribers; - @Override - public String getComponentType(){ - return "publish-subscribe-channel"; - } + /** * Create a PublishSubscribeChannel that will use an {@link Executor} * to invoke the handlers. If this is null, each invocation will occur in @@ -72,6 +69,11 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { } + @Override + public String getComponentType(){ + return "publish-subscribe-channel"; + } + /** * Provide an {@link ErrorHandler} strategy for handling Exceptions that * occur downstream from this channel. This will only be applied if @@ -144,9 +146,11 @@ public class PublishSubscribeChannel extends AbstractSubscribableChannel { /** * Callback method for initialization. + * @throws Exception the exception. */ @Override - public final void onInit() { + public final void onInit() throws Exception { + super.onInit(); if (this.executor != null) { if (!(this.executor instanceof ErrorHandlingTaskExecutor)) { if (this.errorHandler == null) { diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java index c6c7dcf910..ee4e53d5fb 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/QueueChannel.java @@ -24,6 +24,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; +import org.springframework.integration.channel.management.QueueChannelManagement; import org.springframework.integration.core.MessageSelector; import org.springframework.messaging.Message; import org.springframework.util.Assert; @@ -40,7 +41,8 @@ import org.springframework.util.Assert; * @author Gary Russell * @author Artem Bilan */ -public class QueueChannel extends AbstractPollableChannel implements QueueChannelOperations { +public class QueueChannel extends AbstractPollableChannel implements QueueChannelOperations, + QueueChannelManagement { private final Queue> queue; @@ -76,7 +78,6 @@ public class QueueChannel extends AbstractPollableChannel implements QueueChanne this(new LinkedBlockingQueue>()); } - @Override protected boolean doSend(Message message, long timeout) { Assert.notNull(message, "'message' must not be null"); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelReceiveMetrics.java similarity index 51% rename from spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelReceiveMetrics.java index 71b20d327a..9c6934172a 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelReceiveMetrics.java @@ -14,56 +14,37 @@ * limitations under the License. */ -package org.springframework.integration.monitor; +package org.springframework.integration.channel.management; import java.util.concurrent.atomic.AtomicLong; -import org.aopalliance.intercept.MethodInvocation; - -import org.springframework.jmx.export.annotation.ManagedMetric; import org.springframework.jmx.export.annotation.ManagedOperation; -import org.springframework.jmx.support.MetricType; -import org.springframework.messaging.MessageChannel; /** * @author Dave Syer * @author Gary Russell * @since 2.0 */ -public class PollableChannelMetrics extends DirectChannelMetrics { +public class ChannelReceiveMetrics extends ChannelSendMetrics { private final AtomicLong receiveCount = new AtomicLong(); private final AtomicLong receiveErrorCount = new AtomicLong(); - public PollableChannelMetrics(MessageChannel messageChannel, String name) { - super(messageChannel, name); + public ChannelReceiveMetrics(String name) { + super(name); } - @Override - protected Object doInvoke(MethodInvocation invocation, String method, MessageChannel channel) throws Throwable { - if ("receive".equals(method)) { - return monitorReceive(invocation, channel); - } - return super.doInvoke(invocation, method, channel); - } - - private Object monitorReceive(MethodInvocation invocation, MessageChannel channel) throws Throwable { + public void afterReceive() { if (logger.isTraceEnabled()) { - logger.trace("Recording receive on channel(" + channel + ") "); - } - try { - Object object = invocation.proceed(); - if (object != null) { - this.receiveCount.incrementAndGet(); - } - return object; - } - catch (Throwable e) {//NOSONAR - rethrown below - this.receiveErrorCount.incrementAndGet(); - throw e; + logger.trace("Recording receive on channel(" + getName() + ") "); } + this.receiveCount.incrementAndGet(); + } + + public void afterError() { + this.receiveErrorCount.incrementAndGet(); } @Override @@ -74,22 +55,18 @@ public class PollableChannelMetrics extends DirectChannelMetrics { this.receiveCount.set(0); } - @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Count") public int getReceiveCount() { return (int) this.receiveCount.get(); } - @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Count") public long getReceiveCountLong() { return this.receiveCount.get(); } - @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Error Count") public int getReceiveErrorCount() { return (int) this.receiveErrorCount.get(); } - @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Error Count") public long getReceiveErrorCountLong() { return this.receiveErrorCount.get(); } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelSendMetrics.java similarity index 68% rename from spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelSendMetrics.java index 8814ccf4f2..76f16a7665 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelSendMetrics.java @@ -11,18 +11,18 @@ * specific language governing permissions and limitations under the License. */ -package org.springframework.integration.monitor; +package org.springframework.integration.channel.management; import java.util.concurrent.atomic.AtomicLong; -import org.aopalliance.intercept.MethodInterceptor; -import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.integration.support.management.ExponentialMovingAverage; +import org.springframework.integration.support.management.ExponentialMovingAverageRate; +import org.springframework.integration.support.management.ExponentialMovingAverageRatio; +import org.springframework.integration.support.management.Statistics; import org.springframework.jmx.export.annotation.ManagedResource; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; import org.springframework.util.StopWatch; /** @@ -35,7 +35,7 @@ import org.springframework.util.StopWatch; * @since 2.0 */ @ManagedResource -public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMetrics { +public class ChannelSendMetrics { protected final Log logger = LogFactory.getLog(getClass()); @@ -45,7 +45,6 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe public static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10; - private final ExponentialMovingAverage sendDuration = new ExponentialMovingAverage( DEFAULT_MOVING_AVERAGE_WINDOW); @@ -64,11 +63,8 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe private final String name; - private final MessageChannel messageChannel; - - public DirectChannelMetrics(MessageChannel messageChannel, String name) { - this.messageChannel = messageChannel; + public ChannelSendMetrics(String name) { this.name = name; } @@ -79,44 +75,27 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe } } - public MessageChannel getMessageChannel() { - return messageChannel; - } - public String getName() { return name; } - @Override - public Object invoke(MethodInvocation invocation) throws Throwable { - String method = invocation.getMethod().getName(); - MessageChannel channel = (MessageChannel) invocation.getThis(); - return doInvoke(invocation, method, channel); - } - - protected Object doInvoke(MethodInvocation invocation, String method, MessageChannel channel) throws Throwable { - if ("send".equals(method)) { - Message message = (Message) invocation.getArguments()[0]; - return monitorSend(invocation, channel, message); - } - return invocation.proceed(); - } - - private Object monitorSend(MethodInvocation invocation, MessageChannel channel, Message message) throws Throwable { + public StopWatch beforeSend() { if (logger.isTraceEnabled()) { - logger.trace("Recording send on channel(" + channel + ") : message(" + message + ")"); + logger.trace("Recording send on channel(" + this.name + ")"); } - final StopWatch timer = new StopWatch(channel + ".send:execution"); - try { - timer.start(); + final StopWatch timer = new StopWatch(this.name + ".send:execution"); + timer.start(); - sendCount.incrementAndGet(); - sendRate.increment(); + this.sendCount.incrementAndGet(); + this.sendRate.increment(); - Object result = invocation.proceed(); + return timer; + } + public void afterSend(StopWatch timer, boolean result) { + if (timer != null) { timer.stop(); - if ((Boolean)result) { + if (result) { sendSuccessRatio.success(); sendDuration.append(timer.getTotalTimeMillis()); } @@ -125,22 +104,12 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe sendErrorCount.incrementAndGet(); sendErrorRate.increment(); } - return result; - } - catch (Throwable e) {//NOSONAR - rethrown below - sendSuccessRatio.failure(); - sendErrorCount.incrementAndGet(); - sendErrorRate.increment(); - throw e; - } - finally { if (logger.isTraceEnabled()) { logger.trace(timer); } } } - @Override public synchronized void reset() { sendDuration.reset(); sendErrorRate.reset(); @@ -150,77 +119,62 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe sendErrorCount.set(0); } - @Override public int getSendCount() { return (int) sendCount.get(); } - @Override public long getSendCountLong() { return sendCount.get(); } - @Override public int getSendErrorCount() { return (int) sendErrorCount.get(); } - @Override public long getSendErrorCountLong() { return sendErrorCount.get(); } - @Override public double getTimeSinceLastSend() { return sendRate.getTimeSinceLastMeasurement(); } - @Override public double getMeanSendRate() { return sendRate.getMean(); } - @Override public double getMeanErrorRate() { return sendErrorRate.getMean(); } - @Override public double getMeanErrorRatio() { return 1 - sendSuccessRatio.getMean(); } - @Override public double getMeanSendDuration() { return sendDuration.getMean(); } - @Override public double getMinSendDuration() { return sendDuration.getMin(); } - @Override public double getMaxSendDuration() { return sendDuration.getMax(); } - @Override public double getStandardDeviationSendDuration() { return sendDuration.getStandardDeviation(); } - @Override public Statistics getSendDuration() { return sendDuration.getStatistics(); } - @Override public Statistics getSendRate() { return sendRate.getStatistics(); } - @Override public Statistics getErrorRate() { return sendErrorRate.getStatistics(); } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/MessageChannelMetrics.java similarity index 90% rename from spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/management/MessageChannelMetrics.java index 1f95cc2abb..c8d23235c8 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/MessageChannelMetrics.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -14,8 +14,10 @@ * limitations under the License. */ -package org.springframework.integration.monitor; +package org.springframework.integration.channel.management; +import org.springframework.integration.support.management.Statistics; +import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedMetric; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.support.MetricType; @@ -25,6 +27,7 @@ import org.springframework.jmx.support.MetricType; * channel types. * * @author Dave Syer + * @author Gary Russell * @since 2.0 */ public interface MessageChannelMetrics { @@ -32,6 +35,12 @@ public interface MessageChannelMetrics { @ManagedOperation void reset(); + @ManagedOperation + void enableStats(boolean statsEnabled); + + @ManagedAttribute + boolean isStatsEnabled(); + /** * @return the number of successful sends */ diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/PollableChannelManagement.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/PollableChannelManagement.java new file mode 100644 index 0000000000..96721b8826 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/PollableChannelManagement.java @@ -0,0 +1,40 @@ +/* + * Copyright 2015 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.management; + +import org.springframework.jmx.export.annotation.ManagedMetric; +import org.springframework.jmx.support.MetricType; + +/** + * @author Gary Russell + * @since 4.2 + * + */ +public interface PollableChannelManagement extends MessageChannelMetrics { + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Count") + int getReceiveCount(); + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Count") + long getReceiveCountLong(); + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Error Count") + int getReceiveErrorCount(); + + @ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Error Count") + long getReceiveErrorCountLong(); + +} diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/QueueChannelManagement.java similarity index 60% rename from spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/management/QueueChannelManagement.java index 48a22d5026..7acf6d7d79 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/QueueChannelManagement.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2015 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. @@ -13,35 +13,22 @@ * See the License for the specific language governing permissions and * limitations under the License. */ +package org.springframework.integration.channel.management; -package org.springframework.integration.monitor; - -import org.springframework.integration.channel.QueueChannel; import org.springframework.jmx.export.annotation.ManagedMetric; import org.springframework.jmx.support.MetricType; /** - * @author Dave Syer - * @since 2.0 + * @author Gary Russell + * @since 4.2 + * */ -public class QueueChannelMetrics extends PollableChannelMetrics { - - private final QueueChannel channel; - - - public QueueChannelMetrics(QueueChannel channel, String name) { - super(channel, name); - this.channel = channel; - } +public interface QueueChannelManagement extends PollableChannelManagement { @ManagedMetric(metricType = MetricType.GAUGE, displayName = "QueueChannel Queue Size") - public int getQueueSize() { - return channel.getQueueSize(); - } + int getQueueSize(); @ManagedMetric(metricType = MetricType.GAUGE, displayName = "QueueChannel Remaining Capacity") - public int getRemainingCapacity() { - return channel.getRemainingCapacity(); - } + int getRemainingCapacity(); } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverage.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java similarity index 96% rename from spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverage.java rename to spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java index 0a0e6ffd2c..6b424e2863 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverage.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java @@ -1,5 +1,5 @@ /* - * Copyright 2009-2014 the original author or authors. + * Copyright 2009-2015 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 @@ -11,7 +11,9 @@ * specific language governing permissions and limitations under the License. */ -package org.springframework.integration.monitor; +package org.springframework.integration.support.management; + + /** * Cumulative statistics for a series of real numbers with higher weight given to recent data but without storing any diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRate.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java similarity index 97% rename from spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRate.java rename to spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java index 8c525b8f59..0c9def2063 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java @@ -1,5 +1,5 @@ /* - * Copyright 2009-2014 the original author or authors. + * Copyright 2009-2015 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 @@ -11,7 +11,9 @@ * specific language governing permissions and limitations under the License. */ -package org.springframework.integration.monitor; +package org.springframework.integration.support.management; + + /** * Cumulative statistics for an event rate with higher weight given to recent data but without storing any history. diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatio.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java similarity index 97% rename from spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatio.java rename to spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java index 1752f9ab89..a33bf9049f 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatio.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java @@ -1,5 +1,5 @@ /* - * Copyright 2009-2014 the original author or authors. + * Copyright 2009-2015 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 @@ -11,7 +11,9 @@ * specific language governing permissions and limitations under the License. */ -package org.springframework.integration.monitor; +package org.springframework.integration.support.management; + + /** * Cumulative statistics for success ratio with higher weight given to recent data but without storing any history. diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/Statistics.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/Statistics.java similarity index 93% rename from spring-integration-jmx/src/main/java/org/springframework/integration/monitor/Statistics.java rename to spring-integration-core/src/main/java/org/springframework/integration/support/management/Statistics.java index 5f6b838963..e7f57cc615 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/Statistics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/Statistics.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.monitor; +package org.springframework.integration.support.management; /** * @author Dave Syer diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRateTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRateTests.java similarity index 96% rename from spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRateTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRateTests.java index edb35ce8e3..97751fdaf5 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRateTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRateTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2013 the original author or authors. + * Copyright 2002-2015 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 @@ -10,7 +10,7 @@ * 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.monitor; +package org.springframework.integration.support.management; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRatioTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRatioTests.java similarity index 94% rename from spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRatioTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRatioTests.java index ce2de51154..cbaaab2a89 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageRatioTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageRatioTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2010 the original author or authors. + * Copyright 2002-2015 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. @@ -13,7 +13,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.springframework.integration.monitor; +package org.springframework.integration.support.management; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -23,11 +23,11 @@ import org.junit.Test; /** * @author Dave Syer - * + * */ public class ExponentialMovingAverageRatioTests { - private ExponentialMovingAverageRatio history = new ExponentialMovingAverageRatio( + private final ExponentialMovingAverageRatio history = new ExponentialMovingAverageRatio( 0.5, 10); @Test diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageTests.java b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java similarity index 89% rename from spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageTests.java rename to spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java index 1c2c0578b7..4cea8cfe4f 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ExponentialMovingAverageTests.java +++ b/spring-integration-core/src/test/java/org/springframework/integration/support/management/ExponentialMovingAverageTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2011 the original author or authors. + * Copyright 2002-2015 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. @@ -14,7 +14,7 @@ * limitations under the License. */ -package org.springframework.integration.monitor; +package org.springframework.integration.support.management; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -28,7 +28,7 @@ import org.junit.Test; */ public class ExponentialMovingAverageTests { - private ExponentialMovingAverage history = new ExponentialMovingAverage(10); + private final ExponentialMovingAverage history = new ExponentialMovingAverage(10); @Test public void testGetCount() { diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/JmxIntegrationConfigurationInitializer.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/JmxIntegrationConfigurationInitializer.java index ad62f6d659..b7069b6069 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/JmxIntegrationConfigurationInitializer.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/JmxIntegrationConfigurationInitializer.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -21,12 +21,12 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory; import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.integration.config.IntegrationConfigurationInitializer; -import org.springframework.integration.monitor.IntegrationMBeanExporter; /** * The JMX Integration infrastructure {@code beanFactory} initializer. * * @author Artem Bilan + * @author Gary Russell * @since 4.0 */ public class JmxIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer { @@ -39,10 +39,8 @@ public class JmxIntegrationConfigurationInitializer implements IntegrationConfig } private void registerMBeanExporterHelperIfNecessary(ConfigurableListableBeanFactory beanFactory) { - if (beanFactory.getBeanNamesForType(IntegrationMBeanExporter.class).length > 0) { - ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(MBEAN_EXPORTER_HELPER_BEAN_NAME, - new RootBeanDefinition(MBeanExporterHelper.class)); - } + ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(MBEAN_EXPORTER_HELPER_BEAN_NAME, + new RootBeanDefinition(MBeanExporterHelper.class)); } } diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterHelper.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterHelper.java index 7fb50e065b..e8f748a2a7 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterHelper.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterHelper.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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 @@ -37,8 +37,12 @@ import org.springframework.util.StringUtils; * of bean names that will be exported by the IntegrationMBeanExporter and merges it with the list * of 'excludedBeans' of MBeanExporter so it will not attempt to export them again. * + * Since 4.2, we unconditionally exclude all channels from standard context exporter (channels + * are now managed resources and would otherwise be picked up). + * * @author Oleg Zhurakousky * @author Artem Bilan + * @author Gary Russell * @since 2.1 * */ @@ -50,10 +54,19 @@ class MBeanExporterHelper implements BeanPostProcessor, Ordered, BeanFactoryAwar private final Set siBeanNames = new HashSet(); + /** + * For backwards compatibility, we don't want a context MBeanExporter to export + * integration components that are now managed resources. When there *is* an IMBE, + * *all* managed resources in org.springframework.integration are exluded. + */ + private final Set siBeansToExcludeWhenNoIMBE = new HashSet(); + private volatile DefaultListableBeanFactory beanFactory; private volatile boolean capturedAutoChannelCandidates; + private volatile boolean hasIntegrationExporter; + @Override public void setBeanFactory(BeanFactory beanFactory) throws BeansException { Assert.isInstanceOf(DefaultListableBeanFactory.class, beanFactory); @@ -71,8 +84,16 @@ class MBeanExporterHelper implements BeanPostProcessor, Ordered, BeanFactoryAwar className = ((StandardMethodMetadata) def.getSource()).getIntrospectedMethod().getReturnType().getName(); } if (StringUtils.hasText(className)){ - if (className.startsWith(SI_ROOT_PACKAGE) && !(className.endsWith(IntegrationMBeanExporter.class.getName()))){ - siBeanNames.add(beanName); + if (className.startsWith(SI_ROOT_PACKAGE) + && !(className.endsWith(IntegrationMBeanExporter.class.getName()))){ + this.siBeanNames.add(beanName); + } + if (className.startsWith(SI_ROOT_PACKAGE + "channel.") + && !(className.endsWith(IntegrationMBeanExporter.class.getName()))){ + this.siBeansToExcludeWhenNoIMBE.add(beanName); + } + else if (className.equals(IntegrationMBeanExporter.class.getName())) { + this.hasIntegrationExporter = true; } } } @@ -88,6 +109,7 @@ class MBeanExporterHelper implements BeanPostProcessor, Ordered, BeanFactoryAwar Collection autoCreateChannelCandidatesNames = (Collection) new DirectFieldAccessor(autoCreateChannelCandidates).getPropertyValue("channelNames"); this.siBeanNames.addAll(autoCreateChannelCandidatesNames); + this.siBeansToExcludeWhenNoIMBE.addAll(autoCreateChannelCandidatesNames); } this.capturedAutoChannelCandidates = true; } @@ -97,9 +119,19 @@ class MBeanExporterHelper implements BeanPostProcessor, Ordered, BeanFactoryAwar @SuppressWarnings("unchecked") Set excludedNames = (Set) mbeDfa.getPropertyValue(EXCLUDED_BEANS_PROPERTY_NAME); if (excludedNames != null) { - siBeanNames.addAll(excludedNames); + excludedNames = new HashSet(excludedNames); } - mbeDfa.setPropertyValue(EXCLUDED_BEANS_PROPERTY_NAME, siBeanNames); + else { + excludedNames = new HashSet(); + } + if (!this.hasIntegrationExporter) { + excludedNames.addAll(this.siBeansToExcludeWhenNoIMBE); + } + else { + excludedNames.addAll(this.siBeanNames); + } + //TODO: SF 4.1.5 and above now has additive exclusions (SPR-12686) + mbeDfa.setPropertyValue(EXCLUDED_BEANS_PROPERTY_NAME, excludedNames); } return bean; diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java index daa10545af..ae26538601 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java @@ -53,6 +53,8 @@ import org.springframework.context.ApplicationContextAware; import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.channel.management.MessageChannelMetrics; +import org.springframework.integration.channel.management.PollableChannelManagement; import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.core.MessageProducer; @@ -61,6 +63,7 @@ import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.history.MessageHistoryConfigurer; import org.springframework.integration.support.context.NamedComponent; +import org.springframework.integration.support.management.Statistics; import org.springframework.jmx.export.MBeanExporter; import org.springframework.jmx.export.UnableToRegisterMBeanException; import org.springframework.jmx.export.annotation.AnnotationJmxAttributeSource; @@ -73,7 +76,6 @@ import org.springframework.jmx.export.naming.MetadataNamingStrategy; import org.springframework.jmx.support.MetricType; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.messaging.PollableChannel; import org.springframework.util.Assert; import org.springframework.util.PatternMatchUtils; import org.springframework.util.ReflectionUtils; @@ -131,17 +133,17 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private final Set inboundLifecycleMessageProducers = new HashSet(); - private final Set channels = new HashSet(); + private final Set channels = new HashSet(); private final Map exposedBeans = new HashMap(); - private final Map channelsByName = new HashMap(); + private final Map channelsByName = new HashMap(); private final Map handlersByName = new HashMap(); private final Map sourcesByName = new HashMap(); - private final Map allChannelsByName = new HashMap(); + private final Map allChannelsByName = new HashMap(); private final Map allHandlersByName = new HashMap(); @@ -268,23 +270,11 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP bean = advised; } - if (bean instanceof MessageChannel) { - DirectChannelMetrics monitor; - MessageChannel target = (MessageChannel) extractTarget(bean); - if (bean instanceof PollableChannel) { - if (target instanceof QueueChannel) { - monitor = new QueueChannelMetrics((QueueChannel) target, beanName); - } - else { - monitor = new PollableChannelMetrics(target, beanName); - } - } - else { - monitor = new DirectChannelMetrics(target, beanName); - } - Object advised = applyChannelInterceptor(bean, monitor, beanClassLoader); + if (bean instanceof MessageChannel && bean instanceof MessageChannelMetrics + && bean instanceof NamedComponent) { + MessageChannelMetrics monitor = (MessageChannelMetrics) extractTarget(bean); + monitor.enableStats(true);//TODO: INT-3638 channels.add(monitor); - bean = advised; } if (bean instanceof MessageProducer && bean instanceof Lifecycle) { @@ -575,9 +565,9 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP @ManagedOperation public void stopActiveChannels() { // Stop any "active" channels (JMS etc). - for (Entry entry : this.allChannelsByName.entrySet()) { - DirectChannelMetrics metrics = entry.getValue(); - MessageChannel channel = metrics.getMessageChannel(); + for (Entry entry : this.allChannelsByName.entrySet()) { + MessageChannelMetrics metrics = entry.getValue(); + MessageChannel channel = (MessageChannel) metrics; if (channel instanceof Lifecycle) { if (logger.isInfoEnabled()) { logger.info("Stopping channel " + channel); @@ -648,8 +638,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP public int getQueuedMessageCount() { int count = 0; for (MessageChannelMetrics monitor : channels) { - if (monitor instanceof QueueChannelMetrics) { - count += ((QueueChannelMetrics) monitor).getQueueSize(); + if (monitor instanceof QueueChannel) { + count += ((QueueChannel) monitor).getQueueSize(); } } return count; @@ -686,8 +676,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP public long getChannelReceiveCountLong(String name) { if (channelsByName.containsKey(name)) { - if (channelsByName.get(name) instanceof PollableChannelMetrics) { - return ((PollableChannelMetrics) channelsByName.get(name)).getReceiveCountLong(); + if (channelsByName.get(name) instanceof PollableChannelManagement) { + return ((PollableChannelManagement) channelsByName.get(name)).getReceiveCountLong(); } } logger.debug("No channel found for (" + name + ")"); @@ -720,8 +710,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } private void registerChannels() { - for (DirectChannelMetrics monitor : channels) { - String name = monitor.getName(); + for (MessageChannelMetrics monitor : channels) { + String name = ((NamedComponent) monitor).getComponentName(); this.allChannelsByName.put(name, monitor); if (!PatternMatchUtils.simpleMatch(this.componentNamePatterns, name)) { continue; @@ -734,12 +724,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP channelsByName.put(name, monitor); } registerBeanNameOrInstance(monitor, beanKey); - // Expose the raw bean if it is managed - MessageChannel bean = monitor.getMessageChannel(); - if (assembler.includeBean(bean.getClass(), monitor.getName())) { - registerBeanInstance(bean, - this.getMonitoredIntegrationObjectBeanKey(bean, name)); - } } } } @@ -830,13 +814,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } } - private Object applyChannelInterceptor(Object bean, DirectChannelMetrics interceptor, ClassLoader beanClassLoader) { - NameMatchMethodPointcutAdvisor channelsAdvice = new NameMatchMethodPointcutAdvisor(interceptor); - channelsAdvice.addMethodName("send"); - channelsAdvice.addMethodName("receive"); - return applyAdvice(bean, channelsAdvice, beanClassLoader); - } - private Object applyHandlerInterceptor(Object bean, SimpleMessageHandlerMetrics interceptor, ClassLoader beanClassLoader) { NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(interceptor); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java index 2e7a93ee53..712bc73f86 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java @@ -17,6 +17,7 @@ package org.springframework.integration.monitor; import org.springframework.context.Lifecycle; +import org.springframework.integration.support.management.Statistics; import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.export.annotation.ManagedResource; diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java index 4d102a3435..e1015f870d 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java @@ -16,6 +16,7 @@ package org.springframework.integration.monitor; +import org.springframework.integration.support.management.Statistics; import org.springframework.jmx.export.annotation.ManagedMetric; import org.springframework.jmx.export.annotation.ManagedOperation; import org.springframework.jmx.support.MetricType; diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java index 09be02dd13..8e7e96becf 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java @@ -23,6 +23,8 @@ import org.aopalliance.intercept.MethodInvocation; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.integration.support.management.ExponentialMovingAverage; +import org.springframework.integration.support.management.Statistics; import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests-context.xml index 24f3c22539..96a85757fd 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests-context.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/MBeanAttributeFilterTests-context.xml @@ -64,6 +64,7 @@ SendCountLong SendErrorCount SendErrorCountLong + StatsEnabled diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/DynamicRouterTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/DynamicRouterTests.java index dfee72fdb1..fc589d4f46 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/DynamicRouterTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/DynamicRouterTests.java @@ -21,16 +21,15 @@ import static org.junit.Assert.assertEquals; import java.util.HashMap; import java.util.Map; -import org.junit.Ignore; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.messaging.MessageChannel; import org.springframework.integration.channel.QueueChannel; -import org.springframework.messaging.support.GenericMessage; import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.support.GenericMessage; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; @@ -95,7 +94,6 @@ public class DynamicRouterTests { } @Test @DirtiesContext - @Ignore // Requires Spring 3.2.3 TODO: Remove when minimum SF is >= 3.2.3 public void testRouteChangeMapNamedArgs() throws Exception { routingChannel.send(new GenericMessage("123")); assertEquals("123", processAChannel.receive(0).getPayload()); diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/configuration/EnableMBeanExportTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/configuration/EnableMBeanExportTests.java index 7c605f7cb3..3d3486f62a 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/configuration/EnableMBeanExportTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/configuration/EnableMBeanExportTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2014 the original author or authors. + * Copyright 2014-2015 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. @@ -33,7 +33,6 @@ import org.hamcrest.Matchers; import org.junit.Test; import org.junit.runner.RunWith; -import org.springframework.aop.support.AopUtils; import org.springframework.beans.factory.BeanFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContextInitializer; @@ -42,7 +41,6 @@ import org.springframework.context.annotation.Configuration; import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.config.EnableIntegration; -import org.springframework.integration.context.IntegrationContextUtils; import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport; import org.springframework.integration.monitor.IntegrationMBeanExporter; import org.springframework.integration.test.util.TestUtils; @@ -54,6 +52,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; /** * @author Artem Bilan + * @author Gary Russell * @since 4.0 */ @ContextConfiguration(initializers = EnableMBeanExportTests.EnvironmentApplicationContextInitializer.class) @@ -73,8 +72,6 @@ public class EnableMBeanExportTests { @SuppressWarnings("unchecked") @Test public void testEnableMBeanExport() throws MalformedObjectNameException, ClassNotFoundException { - assertTrue(AopUtils.isAopProxy(this.beanFactory.getBean(IntegrationContextUtils.NULL_CHANNEL_BEAN_NAME))); - assertTrue(AopUtils.isAopProxy(this.beanFactory.getBean(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME))); assertSame(this.mBeanServer, this.exporter.getServer()); String[] componentNamePatterns = TestUtils.getPropertyValue(this.exporter, "componentNamePatterns", String[].class); diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java index 32cfa7b55c..4bb93898e3 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MBeanExporterIntegrationTests.java @@ -34,6 +34,7 @@ import org.springframework.beans.factory.FactoryBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.context.Lifecycle; import org.springframework.context.support.GenericXmlApplicationContext; +import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.channel.QueueChannel; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.context.OrderlyShutdownCapable; @@ -350,17 +351,12 @@ public class MBeanExporterIntegrationTests { boolean isStopCalled(); } - public static class ActiveChannelImpl implements MessageChannel, Lifecycle, ActiveChannel { + public static class ActiveChannelImpl extends AbstractMessageChannel implements Lifecycle, ActiveChannel { private boolean stopCalled; @Override - public boolean send(Message message) { - return false; - } - - @Override - public boolean send(Message message, long timeout) { + protected boolean doSend(Message message, long timeout) { return false; } diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageMetricsAdviceTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageMetricsAdviceTests.java index f9a169456d..8ba122b758 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageMetricsAdviceTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/monitor/MessageMetricsAdviceTests.java @@ -1,3 +1,15 @@ +/* + * Copyright 2011-2015 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.monitor; import static org.hamcrest.Matchers.equalTo; @@ -8,22 +20,23 @@ import org.aopalliance.intercept.MethodInterceptor; import org.aopalliance.intercept.MethodInvocation; import org.junit.Before; import org.junit.Test; + import org.springframework.aop.framework.Advised; import org.springframework.aop.framework.ProxyFactory; import org.springframework.aop.support.NameMatchMethodPointcutAdvisor; import org.springframework.beans.factory.support.DefaultListableBeanFactory; +import org.springframework.integration.channel.NullChannel; +import org.springframework.integration.support.MessageBuilder; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessagingException; -import org.springframework.integration.channel.NullChannel; import org.springframework.messaging.MessageHandler; -import org.springframework.integration.monitor.IntegrationMBeanExporter; -import org.springframework.integration.support.MessageBuilder; +import org.springframework.messaging.MessagingException; import org.springframework.util.ClassUtils; /** * @author Tareq Abedrabbo * @author Dave Syer + * @author Gary Russell * @since 2.0.4 */ public class MessageMetricsAdviceTests { @@ -72,19 +85,6 @@ public class MessageMetricsAdviceTests { exported.handleMessage(MessageBuilder.withPayload("test").build()); } - @Test - public void adviceExportedChannel() throws Exception { - Advised advised = (Advised) mBeanExporter.postProcessAfterInitialization(channel, "testChannel"); - - DummyInterceptor interceptor = new DummyInterceptor(); - advised.addAdvice(interceptor); - - assertThat(advised.getAdvisors().length, equalTo(2)); - - ((MessageChannel) advised).send(MessageBuilder.withPayload("test").build()); - assertThat(interceptor.invoked, is(true)); - } - @Test public void exportAdvisedChannel() throws Exception { @@ -105,6 +105,7 @@ public class MessageMetricsAdviceTests { @SuppressWarnings("unused") boolean invoked = false; + @Override public void handleMessage(Message message) throws MessagingException { invoked = true; } @@ -114,6 +115,7 @@ public class MessageMetricsAdviceTests { boolean invoked = false; + @Override public Object invoke(MethodInvocation invocation) throws Throwable { invoked = true; return invocation.proceed(); diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration_/mbeanexporterhelper/INT_2626Tests.java b/spring-integration-jmx/src/test/java/org/springframework/integration_/mbeanexporterhelper/INT_2626Tests.java index 40c0aed41f..d5939eee77 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration_/mbeanexporterhelper/INT_2626Tests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration_/mbeanexporterhelper/INT_2626Tests.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2012 the original author or authors. + * Copyright 2002-2015 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 @@ -13,16 +13,18 @@ package org.springframework.integration_.mbeanexporterhelper; import org.junit.Test; + import org.springframework.context.support.ClassPathXmlApplicationContext; /** * @author Oleg Zhurakousky + * @author Gary Russell * */ public class INT_2626Tests { @Test // This context failed to load before the INT-2626 fix was applied public void testInt2626(){ - new ClassPathXmlApplicationContext("INT-2626-config.xml", this.getClass()); + new ClassPathXmlApplicationContext("INT-2626-config.xml", this.getClass()).close(); } }