diff --git a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java index 7cbfb6282c..a431398371 100644 --- a/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java +++ b/spring-integration-amqp/src/main/java/org/springframework/integration/amqp/channel/PollableAmqpChannel.java @@ -24,7 +24,6 @@ import org.springframework.amqp.core.AmqpTemplate; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.core.RabbitAdmin; import org.springframework.amqp.rabbit.core.RabbitTemplate; -import org.springframework.integration.channel.management.ChannelReceiveMetrics; import org.springframework.integration.channel.management.PollableChannelManagement; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; @@ -58,16 +57,6 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable } - @Override - protected void initMetrics() { - setChannelMetrics(new ChannelReceiveMetrics(this.getComponentName())); - } - - @Override - protected ChannelReceiveMetrics getMetrics() { - return (ChannelReceiveMetrics) super.getMetrics(); - } - /** * Provide an explicitly configured queue name. If this is not provided, then a Queue will be created * implicitly with the channelName as its name. The implicit creation will require that either an AmqpAdmin @@ -136,6 +125,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable ChannelInterceptorList interceptorList = getInterceptors(); Deque interceptorStack = null; boolean counted = false; + boolean countsEnabled = isCountsEnabled(); try { if (interceptorList.getInterceptors().size() > 0) { interceptorStack = new ArrayDeque(); @@ -148,7 +138,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable if (object == null) { return null; } - if (isCountsEnabled()) { + if (countsEnabled) { getMetrics().afterReceive(); counted = true; } @@ -166,7 +156,7 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable return message; } catch (RuntimeException e) { - if (isCountsEnabled() && !counted) { + if (countsEnabled && !counted) { getMetrics().afterError(); } if (interceptorStack != null) { 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 aced7e05c3..56d150a187 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,14 +26,17 @@ 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.AbstractMessageChannelMetrics; +import org.springframework.integration.channel.management.DefaultMessageChannelMetrics; 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.ConfigurableMetricsAware; import org.springframework.integration.support.management.IntegrationManagedResource; +import org.springframework.integration.support.management.MetricsContext; import org.springframework.integration.support.management.Statistics; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; @@ -57,7 +60,8 @@ import org.springframework.util.StringUtils; */ @IntegrationManagedResource public abstract class AbstractMessageChannel extends IntegrationObjectSupport - implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics { + implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics, + ConfigurableMetricsAware { private final ChannelInterceptorList interceptors = new ChannelInterceptorList(); @@ -75,7 +79,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport private volatile boolean statsEnabled; - private volatile ChannelSendMetrics channelMetrics; + private volatile AbstractMessageChannelMetrics channelMetrics = new DefaultMessageChannelMetrics(); @Override public String getComponentType() { @@ -106,9 +110,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport this.countsEnabled = true; } this.statsEnabled = statsEnabled; - if (this.channelMetrics != null) { - this.channelMetrics.setFullStatsEnabled(statsEnabled); - } + this.channelMetrics.setFullStatsEnabled(statsEnabled); } @Override @@ -116,6 +118,16 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport return this.statsEnabled; } + protected AbstractMessageChannelMetrics getMetrics() { + return this.channelMetrics; + } + + @Override + public void configureMetrics(AbstractMessageChannelMetrics metrics) { + Assert.notNull(metrics, "'metrics' must not be null"); + this.channelMetrics = metrics; + } + /** * Specify the Message payload datatype(s) supported by this channel. If a * payload type does not match directly, but the 'conversionService' is @@ -325,16 +337,9 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport } } } - initMetrics(); - } - - protected void initMetrics() { - setChannelMetrics(new ChannelSendMetrics(getComponentName())); - } - - protected void setChannelMetrics(ChannelSendMetrics channelMetrics) { - this.channelMetrics = channelMetrics; - this.channelMetrics.setFullStatsEnabled(this.statsEnabled); + if (this.statsEnabled) { + this.channelMetrics.setFullStatsEnabled(true); + } } /** @@ -393,39 +398,42 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport Deque interceptorStack = null; boolean sent = false; - boolean statsProcessed = false; - long start = 0; + boolean metricsProcessed = false; + MetricsContext metrics = null; + boolean countsEnabled = this.countsEnabled; + ChannelInterceptorList interceptors = this.interceptors; + AbstractMessageChannelMetrics channelMetrics = this.channelMetrics; try { if (this.datatypes.length > 0) { message = this.convertPayloadIfNecessary(message); } - if (this.interceptors.getInterceptors().size() > 0) { + if (interceptors.getInterceptors().size() > 0) { interceptorStack = new ArrayDeque(); - message = this.interceptors.preSend(message, this, interceptorStack); + message = interceptors.preSend(message, this, interceptorStack); if (message == null) { return false; } } - if (this.countsEnabled) { - start = this.channelMetrics.beforeSend(); + if (countsEnabled) { + metrics = channelMetrics.beforeSend(); } sent = this.doSend(message, timeout); - if (this.countsEnabled) { - this.channelMetrics.afterSend(start, sent); - statsProcessed = true; + if (countsEnabled) { + channelMetrics.afterSend(metrics, sent); + metricsProcessed = true; } - this.interceptors.postSend(message, this, sent); + interceptors.postSend(message, this, sent); if (interceptorStack != null) { - this.interceptors.afterSendCompletion(message, this, sent, null, interceptorStack); + interceptors.afterSendCompletion(message, this, sent, null, interceptorStack); } return sent; } catch (Exception e) { - if (this.countsEnabled && !statsProcessed) { - this.channelMetrics.afterSend(start, false); + if (countsEnabled && !metricsProcessed) { + channelMetrics.afterSend(metrics, false); } if (interceptorStack != null) { - this.interceptors.afterSendCompletion(message, this, sent, e, interceptorStack); + interceptors.afterSendCompletion(message, this, sent, e, interceptorStack); } if (e instanceof MessagingException) { throw (MessagingException) e; @@ -435,10 +443,6 @@ 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) { 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 369a7a1dfa..6ae74c4b5d 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 @@ -19,7 +19,6 @@ 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; @@ -36,16 +35,6 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp 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(); @@ -96,6 +85,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp ChannelInterceptorList interceptorList = getInterceptors(); Deque interceptorStack = null; boolean counted = false; + boolean countsEnabled = isCountsEnabled(); try { if (interceptorList.getInterceptors().size() > 0) { interceptorStack = new ArrayDeque(); @@ -105,7 +95,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp } } Message message = this.doReceive(timeout); - if (isCountsEnabled()) { + if (countsEnabled) { getMetrics().afterReceive(); counted = true; } @@ -116,7 +106,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp return message; } catch (RuntimeException e) { - if (isCountsEnabled() && !counted) { + if (countsEnabled && !counted) { getMetrics().afterError(); } if (interceptorStack != null) { 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 6088ca40d7..97ec50c39b 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 @@ -20,13 +20,16 @@ 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.AbstractMessageChannelMetrics; +import org.springframework.integration.channel.management.DefaultMessageChannelMetrics; import org.springframework.integration.channel.management.MessageChannelMetrics; import org.springframework.integration.support.context.NamedComponent; +import org.springframework.integration.support.management.ConfigurableMetricsAware; import org.springframework.integration.support.management.IntegrationManagedResource; import org.springframework.integration.support.management.Statistics; import org.springframework.messaging.Message; import org.springframework.messaging.PollableChannel; +import org.springframework.util.Assert; import org.springframework.util.StringUtils; /** @@ -39,11 +42,12 @@ import org.springframework.util.StringUtils; * @author Gary Russell */ @IntegrationManagedResource -public class NullChannel implements PollableChannel, MessageChannelMetrics, BeanNameAware, NamedComponent { +public class NullChannel implements PollableChannel, MessageChannelMetrics, + ConfigurableMetricsAware, BeanNameAware, NamedComponent { private final Log logger = LogFactory.getLog(this.getClass()); - private volatile ChannelSendMetrics channelMetrics = new ChannelSendMetrics("nullChannel"); + private volatile AbstractMessageChannelMetrics channelMetrics = new DefaultMessageChannelMetrics("nullChannel"); private volatile boolean countsEnabled; @@ -54,7 +58,7 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics, Bean @Override public void setBeanName(String beanName) { this.beanName = beanName; - this.channelMetrics = new ChannelSendMetrics(getComponentName()); + this.channelMetrics = new DefaultMessageChannelMetrics(getComponentName()); } @Override @@ -67,6 +71,12 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics, Bean return "channel"; } + @Override + public void configureMetrics(AbstractMessageChannelMetrics metrics) { + Assert.notNull(metrics, "'metrics' must not be null"); + this.channelMetrics = metrics; + } + @Override public void reset() { this.channelMetrics.reset(); diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/AbstractMessageChannelMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/AbstractMessageChannelMetrics.java new file mode 100644 index 0000000000..2be316cc52 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/AbstractMessageChannelMetrics.java @@ -0,0 +1,120 @@ +/* + * 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.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.support.management.ConfigurableMetrics; +import org.springframework.integration.support.management.MetricsContext; +import org.springframework.integration.support.management.Statistics; + +/** + * Abstract base class for channel metrics implementations. + * + * @author Gary Russell + * @since 4.2 + * + */ +public abstract class AbstractMessageChannelMetrics implements ConfigurableMetrics { + + protected final Log logger = LogFactory.getLog(getClass()); + + protected final String name; + + private volatile boolean fullStatsEnabled; + + public AbstractMessageChannelMetrics(String name) { + this.name = name; + } + + /** + * When false, simple counts are maintained; when true complete statistics + * are maintained. + * @param fullStatsEnabled true for complete statistics. + */ + public void setFullStatsEnabled(boolean fullStatsEnabled) { + this.fullStatsEnabled = fullStatsEnabled; + } + + protected boolean isFullStatsEnabled() { + return fullStatsEnabled; + } + + /** + * Begin a send event. + * @return the context to be used in a subsequent {@link #afterSend(MetricsContext, boolean)} + * call. + */ + public abstract MetricsContext beforeSend(); + + /** + * End a send event. Note that implementations typically will not validate that the + * context is of the correct type and not null; callers should take care to ensure + * the context is the object returned by the previous {@link #beforeSend()} call. + * @param context the context. + * @param result true for success, false otherwise. + */ + public abstract void afterSend(MetricsContext context, boolean result); + + /** + * Reset all counters/statistics. + */ + public abstract void reset(); + + public abstract int getSendCount(); + + public abstract long getSendCountLong(); + + public abstract int getSendErrorCount(); + + public abstract long getSendErrorCountLong(); + + public abstract double getTimeSinceLastSend(); + + public abstract double getMeanSendRate(); + + public abstract double getMeanErrorRate(); + + public abstract double getMeanErrorRatio(); + + public abstract double getMeanSendDuration(); + + public abstract double getMinSendDuration(); + + public abstract double getMaxSendDuration(); + + public abstract double getStandardDeviationSendDuration(); + + public abstract Statistics getSendDuration(); + + public abstract Statistics getSendRate(); + + public abstract Statistics getErrorRate(); + + public abstract void afterReceive(); + + public abstract void afterError(); + + public abstract int getReceiveCount(); + + public abstract long getReceiveCountLong(); + + public abstract int getReceiveErrorCount(); + + public abstract long getReceiveErrorCountLong(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelReceiveMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelReceiveMetrics.java deleted file mode 100644 index 9c6934172a..0000000000 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelReceiveMetrics.java +++ /dev/null @@ -1,80 +0,0 @@ -/* - * 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 - * - * 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 java.util.concurrent.atomic.AtomicLong; - -import org.springframework.jmx.export.annotation.ManagedOperation; - -/** - * @author Dave Syer - * @author Gary Russell - * @since 2.0 - */ -public class ChannelReceiveMetrics extends ChannelSendMetrics { - - private final AtomicLong receiveCount = new AtomicLong(); - - private final AtomicLong receiveErrorCount = new AtomicLong(); - - - public ChannelReceiveMetrics(String name) { - super(name); - } - - public void afterReceive() { - if (logger.isTraceEnabled()) { - logger.trace("Recording receive on channel(" + getName() + ") "); - } - this.receiveCount.incrementAndGet(); - } - - public void afterError() { - this.receiveErrorCount.incrementAndGet(); - } - - @Override - @ManagedOperation - public synchronized void reset() { - super.reset(); - this.receiveErrorCount.set(0); - this.receiveCount.set(0); - } - - public int getReceiveCount() { - return (int) this.receiveCount.get(); - } - - public long getReceiveCountLong() { - return this.receiveCount.get(); - } - - public int getReceiveErrorCount() { - return (int) this.receiveErrorCount.get(); - } - - public long getReceiveErrorCountLong() { - return this.receiveErrorCount.get(); - } - - @Override - public String toString() { - return String.format("MessageChannelMonitor: [name=%s, sends=%d, receives=%d]", - getName(), getSendCount(), this.receiveCount.get()); - } - -} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelSendMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java similarity index 52% rename from spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelSendMetrics.java rename to spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java index 24524bf07b..2f9aabd1a2 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/channel/management/ChannelSendMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/channel/management/DefaultMessageChannelMetrics.java @@ -21,6 +21,7 @@ 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.MetricsContext; import org.springframework.integration.support.management.Statistics; /** @@ -32,7 +33,7 @@ import org.springframework.integration.support.management.Statistics; * @author Gary Russell * @since 2.0 */ -public class ChannelSendMetrics { +public class DefaultMessageChannelMetrics extends AbstractMessageChannelMetrics { protected final Log logger = LogFactory.getLog(getClass()); @@ -58,17 +59,16 @@ public class ChannelSendMetrics { private final AtomicLong sendErrorCount = new AtomicLong(); - private final String name; + protected final AtomicLong receiveCount = new AtomicLong(); - private volatile boolean fullStatsEnabled; + protected final AtomicLong receiveErrorCount = new AtomicLong(); - - public ChannelSendMetrics(String name) { - this.name = name; + public DefaultMessageChannelMetrics() { + super(null); } - public void setFullStatsEnabled(boolean fullStatsEnabled) { - this.fullStatsEnabled = fullStatsEnabled; + public DefaultMessageChannelMetrics(String name) { + super(name); } public void destroy() { @@ -77,118 +77,176 @@ public class ChannelSendMetrics { } } - public String getName() { - return name; - } - - public long beforeSend() { + @Override + public MetricsContext beforeSend() { if (logger.isTraceEnabled()) { logger.trace("Recording send on channel(" + this.name + ")"); } long start = 0; - if (this.fullStatsEnabled) { - start = System.currentTimeMillis(); + if (isFullStatsEnabled()) { + start = System.nanoTime(); this.sendRate.increment(); } this.sendCount.incrementAndGet(); - - return start; + return new DefaultChannelMetricsContext(start); } - public void afterSend(long start, boolean result) { - if (start > 0) { - long now = System.currentTimeMillis(); - long elapsed = now - start; - if (result && this.fullStatsEnabled) { - sendSuccessRatio.success(now); - sendDuration.append(elapsed); - } - else { - if (this.fullStatsEnabled) { - sendSuccessRatio.failure(now); - sendErrorCount.incrementAndGet(); - } - sendErrorRate.increment(); - } - if (logger.isTraceEnabled()) { - logger.trace("Elapsed: " + this.name + ": " + elapsed); + @Override + public void afterSend(MetricsContext context, boolean result) { + if (result && isFullStatsEnabled()) { + long now = System.nanoTime(); + this.sendSuccessRatio.success(now); + this.sendDuration.appendNanos(now - ((DefaultChannelMetricsContext) context).start); + } + else { + if (isFullStatsEnabled()) { + this.sendSuccessRatio.failure(System.nanoTime()); + this.sendErrorRate.increment(); } + this.sendErrorCount.incrementAndGet(); + } + if (logger.isTraceEnabled()) { + logger.trace("Elapsed: " + this.name + + ": " + ((System.nanoTime() - ((DefaultChannelMetricsContext) context).start) / 1000000.) + "ms"); } } + @Override public synchronized void reset() { - sendDuration.reset(); - sendErrorRate.reset(); - sendSuccessRatio.reset(); - sendRate.reset(); - sendCount.set(0); - sendErrorCount.set(0); + this.sendDuration.reset(); + this.sendErrorRate.reset(); + this.sendSuccessRatio.reset(); + this.sendRate.reset(); + this.sendCount.set(0); + this.sendErrorCount.set(0); + this.receiveErrorCount.set(0); + this.receiveCount.set(0); } + @Override public int getSendCount() { - return (int) sendCount.get(); + return (int) this.sendCount.get(); } + @Override public long getSendCountLong() { - return sendCount.get(); + return this.sendCount.get(); } + @Override public int getSendErrorCount() { - return (int) sendErrorCount.get(); + return (int) this.sendErrorCount.get(); } + @Override public long getSendErrorCountLong() { - return sendErrorCount.get(); + return this.sendErrorCount.get(); } + @Override public double getTimeSinceLastSend() { - return sendRate.getTimeSinceLastMeasurement(); + return this.sendRate.getTimeSinceLastMeasurement(); } + @Override public double getMeanSendRate() { - return sendRate.getMean(); + return this.sendRate.getMean(); } + @Override public double getMeanErrorRate() { - return sendErrorRate.getMean(); + return this.sendErrorRate.getMean(); } + @Override public double getMeanErrorRatio() { - return 1 - sendSuccessRatio.getMean(); + return 1 - this.sendSuccessRatio.getMean(); } + @Override public double getMeanSendDuration() { - return sendDuration.getMean(); + return this.sendDuration.getMean(); } + @Override public double getMinSendDuration() { - return sendDuration.getMin(); + return this.sendDuration.getMin(); } + @Override public double getMaxSendDuration() { - return sendDuration.getMax(); + return this.sendDuration.getMax(); } + @Override public double getStandardDeviationSendDuration() { - return sendDuration.getStandardDeviation(); + return this.sendDuration.getStandardDeviation(); } + @Override public Statistics getSendDuration() { - return sendDuration.getStatistics(); + return this.sendDuration.getStatistics(); } + @Override public Statistics getSendRate() { - return sendRate.getStatistics(); + return this.sendRate.getStatistics(); } + @Override public Statistics getErrorRate() { - return sendErrorRate.getStatistics(); + return this.sendErrorRate.getStatistics(); + } + + @Override + public void afterReceive() { + if (logger.isTraceEnabled()) { + logger.trace("Recording receive on channel(" + this.name + ") "); + } + this.receiveCount.incrementAndGet(); + } + + @Override + public void afterError() { + this.receiveErrorCount.incrementAndGet(); + } + + @Override + public int getReceiveCount() { + return (int) this.receiveCount.get(); + } + + @Override + public long getReceiveCountLong() { + return this.receiveCount.get(); + } + + @Override + public int getReceiveErrorCount() { + return (int) this.receiveErrorCount.get(); + } + + @Override + public long getReceiveErrorCountLong() { + return this.receiveErrorCount.get(); } @Override public String toString() { - return String.format("MessageChannelMonitor: [name=%s, sends=%d]", name, sendCount.get()); + return String.format("MessageChannelMonitor: [name=%s, sends=%d" + + (this.receiveCount.get() == 0 ? "" : this.receiveCount.get()) + + "]", name, sendCount.get()); + } + + private static class DefaultChannelMetricsContext implements MetricsContext { + + private final long start; + + public DefaultChannelMetricsContext(long start) { + this.start = start; + } + } } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java index c110618c41..ef2efffabd 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/AbstractMessageHandler.java @@ -19,11 +19,14 @@ package org.springframework.integration.handler; import org.springframework.core.Ordered; import org.springframework.integration.context.IntegrationObjectSupport; import org.springframework.integration.context.Orderable; +import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics; +import org.springframework.integration.handler.management.DefaultMessageHandlerMetrics; import org.springframework.integration.handler.management.MessageHandlerMetrics; -import org.springframework.integration.handler.management.SimpleMessageHandlerMetrics; import org.springframework.integration.history.MessageHistory; import org.springframework.integration.history.TrackableComponent; +import org.springframework.integration.support.management.ConfigurableMetricsAware; import org.springframework.integration.support.management.IntegrationManagedResource; +import org.springframework.integration.support.management.MetricsContext; import org.springframework.integration.support.management.Statistics; import org.springframework.messaging.Message; import org.springframework.messaging.MessageHandler; @@ -43,13 +46,13 @@ import org.springframework.util.Assert; */ @IntegrationManagedResource public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler, - MessageHandlerMetrics, TrackableComponent, Orderable { + MessageHandlerMetrics, ConfigurableMetricsAware, TrackableComponent, Orderable { private volatile boolean shouldTrack = false; private volatile int order = Ordered.LOWEST_PRECEDENCE; - private final SimpleMessageHandlerMetrics handlerMetrics = new SimpleMessageHandlerMetrics(); + private volatile AbstractMessageHandlerMetrics handlerMetrics = new DefaultMessageHandlerMetrics(); private volatile boolean statsEnabled; @@ -79,6 +82,19 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im this.shouldTrack = shouldTrack; } + @Override + public void configureMetrics(AbstractMessageHandlerMetrics metrics) { + Assert.notNull(metrics, "'metrics' must not be null"); + this.handlerMetrics = metrics; + } + + @Override + protected void onInit() throws Exception { + if (this.statsEnabled) { + this.handlerMetrics.setFullStatsEnabled(true); + } + } + @Override public final void handleMessage(Message message) { Assert.notNull(message, "Message must not be null"); @@ -86,19 +102,25 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im if (this.logger.isDebugEnabled()) { this.logger.debug(this + " received message: " + message); } - long start = 0; + MetricsContext start = null; + boolean countsEnabled = this.countsEnabled; + AbstractMessageHandlerMetrics handlerMetrics = this.handlerMetrics; try { if (message != null && this.shouldTrack) { message = MessageHistory.write(message, this, this.getMessageBuilderFactory()); } - if (this.countsEnabled) { - start = this.handlerMetrics.beforeHandle(message); + if (countsEnabled) { + start = handlerMetrics.beforeHandle(message); } this.handleMessageInternal(message); - this.handlerMetrics.afterHandle(start, true); + if (countsEnabled) { + handlerMetrics.afterHandle(start, true); + } } catch (Exception e) { - this.handlerMetrics.afterHandle(start, false); + if (countsEnabled) { + handlerMetrics.afterHandle(start, false); + } if (e instanceof MessagingException) { throw (MessagingException) e; } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java new file mode 100644 index 0000000000..f3a0c28526 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/AbstractMessageHandlerMetrics.java @@ -0,0 +1,96 @@ +/* + * 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.handler.management; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.integration.support.management.ConfigurableMetrics; +import org.springframework.integration.support.management.MetricsContext; +import org.springframework.integration.support.management.Statistics; +import org.springframework.messaging.Message; + +/** + * Abstract base class for handler metrics implementations. + * + * @author Gary Russell + * @since 4.2 + * + */ +public abstract class AbstractMessageHandlerMetrics implements ConfigurableMetrics { + + protected final Log logger = LogFactory.getLog(getClass()); + + protected final String name; + + private volatile boolean fullStatsEnabled; + + public AbstractMessageHandlerMetrics(String name) { + this.name = name; + } + + /** + * When false, simple counts are maintained; when true complete statistics + * are maintained. + * @param fullStatsEnabled true for complete statistics. + */ + public void setFullStatsEnabled(boolean fullStatsEnabled) { + this.fullStatsEnabled = fullStatsEnabled; + } + + protected boolean isFullStatsEnabled() { + return fullStatsEnabled; + } + + /** + * Begin a handle event. + * @param message the message to handle. + * @return the context. + */ + public abstract MetricsContext beforeHandle(Message message); + + /** + * End a handle event + * @param context the context. + * @param success true for success, false otherwise. + */ + public abstract void afterHandle(MetricsContext context, boolean success); + + public abstract void reset(); + + public abstract long getHandleCountLong(); + + public abstract int getHandleCount(); + + public abstract int getErrorCount(); + + public abstract long getErrorCountLong(); + + public abstract double getMeanDuration(); + + public abstract double getMinDuration(); + + public abstract double getMaxDuration(); + + public abstract double getStandardDeviationDuration(); + + public abstract int getActiveCount(); + + public abstract long getActiveCountLong(); + + public abstract Statistics getDuration(); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/SimpleMessageHandlerMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java similarity index 73% rename from spring-integration-core/src/main/java/org/springframework/integration/handler/management/SimpleMessageHandlerMetrics.java rename to spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java index e3ba4cea7d..4108f859ea 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/handler/management/SimpleMessageHandlerMetrics.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/management/DefaultMessageHandlerMetrics.java @@ -18,10 +18,8 @@ package org.springframework.integration.handler.management; import java.util.concurrent.atomic.AtomicLong; -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.MetricsContext; import org.springframework.integration.support.management.Statistics; import org.springframework.messaging.Message; @@ -30,9 +28,7 @@ import org.springframework.messaging.Message; * @author Gary Russell * @since 2.0 */ -public class SimpleMessageHandlerMetrics { - - private static final Log logger = LogFactory.getLog(SimpleMessageHandlerMetrics.class); +public class DefaultMessageHandlerMetrics extends AbstractMessageHandlerMetrics { private static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10; @@ -45,47 +41,48 @@ public class SimpleMessageHandlerMetrics { private final ExponentialMovingAverage duration = new ExponentialMovingAverage(DEFAULT_MOVING_AVERAGE_WINDOW); - private volatile String name; - - private volatile boolean fullStatsEnabled; - - public void setName(String name) { - this.name = name; + public DefaultMessageHandlerMetrics() { + super(null); } - public void setFullStatsEnabled(boolean fullStatsEnabled) { - this.fullStatsEnabled = fullStatsEnabled; + public DefaultMessageHandlerMetrics(String name) { + super(name); } - public long beforeHandle(Message message) throws Exception { + + @Override + public MetricsContext beforeHandle(Message message) { if (logger.isTraceEnabled()) { logger.trace("messageHandler(" + this.name + ") message(" + message + ") :"); } long start = 0; - if (this.fullStatsEnabled) { - start = System.currentTimeMillis(); + if (isFullStatsEnabled()) { + start = System.nanoTime(); } this.handleCount.incrementAndGet(); this.activeCount.incrementAndGet(); - return start; + return new DefaultHandlerMetricsContext(start); } - public void afterHandle(long start, boolean success) { + @Override + public void afterHandle(MetricsContext context, boolean success) { this.activeCount.decrementAndGet(); - if (this.fullStatsEnabled && success) { - this.duration.append(System.currentTimeMillis() - start); + if (isFullStatsEnabled() && success) { + this.duration.appendNanos(System.nanoTime() - ((DefaultHandlerMetricsContext) context).start); } else if (!success) { this.errorCount.incrementAndGet(); } } + @Override public synchronized void reset() { this.duration.reset(); this.errorCount.set(0); this.handleCount.set(0); } + @Override public long getHandleCountLong() { if (logger.isTraceEnabled()) { logger.trace("Getting Handle Count:" + this); @@ -93,44 +90,64 @@ public class SimpleMessageHandlerMetrics { return this.handleCount.get(); } + @Override public int getHandleCount() { return (int) getHandleCountLong(); } + @Override public int getErrorCount() { return (int) this.errorCount.get(); } + @Override public long getErrorCountLong() { return this.errorCount.get(); } + @Override public double getMeanDuration() { return this.duration.getMean(); } + @Override public double getMinDuration() { return this.duration.getMin(); } + @Override public double getMaxDuration() { return this.duration.getMax(); } + @Override public double getStandardDeviationDuration() { return this.duration.getStandardDeviation(); } + @Override public int getActiveCount() { return (int) this.activeCount.get(); } + @Override public long getActiveCountLong() { return this.activeCount.get(); } + @Override public Statistics getDuration() { return this.duration.getStatistics(); } + private static class DefaultHandlerMetricsContext implements MetricsContext { + + private final long start; + + public DefaultHandlerMetricsContext(long start) { + this.start = start; + } + + } + } diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ConfigurableMetrics.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ConfigurableMetrics.java new file mode 100644 index 0000000000..99959b1f16 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ConfigurableMetrics.java @@ -0,0 +1,27 @@ +/* + * 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.support.management; + +/** + * Marker interface for metrics. + * + * @author Gary Russell + * @since 4.2 + * + */ +public interface ConfigurableMetrics { + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ConfigurableMetricsAware.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ConfigurableMetricsAware.java new file mode 100644 index 0000000000..48547e8d44 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ConfigurableMetricsAware.java @@ -0,0 +1,29 @@ +/* + * 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.support.management; + +/** + * Classes implementing this interface can accept a {@link ConfigurableMetrics}. + * + * @author Gary Russell + * @since 4.2 + * + */ +public interface ConfigurableMetricsAware { + + void configureMetrics(M metrics); + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java index 6b424e2863..e8a2c6e503 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverage.java @@ -66,9 +66,18 @@ public class ExponentialMovingAverage { /** * Add a new measurement to the series. * - * @param value the measurement to append + * @param value the measurement to append (milliseconds) */ public synchronized void append(double value) { + appendNanos(value * 1000000); + } + + /** + * Add a new measurement to the series. + * + * @param value the measurement to append (nanoseconds) + */ + public synchronized void appendNanos(double value) { if (value > max || count == 0) { max = value; } @@ -76,7 +85,8 @@ public class ExponentialMovingAverage { min = value; } sum = decay * sum + value; - sumSquares = decay * sumSquares + value * value; + double valueMillis = value / 1000000.; + sumSquares = decay * sumSquares + valueMillis * valueMillis; weight = decay * weight + 1; count++;//NOSONAR - false positive, we're synchronized } @@ -96,9 +106,16 @@ public class ExponentialMovingAverage { } /** - * @return the mean value + * @return the mean value (milliseconds) */ public double getMean() { + return weight > 0 ? sum / weight / 1000000. : 0.; + } + + /** + * @return the mean value (nanoseconds) + */ + public double getMeanNanos() { return weight > 0 ? sum / weight : 0.; } @@ -112,24 +129,24 @@ public class ExponentialMovingAverage { } /** - * @return the maximum value recorded (not weighted) + * @return the maximum value recorded (not weighted, milliseconds). */ public double getMax() { - return max; + return max / 1000000.; } /** - * @return the minimum value recorded (not weighted) + * @return the minimum value recorded (not weighted, milliseconds). */ public double getMin() { - return min; + return min / 1000000.; } /** * @return summary statistics (count, mean, standard deviation etc.) */ public Statistics getStatistics() { - return new Statistics(count, min, max, getMean(), getStandardDeviation()); + return new Statistics(count, getMin(), getMax(), getMean(), getStandardDeviation()); } @Override diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java index 0c9def2063..3365cc4734 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRate.java @@ -44,7 +44,7 @@ public class ExponentialMovingAverageRate { private volatile double max; - private volatile long t0 = System.currentTimeMillis(); + private volatile long t0 = System.nanoTime(); private final double lapse; @@ -58,8 +58,8 @@ public class ExponentialMovingAverageRate { */ public ExponentialMovingAverageRate(double period, double lapsePeriod, int window) { rates = new ExponentialMovingAverage(window); - this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to milliseconds - this.period = period * 1000; // convert to milliseconds + this.lapse = lapsePeriod > 0 ? 0.000000001 / lapsePeriod : 0; // convert to nanoseconds + this.period = period * 1000000000; // convert to nanoseconds } @@ -68,7 +68,7 @@ public class ExponentialMovingAverageRate { max = 0; weight = 0; sum = 0; - t0 = System.currentTimeMillis(); + t0 = System.nanoTime(); rates.reset(); } @@ -76,15 +76,16 @@ public class ExponentialMovingAverageRate { * Add a new event to the series. */ public synchronized void increment() { - long t = System.currentTimeMillis(); - double value = t > t0 ? (t - t0) / period : 0; + long t = System.nanoTime(); + double delta = t - t0; + double value = delta > 0 ? delta / period : 0; if (value > max || getCount() == 0) { max = value; } if (value < min || getCount() == 0) { min = value; } - double alpha = Math.exp((t0 - t) * lapse); + double alpha = Math.exp(-delta * lapse); t0 = t; sum = alpha * sum + value; weight = alpha * weight + 1; @@ -110,7 +111,7 @@ public class ExponentialMovingAverageRate { * @return the time in seconds since the last measurement */ public double getTimeSinceLastMeasurement() { - return (System.currentTimeMillis() - t0) / 1000.; + return (System.nanoTime() - t0) / 1000000000.; } /** @@ -121,9 +122,9 @@ public class ExponentialMovingAverageRate { if (count == 0) { return 0; } - long t = System.currentTimeMillis(); - double value = t > t0 ? (t - t0) / period : 0; - return count / (count / rates.getMean() + value); + long delta = System.nanoTime() - t0; + double value = delta > 0 ? delta / period : 0; + return count / (count / rates.getMeanNanos() + value); } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java index ff4945da68..1ce9f6b01d 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/ExponentialMovingAverageRatio.java @@ -36,7 +36,7 @@ public class ExponentialMovingAverageRatio { private volatile double sum; - private volatile long t0 = System.currentTimeMillis(); + private volatile long t0 = System.nanoTime(); private final double lapse; @@ -57,12 +57,12 @@ public class ExponentialMovingAverageRatio { * Add a new event with successful outcome. */ public void success() { - append(1, System.currentTimeMillis()); + append(1, System.nanoTime()); } /** * Add a new event with successful outcome. - * @param t The current timestamp. + * @param t The current timestamp (System.nanoTime()). */ public void success(long t) { append(1, t); @@ -72,12 +72,12 @@ public class ExponentialMovingAverageRatio { * Add a new event with failed outcome. */ public void failure() { - append(0, System.currentTimeMillis()); + append(0, System.nanoTime()); } /** * Add a new event with failed outcome. - * @param t the current timestamp. + * @param t the current timestamp (System.nanoTime()). */ public void failure(long t) { append(0, t); @@ -86,16 +86,16 @@ public class ExponentialMovingAverageRatio { public synchronized void reset() { weight = 0; sum = 0; - t0 = System.currentTimeMillis(); + t0 = System.nanoTime(); cumulative.reset(); } private synchronized void append(int value, long t) { - double alpha = Math.exp((t0 - t) * lapse); + double alpha = Math.exp((t0 - t) / 1000000. * lapse); t0 = t; sum = alpha * sum + value; weight = alpha * weight + 1; - cumulative.append(sum / weight); + cumulative.appendNanos(sum / weight); } /** @@ -116,7 +116,7 @@ public class ExponentialMovingAverageRatio { * @return the time in seconds since the last measurement */ public double getTimeSinceLastMeasurement() { - return (System.currentTimeMillis() - t0) / 1000.; + return (System.nanoTime() - t0) / 1000000000.; } /** @@ -128,9 +128,9 @@ public class ExponentialMovingAverageRatio { // Optimistic to start: success rate is 100% return 1; } - long t = System.currentTimeMillis(); - double alpha = Math.exp((t0 - t) * lapse); - return alpha * cumulative.getMean() + 1 - alpha; + long t = System.nanoTime(); + double alpha = Math.exp((t0 - t) / 1000000. * lapse); + return alpha * cumulative.getMeanNanos() + 1 - alpha; } /** diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/MetricsContext.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/MetricsContext.java new file mode 100644 index 0000000000..7644d3b5af --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/MetricsContext.java @@ -0,0 +1,28 @@ +/* + * 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.support.management; + +/** + * Interface representing an opaque object containing state between initiating an + * event and concluding it. + * + * @author Gary Russell + * @since 4.2 + * + */ +public interface MetricsContext { + +} diff --git a/spring-integration-core/src/main/java/org/springframework/integration/support/management/package-info.java b/spring-integration-core/src/main/java/org/springframework/integration/support/management/package-info.java index da2da4414b..7ebb413ae8 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/support/management/package-info.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/support/management/package-info.java @@ -1,4 +1,4 @@ /** - * Provides classes related to management. + * Provides classes related to management support. */ package org.springframework.integration.support.management; diff --git a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java index 94a913a77d..10f3b37702 100644 --- a/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java +++ b/spring-integration-jms/src/main/java/org/springframework/integration/jms/PollableJmsChannel.java @@ -19,7 +19,6 @@ package org.springframework.integration.jms; 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.jms.core.JmsTemplate; import org.springframework.messaging.Message; @@ -46,16 +45,6 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh this.messageSelector = messageSelector; } - @Override - protected void initMetrics() { - setChannelMetrics(new ChannelReceiveMetrics(this.getComponentName())); - } - - @Override - protected ChannelReceiveMetrics getMetrics() { - return (ChannelReceiveMetrics) super.getMetrics(); - } - @Override public int getReceiveCount() { return getMetrics().getReceiveCount(); @@ -81,6 +70,7 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh ChannelInterceptorList interceptorList = getInterceptors(); Deque interceptorStack = null; boolean counted = false; + boolean countsEnabled = isCountsEnabled(); try { if (interceptorList.getInterceptors().size() > 0) { interceptorStack = new ArrayDeque(); @@ -100,7 +90,7 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh if (object == null) { return null; } - if (isCountsEnabled()) { + if (countsEnabled) { getMetrics().afterReceive(); counted = true; } @@ -118,7 +108,7 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh return message; } catch (RuntimeException e) { - if (isCountsEnabled() && !counted) { + if (countsEnabled && !counted) { getMetrics().afterError(); } if (interceptorStack != null) { diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java index 27865e8508..1353840187 100644 --- a/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/MBeanExporterParser.java @@ -60,6 +60,7 @@ public class MBeanExporterParser extends AbstractSingleBeanDefinitionParser { IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "managed-components", "componentNamePatterns"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "counts-enabled", "enabledCountsPatterns"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "stats-enabled", "enabledStatsPatterns"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "metrics-factory"); IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "object-naming-strategy", "namingStrategy"); builder.addPropertyValue("server", mbeanServer); diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DefaultMetricsFactory.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DefaultMetricsFactory.java new file mode 100644 index 0000000000..e59c3ef3ba --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DefaultMetricsFactory.java @@ -0,0 +1,44 @@ +/* + * 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.monitor; + +import org.springframework.integration.channel.management.AbstractMessageChannelMetrics; +import org.springframework.integration.channel.management.DefaultMessageChannelMetrics; +import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics; +import org.springframework.integration.handler.management.DefaultMessageHandlerMetrics; + + + +/** + * Default implementation. + * + * @author Gary Russell + * @since 4.2 + * + */ +public class DefaultMetricsFactory implements MetricsFactory { + + @Override + public AbstractMessageChannelMetrics createChannelMetrics(String name) { + return new DefaultMessageChannelMetrics(name); + } + + @Override + public AbstractMessageHandlerMetrics createHandlerMetrics(String name) { + return new DefaultMessageHandlerMetrics(name); + } + +} 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 41b6c11cc6..02874e0e2e 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 @@ -34,11 +34,8 @@ import javax.management.modelmbean.ModelMBean; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; -import org.springframework.aop.PointcutAdvisor; import org.springframework.aop.TargetSource; import org.springframework.aop.framework.Advised; -import org.springframework.aop.framework.ProxyFactory; -import org.springframework.aop.support.AopUtils; import org.springframework.beans.BeansException; import org.springframework.beans.annotation.AnnotationBeanUtils; import org.springframework.beans.factory.BeanFactory; @@ -52,6 +49,7 @@ import org.springframework.context.Lifecycle; import org.springframework.context.SmartLifecycle; import org.springframework.core.annotation.AnnotationUtils; import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.channel.management.AbstractMessageChannelMetrics; import org.springframework.integration.channel.management.MessageChannelMetrics; import org.springframework.integration.channel.management.PollableChannelManagement; import org.springframework.integration.context.IntegrationContextUtils; @@ -59,12 +57,14 @@ import org.springframework.integration.context.OrderlyShutdownCapable; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.endpoint.AbstractEndpoint; import org.springframework.integration.endpoint.management.MessageSourceMetrics; -import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; +import org.springframework.integration.handler.AbstractMessageProducingHandler; +import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics; import org.springframework.integration.handler.management.MessageHandlerMetrics; import org.springframework.integration.history.MessageHistoryConfigurer; import org.springframework.integration.history.TrackableComponent; import org.springframework.integration.router.MappingMessageRouterManagement; import org.springframework.integration.support.context.NamedComponent; +import org.springframework.integration.support.management.ConfigurableMetricsAware; import org.springframework.integration.support.management.IntegrationManagedResource; import org.springframework.integration.support.management.Statistics; import org.springframework.jmx.export.MBeanExporter; @@ -188,6 +188,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP private StringValueResolver embeddedValueResolver; + private MetricsFactory metricsFactory = new DefaultMetricsFactory(); + public IntegrationMBeanExporter() { super(); @@ -294,6 +296,15 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP this.embeddedValueResolver = resolver; } + /** + * Set a metrics factory. + * @param metricsFactory the factory. + * @since 4.2 + */ + public void setMetricsFactory(MetricsFactory metricsFactory) { + this.metricsFactory = metricsFactory; + } + @Override public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { @@ -333,7 +344,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP if (bean instanceof MessageProducer && bean instanceof Lifecycle) { Lifecycle target = (Lifecycle) extractTarget(bean); - if (!(target instanceof AbstractReplyProducingMessageHandler)) { // TODO: change to AMPMH + if (!(target instanceof AbstractMessageProducingHandler)) { this.inboundLifecycleMessageProducers.add(target); } } @@ -749,6 +760,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } } + @SuppressWarnings("unchecked") private void registerChannels() { for (MessageChannelMetrics monitor : channels) { String name = ((NamedComponent) monitor).getComponentName(); @@ -763,6 +775,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP if (name != null) { channelsByName.put(name, monitor); } + AbstractMessageChannelMetrics metrics = this.metricsFactory.createChannelMetrics(name); + Assert.state(metrics != null, "'metrics' must not be null"); Boolean enabled = smartMatch(this.enabledCountsPatterns, name); if (enabled != null) { monitor.enableCounts(enabled); @@ -770,12 +784,17 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP enabled = smartMatch(this.enabledStatsPatterns, name); if (enabled != null) { monitor.enableStats(enabled); + metrics.setFullStatsEnabled(enabled); + } + if (monitor instanceof ConfigurableMetricsAware) { + ((ConfigurableMetricsAware) monitor).configureMetrics(metrics); } registerBeanNameOrInstance(monitor, beanKey); } } } + @SuppressWarnings("unchecked") private void registerHandlers() { for (MessageHandlerMetrics handler : handlers) { MessageHandlerMetrics monitor = enhanceHandlerMonitor(handler); @@ -790,6 +809,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP if (name != null) { handlersByName.put(name, monitor); } + AbstractMessageHandlerMetrics metrics = this.metricsFactory.createHandlerMetrics(name); + Assert.state(metrics != null, "'metrics' must not be null"); Boolean enabled = smartMatch(this.enabledCountsPatterns, name); if (enabled != null) { monitor.enableCounts(enabled); @@ -797,6 +818,10 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP enabled = smartMatch(this.enabledStatsPatterns, name); if (enabled != null) { monitor.enableStats(enabled); + metrics.setFullStatsEnabled(enabled); + } + if (monitor instanceof ConfigurableMetricsAware) { + ((ConfigurableMetricsAware) monitor).configureMetrics(metrics); } registerBeanNameOrInstance(monitor, beanKey); } @@ -915,27 +940,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP } } - private Object applyAdvice(Object bean, PointcutAdvisor advisor, ClassLoader beanClassLoader) { - Class targetClass = AopUtils.getTargetClass(bean); - if (AopUtils.canApply(advisor.getPointcut(), targetClass)) { - if (bean instanceof Advised) { - ((Advised) bean).addAdvisor(advisor); - return bean; - } - else { - ProxyFactory proxyFactory = new ProxyFactory(bean); - proxyFactory.addAdvisor(advisor); - /** - * N.B. it's not a good idea to use proxyFactory.setProxyTargetClass(true) here because it forces all - * the integration components to be cglib proxyable (i.e. have a default constructor etc.), which they - * are not in general (usually for good reason). - */ - return proxyFactory.getProxy(beanClassLoader); - } - } - return bean; - } - private String getChannelBeanKey(String channel) { String name = "" + channel; if (name.startsWith("org.springframework.integration")) { diff --git a/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MetricsFactory.java b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MetricsFactory.java new file mode 100644 index 0000000000..3e9ebaef0b --- /dev/null +++ b/spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MetricsFactory.java @@ -0,0 +1,45 @@ +/* + * 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.monitor; + +import org.springframework.integration.channel.management.AbstractMessageChannelMetrics; +import org.springframework.integration.handler.management.AbstractMessageHandlerMetrics; + +/** + * Factories implementing this interface provide metric objects for message channels and + * message handlers. + * + * @author Gary Russell + * @since 4.2 + * + */ +public interface MetricsFactory { + + /** + * Factory method to create an {@link AbstractMessageChannelMetrics}. + * @param name the name. + * @return the metrics. + */ + AbstractMessageChannelMetrics createChannelMetrics(String name); + + /** + * Factory method to create an {@link AbstractMessageHandlerMetrics}. + * @param name the name. + * @return the metrics. + */ + AbstractMessageHandlerMetrics createHandlerMetrics(String name); + +} diff --git a/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-4.2.xsd b/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-4.2.xsd index eaff6381dd..f65f8490b4 100644 --- a/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-4.2.xsd +++ b/spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-4.2.xsd @@ -221,6 +221,19 @@ + + + + + + + + + A MetricsFactory responsible for creating objects that maintain metrics for message + channels and message handlers. + + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests-context.xml b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests-context.xml index c71cb21bff..4b28981162 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests-context.xml +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests-context.xml @@ -21,7 +21,8 @@ object-name-static-properties="appProperties" object-naming-strategy="keyNamer" counts-enabled="foo, !baz, ba*" - stats-enabled="fiz, buz" /> + stats-enabled="fiz, buz" + metrics-factory="mf" /> foo @@ -42,4 +43,6 @@ + + diff --git a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java index ae91779768..9e1048c203 100644 --- a/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java +++ b/spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/MBeanExporterParserTests.java @@ -33,6 +33,7 @@ import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationContext; import org.springframework.integration.channel.management.MessageChannelMetrics; import org.springframework.integration.monitor.IntegrationMBeanExporter; +import org.springframework.integration.monitor.MetricsFactory; import org.springframework.integration.test.util.TestUtils; import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.ContextConfiguration; @@ -82,6 +83,8 @@ public class MBeanExporterParserTests { metrics = context.getBean("buz", MessageChannelMetrics.class); assertTrue(metrics.isCountsEnabled()); assertTrue(metrics.isStatsEnabled()); + MetricsFactory factory = context.getBean(MetricsFactory.class); + assertSame(factory, TestUtils.getPropertyValue(exporter, "metricsFactory")); exporter.destroy(); }