Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

polishing

Browse Source
This commit is contained in:
Mark Fisher
2010-11-18 23:19:01 -05:00
parent 1f5dfff2e5
commit b3547433fb
18 changed files with 174 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java
View File

@@ -104,8 +104,7 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
@Override
public String getComponentType() {
// TODO: provide header: ("transport", "jmx");
return "notification-listener";
return "jmx:notification-listening-channel-adapter";
}
/**

1
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/OperationInvokingMessageHandler.java
View File

@@ -208,7 +208,6 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
return map;
}
@SuppressWarnings("rawtypes")
private Map<String, Object> createParameterMapFromList(List parameters) {
Map<String, Object> map = new HashMap<String, Object>();

9
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/OperationInvokingOutboundGatewayParser.java
View File

@@ -13,13 +13,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.jmx.config;
import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.w3c.dom.Element;
/**
* @author Oleg Zhurakousky
@@ -33,10 +35,9 @@ public class OperationInvokingOutboundGatewayParser extends AbstractConsumerEndp
}
@Override
protected BeanDefinitionBuilder parseHandler(Element element,
ParserContext parserContext) {
protected BeanDefinitionBuilder parseHandler(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition(
"org.springframework.integration.jmx.OperationInvokingMessageHandler");
"org.springframework.integration.jmx.OperationInvokingMessageHandler");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "server");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "object-name");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "operation-name");

16
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java
View File

@@ -10,6 +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;
import java.util.concurrent.atomic.AtomicInteger;
@@ -18,6 +19,7 @@ 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.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.jmx.export.annotation.ManagedResource;
@@ -29,6 +31,7 @@ import org.springframework.util.StopWatch;
*
* @author Dave Syer
* @author Helena Edelson
* @since 2.0
*/
@ManagedResource
public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMetrics {
@@ -41,6 +44,7 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe
public static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10;
private ExponentialMovingAverage sendDuration = new ExponentialMovingAverage(
DEFAULT_MOVING_AVERAGE_WINDOW);
@@ -59,10 +63,12 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe
private final String name;
public DirectChannelMetrics(String name) {
this.name = name;
}
public void destroy() {
if (logger.isDebugEnabled()) {
logger.debug(sendDuration);
@@ -87,15 +93,11 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe
return invocation.proceed();
}
private Object monitorSend(MethodInvocation invocation, MessageChannel channel, Message<?> message)
throws Throwable {
private Object monitorSend(MethodInvocation invocation, MessageChannel channel, Message<?> message) throws Throwable {
if (logger.isTraceEnabled()) {
logger.trace("Recording send on channel(" + channel + ") : message(" + message + ")");
}
final StopWatch timer = new StopWatch(channel + ".send:execution");
try {
timer.start();
@@ -108,13 +110,13 @@ public class DirectChannelMetrics implements MethodInterceptor, MessageChannelMe
if ((Boolean)result) {
sendSuccessRatio.success();
sendDuration.append(timer.getTotalTimeMillis());
} else {
}
else {
sendSuccessRatio.failure();
sendErrorCount.incrementAndGet();
sendErrorRate.increment();
}
return result;
}
catch (Throwable e) {
sendSuccessRatio.failure();

11
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverage.java
View File

@@ -10,6 +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;
/**
@@ -21,7 +22,7 @@ package org.springframework.integration.monitor;
* those trends can be approximately reflected.
*
* @author Dave Syer
*
* @since 2.0
*/
public class ExponentialMovingAverage {
@@ -39,6 +40,7 @@ public class ExponentialMovingAverage {
private final double decay;
/**
* Create a moving average accumulator with decay lapse window provided. Measurements older than this will have
* smaller weight than <code>1/e</code>.
@@ -49,6 +51,7 @@ public class ExponentialMovingAverage {
this.decay = 1 - 1. / window;
}
public synchronized void reset() {
weight = 0;
sum = 0;
@@ -64,10 +67,12 @@ public class ExponentialMovingAverage {
* @param value the measurement to append
*/
public synchronized void append(double value) {
if (value > max || count == 0)
if (value > max || count == 0) {
max = value;
if (value < min || count == 0)
}
if (value < min || count == 0) {
min = value;
}
sum = decay * sum + value;
sumSquares = decay * sumSquares + value * value;
weight = decay * weight + 1;

9
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRate.java
View File

@@ -10,6 +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;
/**
@@ -46,6 +47,7 @@ public class ExponentialMovingAverageRate {
private final double period;
/**
* @param period the period to base the rate measurement (in seconds)
* @param lapsePeriod the exponential lapse rate for the rate average (in seconds)
@@ -53,10 +55,11 @@ public class ExponentialMovingAverageRate {
*/
public ExponentialMovingAverageRate(double period, double lapsePeriod, int window) {
rates = new ExponentialMovingAverage(10);
this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs
this.period = period * 1000; // convert to millisecs
this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to milliseconds
this.period = period * 1000; // convert to milliseconds
}
public synchronized void reset() {
min = 0;
max = 0;
@@ -70,7 +73,6 @@ 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;
if (value > max || getCount() == 0) {
@@ -84,7 +86,6 @@ public class ExponentialMovingAverageRate {
sum = alpha * sum + value;
weight = alpha * weight + 1;
rates.append(sum > 0 ? weight / sum : 0);
}
/**

7
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/ExponentialMovingAverageRatio.java
View File

@@ -10,6 +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;
/**
@@ -24,7 +25,7 @@ package org.springframework.integration.monitor;
* the lapse window and <code>i</code> is the sequence number of the measurement.</li>
*
* @author Dave Syer
*
* @since 2.0
*/
public class ExponentialMovingAverageRatio {
@@ -38,6 +39,7 @@ public class ExponentialMovingAverageRatio {
private final ExponentialMovingAverage cumulative;
/**
* @param lapsePeriod the exponential lapse rate for the rate average (in seconds)
* @param window the exponential lapse window (number of measurements)
@@ -47,6 +49,7 @@ public class ExponentialMovingAverageRatio {
this.lapse = lapsePeriod > 0 ? 0.001 / lapsePeriod : 0; // convert to millisecs
}
/**
* Add a new event with successful outcome.
*/
@@ -69,14 +72,12 @@ public class ExponentialMovingAverageRatio {
}
private synchronized void append(int value) {
long t = System.currentTimeMillis();
double alpha = Math.exp((t0 - t) * lapse);
t0 = t;
sum = alpha * sum + value;
weight = alpha * weight + 1;
cumulative.append(sum / weight);
}
/**

46
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2009-2010 the original author or authors.
* Copyright 2002-2010 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,6 +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;
import java.lang.reflect.Field;
@@ -82,13 +83,14 @@ import org.springframework.util.ReflectionUtils;
* @author Oleg Zhurakousky
*/
@ManagedResource
public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostProcessor, BeanFactoryAware,
BeanClassLoaderAware, SmartLifecycle {
public class IntegrationMBeanExporter extends MBeanExporter
implements BeanPostProcessor, BeanFactoryAware, BeanClassLoaderAware, SmartLifecycle {
private static final Log logger = LogFactory.getLog(IntegrationMBeanExporter.class);
public static final String DEFAULT_DOMAIN = "org.springframework.integration";
private final AnnotationJmxAttributeSource attributeSource = new AnnotationJmxAttributeSource();
private ListableBeanFactory beanFactory;
@@ -123,6 +125,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
private final Map<String, String> objectNameStaticProperties = new HashMap<String, String>();
public IntegrationMBeanExporter() {
super();
// Shouldn't be necessary, but to be on the safe side...
@@ -131,6 +134,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
setAssembler(new MetadataMBeanInfoAssembler(attributeSource));
}
@Override
public void setBeanClassLoader(ClassLoader classLoader) {
this.beanClassLoader = classLoader;
@@ -163,7 +167,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof Advised) {
for (Advisor advisor : ((Advised) bean).getAdvisors()) {
Advice advice = advisor.getAdvice();
@@ -194,19 +197,19 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
Object target = extractTarget(bean);
if (target instanceof QueueChannel) {
monitor = new QueueChannelMetrics((QueueChannel) target, beanName);
} else {
}
else {
monitor = new PollableChannelMetrics(beanName);
}
} else {
}
else {
monitor = new DirectChannelMetrics(beanName);
}
Object advised = applyChannelInterceptor(bean, monitor, beanClassLoader);
channels.add(monitor);
return advised;
}
return bean;
}
public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
@@ -215,7 +218,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
@Override
protected void registerBeans() {
// Completely disable sup class registration to avoid duplicates
// Completely disable super class registration to avoid duplicates
}
public final boolean isAutoStartup() {
@@ -230,7 +233,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
this.lifecycleLock.lock();
try {
return this.running;
} finally {
}
finally {
this.lifecycleLock.unlock();
}
}
@@ -245,7 +249,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
logger.info("started " + this);
}
}
} finally {
}
finally {
this.lifecycleLock.unlock();
}
}
@@ -260,7 +265,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
logger.info("stopped " + this);
}
}
} finally {
}
finally {
this.lifecycleLock.unlock();
}
}
@@ -270,7 +276,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
try {
this.stop();
callback.run();
} finally {
}
finally {
this.lifecycleLock.unlock();
}
}
@@ -450,7 +457,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
try {
return extractTarget(advised.getTargetSource().getTarget());
} catch (Exception e) {
}
catch (Exception e) {
logger.error("Could not extract target", e);
return null;
}
@@ -462,7 +470,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
if (bean instanceof Advised) {
((Advised) bean).addAdvisor(advisor);
return bean;
} else {
}
else {
ProxyFactory proxyFactory = new ProxyFactory(bean);
proxyFactory.addAdvisor(advisor);
return proxyFactory.getProxy(beanClassLoader);
@@ -522,7 +531,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
Object field = null;
try {
field = extractTarget(getField(endpoint, "handler"));
} catch (Exception e) {
}
catch (Exception e) {
logger.trace("Could not get handler from bean = " + beanName);
}
if (field == monitor.getMessageHandler()) {
@@ -541,7 +551,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
if (targetSource != null) {
try {
target = targetSource.getTarget();
} catch (Exception e) {
}
catch (Exception e) {
logger.debug("Could not get handler from bean = " + name);
}
}
@@ -659,7 +670,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
monitor.setName(name);
return result;
}
private static Object getField(Object target, String name) {

33
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageHandlerMetrics.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.monitor;
import org.springframework.context.Lifecycle;
@@ -25,9 +26,7 @@ import org.springframework.jmx.export.annotation.ManagedResource;
* be used to stop and start polling endpoints, for instance, in a live system.
*
* @author Dave Syer
*
* @since 2.0
*
*/
@ManagedResource
public class LifecycleMessageHandlerMetrics implements MessageHandlerMetrics, Lifecycle {
@@ -36,68 +35,70 @@ public class LifecycleMessageHandlerMetrics implements MessageHandlerMetrics, Li
private final MessageHandlerMetrics delegate;
public LifecycleMessageHandlerMetrics(Lifecycle lifecycle, MessageHandlerMetrics delegate) {
this.lifecycle = lifecycle;
this.delegate = delegate;
}
@ManagedAttribute
public boolean isRunning() {
return lifecycle.isRunning();
return this.lifecycle.isRunning();
}
@ManagedOperation
public void start() {
lifecycle.start();
this.lifecycle.start();
}
@ManagedOperation
public void stop() {
lifecycle.stop();
this.lifecycle.stop();
}
public void reset() {
delegate.reset();
this.delegate.reset();
}
public int getErrorCount() {
return delegate.getErrorCount();
return this.delegate.getErrorCount();
}
public int getHandleCount() {
return delegate.getHandleCount();
return this.delegate.getHandleCount();
}
public double getMaxDuration() {
return delegate.getMaxDuration();
return this.delegate.getMaxDuration();
}
public double getMeanDuration() {
return delegate.getMeanDuration();
return this.delegate.getMeanDuration();
}
public double getMinDuration() {
return delegate.getMinDuration();
return this.delegate.getMinDuration();
}
public double getStandardDeviationDuration() {
return delegate.getStandardDeviationDuration();
return this.delegate.getStandardDeviationDuration();
}
public Statistics getDuration() {
return delegate.getDuration();
return this.delegate.getDuration();
}
public String getName() {
return delegate.getName();
return this.delegate.getName();
}
public String getSource() {
return delegate.getSource();
return this.delegate.getSource();
}
public int getActiveCount() {
return delegate.getActiveCount();
return this.delegate.getActiveCount();
}
}

21
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/LifecycleMessageSourceMetrics.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.monitor;
import org.springframework.context.Lifecycle;
@@ -22,12 +23,10 @@ import org.springframework.jmx.export.annotation.ManagedResource;
/**
* A {@link MessageSourceMetrics} that exposes in addition the {@link Lifecycle} interface. The lifecycle methods can
* be used to stop and start polling endpoints, for instance, in a live system.
* be used to start and stop polling endpoints, for instance, in a live system.
*
* @author Dave Syer
*
* @since 2.0
*
*/
@ManagedResource
public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Lifecycle {
@@ -36,37 +35,39 @@ public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Life
private final MessageSourceMetrics delegate;
public LifecycleMessageSourceMetrics(Lifecycle lifecycle, MessageSourceMetrics delegate) {
this.lifecycle = lifecycle;
this.delegate = delegate;
}
@ManagedOperation
public void reset() {
delegate.reset();
this.delegate.reset();
}
@ManagedAttribute
public boolean isRunning() {
return lifecycle.isRunning();
return this.lifecycle.isRunning();
}
@ManagedOperation
public void start() {
lifecycle.start();
this.lifecycle.start();
}
@ManagedOperation
public void stop() {
lifecycle.stop();
this.lifecycle.stop();
}
public String getName() {
return delegate.getName();
return this.delegate.getName();
}
public String getSource() {
return delegate.getSource();
return this.delegate.getSource();
}
/**
@@ -74,7 +75,7 @@ public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Life
* @see org.springframework.integration.monitor.MessageSourceMetrics#getMessageCount()
*/
public int getMessageCount() {
return delegate.getMessageCount();
return this.delegate.getMessageCount();
}
}

5
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageChannelMetrics.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.monitor;
import org.springframework.jmx.export.annotation.ManagedMetric;
@@ -24,9 +25,7 @@ import org.springframework.jmx.support.MetricType;
* channel types.
*
* @author Dave Syer
*
* @since 2.0
*
*/
public interface MessageChannelMetrics {
@@ -108,4 +107,4 @@ public interface MessageChannelMetrics {
*/
Statistics getErrorRate();
}
}

11
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageHandlerMetrics.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.monitor;
import org.springframework.jmx.export.annotation.ManagedMetric;
@@ -21,7 +22,6 @@ import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
*
* @since 2.0
*/
public interface MessageHandlerMetrics {
@@ -42,7 +42,7 @@ public interface MessageHandlerMetrics {
int getErrorCount();
/**
* @return the maximum handler duration (milliseconds)
* @return the mean handler duration (milliseconds)
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Mean Duration in Milliseconds")
double getMeanDuration();
@@ -54,11 +54,14 @@ public interface MessageHandlerMetrics {
double getMinDuration();
/**
* @return the standard deviation handler duration (milliseconds)
* @return the maximum handler duration (milliseconds)
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Max Duration in Milliseconds")
double getMaxDuration();
/**
* @return the standard deviation handler duration (milliseconds)
*/
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "Handler Standard Deviation Duration in Milliseconds")
double getStandardDeviationDuration();
@@ -74,4 +77,4 @@ public interface MessageHandlerMetrics {
String getSource();
}
}

1
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/MessageSourceMetrics.java
View File

@@ -19,7 +19,6 @@ import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
*
* @since 2.0
*/
public interface MessageSourceMetrics {

25
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java
View File

@@ -13,11 +13,13 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.monitor;
import java.util.concurrent.atomic.AtomicInteger;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.integration.MessageChannel;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
@@ -25,9 +27,7 @@ import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
public class PollableChannelMetrics extends DirectChannelMetrics {
@@ -35,6 +35,7 @@ public class PollableChannelMetrics extends DirectChannelMetrics {
private final AtomicInteger receiveErrorCount = new AtomicInteger();
/**
* @param name
*/
@@ -42,6 +43,7 @@ public class PollableChannelMetrics extends DirectChannelMetrics {
super(name);
}
@Override
protected Object doInvoke(MethodInvocation invocation, String method, MessageChannel channel) throws Throwable {
if ("receive".equals(method)) {
@@ -56,14 +58,13 @@ public class PollableChannelMetrics extends DirectChannelMetrics {
}
try {
Object object = invocation.proceed();
if (object!=null) {
receiveCount.incrementAndGet();
if (object != null) {
this.receiveCount.incrementAndGet();
}
return object;
}
catch (Throwable e) {
receiveErrorCount.incrementAndGet();
this.receiveErrorCount.incrementAndGet();
throw e;
}
}
@@ -71,24 +72,24 @@ public class PollableChannelMetrics extends DirectChannelMetrics {
@ManagedOperation
public synchronized void reset() {
super.reset();
receiveErrorCount.set(0);
receiveCount.set(0);
this.receiveErrorCount.set(0);
this.receiveCount.set(0);
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Count")
public int getReceiveCount() {
return receiveCount.get();
return this.receiveCount.get();
}
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "MessageChannel Receive Error Count")
public int getReceiveErrorCount() {
return receiveErrorCount.get();
return this.receiveErrorCount.get();
}
@Override
public String toString() {
return String.format("MessageChannelMonitor: [name=%s, sends=%d, receives=%d]", getName(), getSendCount(),
receiveCount.get());
return String.format("MessageChannelMonitor: [name=%s, sends=%d, receives=%d]",
getName(), getSendCount(), this.receiveCount.get());
}
}

6
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/QueueChannelMetrics.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.monitor;
import org.springframework.integration.channel.QueueChannel;
@@ -21,15 +22,15 @@ import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
public class QueueChannelMetrics extends PollableChannelMetrics {
private final QueueChannel channel;
/**
* @param channel
* @param name
*/
public QueueChannelMetrics(QueueChannel channel, String name) {
@@ -37,6 +38,7 @@ public class QueueChannelMetrics extends PollableChannelMetrics {
this.channel = channel;
}
@ManagedMetric(metricType = MetricType.GAUGE, displayName = "QueueChannel Queue Size")
public int getQueueSize() {
return channel.getQueueSize();

82
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.monitor;
import java.util.concurrent.atomic.AtomicInteger;
@@ -21,19 +22,16 @@ 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.Message;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessageRejectedException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.util.StopWatch;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
@ManagedResource
public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHandlerMetrics {
@@ -42,6 +40,7 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa
private static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10;
private final MessageHandler handler;
private final AtomicInteger activeCount = new AtomicInteger();
@@ -50,23 +49,24 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa
private final AtomicInteger errorCount = new AtomicInteger();
private final ExponentialMovingAverage duration = new ExponentialMovingAverage(
DEFAULT_MOVING_AVERAGE_WINDOW);
private final ExponentialMovingAverage duration = new ExponentialMovingAverage(DEFAULT_MOVING_AVERAGE_WINDOW);
private String name;
private volatile String name;
private volatile String source;
private String source;
public SimpleMessageHandlerMetrics(MessageHandler handler) {
this.handler = handler;
}
public void setName(String name) {
this.name = name;
}
public String getName() {
return name;
return this.name;
}
public void setSource(String source) {
@@ -78,7 +78,7 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa
}
public MessageHandler getMessageHandler() {
return handler;
return this.handler;
}
public Object invoke(MethodInvocation invocation) throws Throwable {
@@ -91,77 +91,77 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa
return invocation.proceed();
}
private void handleMessage(Message<?> message) throws MessageRejectedException, MessageHandlingException,
MessageDeliveryException {
private void handleMessage(Message<?> message) throws MessagingException {
if (logger.isTraceEnabled()) {
logger.trace("messageHandler(" + handler + ") message(" + message + ") :");
logger.trace("messageHandler(" + this.handler + ") message(" + message + ") :");
}
String name = this.name;
if (name == null) {
name = handler.toString();
name = this.handler.toString();
}
StopWatch timer = new StopWatch(name + ".handle:execution");
try {
timer.start();
handleCount.incrementAndGet();
activeCount.incrementAndGet();
this.handleCount.incrementAndGet();
this.activeCount.incrementAndGet();
handler.handleMessage(message);
this.handler.handleMessage(message);
timer.stop();
duration.append(timer.getTotalTimeMillis());
} catch (RuntimeException e) {
errorCount.incrementAndGet();
this.duration.append(timer.getTotalTimeMillis());
}
catch (RuntimeException e) {
this.errorCount.incrementAndGet();
throw e;
} catch (Error e) {
errorCount.incrementAndGet();
}
catch (Error e) {
this.errorCount.incrementAndGet();
throw e;
} finally {
activeCount.decrementAndGet();
}
finally {
this.activeCount.decrementAndGet();
}
}
public synchronized void reset() {
duration.reset();
errorCount.set(0);
handleCount.set(0);
this.duration.reset();
this.errorCount.set(0);
this.handleCount.set(0);
}
public int getHandleCount() {
if (logger.isTraceEnabled()) {
logger.trace("Getting Handle Count:" + this);
}
return handleCount.get();
return this.handleCount.get();
}
public int getErrorCount() {
return errorCount.get();
return this.errorCount.get();
}
public double getMeanDuration() {
return duration.getMean();
return this.duration.getMean();
}
public double getMinDuration() {
return duration.getMin();
return this.duration.getMin();
}
public double getMaxDuration() {
return duration.getMax();
return this.duration.getMax();
}
public double getStandardDeviationDuration() {
return duration.getStandardDeviation();
return this.duration.getStandardDeviation();
}
public int getActiveCount() {
return activeCount.get();
return this.activeCount.get();
}
public Statistics getDuration() {
return duration.getStatistics();
return this.duration.getStatistics();
}
@Override
@@ -169,4 +169,4 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa
return String.format("MessageHandlerMonitor: [name=%s, source=%s, duration=%s]", name, source, duration);
}
}
}

18
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageSourceMetrics.java
View File

@@ -21,7 +21,6 @@ import org.springframework.integration.core.MessageSource;
/**
* @author Dave Syer
*
* @since 2.0
*/
public class SimpleMessageSourceMetrics implements MethodInterceptor, MessageSourceMetrics {
@@ -30,21 +29,22 @@ public class SimpleMessageSourceMetrics implements MethodInterceptor, MessageSou
private final MessageSource<?> messageSource;
private String source;
private volatile String source;
private volatile String name;
private String name;
public SimpleMessageSourceMetrics(MessageSource<?> messageSource) {
this.messageSource = messageSource;
}
public void setName(String name) {
this.name = name;
}
public String getName() {
return name;
return this.name;
}
public void setSource(String source) {
@@ -56,22 +56,22 @@ public class SimpleMessageSourceMetrics implements MethodInterceptor, MessageSou
}
public MessageSource<?> getMessageSource() {
return messageSource;
return this.messageSource;
}
public void reset() {
messageCount.set(0);
this.messageCount.set(0);
}
public int getMessageCount() {
return messageCount.get();
return this.messageCount.get();
}
public Object invoke(MethodInvocation invocation) throws Throwable {
String method = invocation.getMethod().getName();
Object result = invocation.proceed();
if ("receive".equals(method) && result!=null) {
messageCount.incrementAndGet();
this.messageCount.incrementAndGet();
}
return result;
}

18
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/Statistics.java
View File

@@ -13,25 +13,26 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.monitor;
/**
* @author Dave Syer
*
* @since 2.0
*
*/
public class Statistics {
private final int count;
private final double min;
private final double max;
private final double mean;
private final double standardDeviation;
/**
*
*/
public Statistics(int count, double min, double max, double mean, double standardDeviation) {
this.count = count;
this.min = min;
@@ -40,6 +41,7 @@ public class Statistics {
this.standardDeviation = standardDeviation;
}
public int getCount() {
return count;
}
@@ -62,8 +64,8 @@ public class Statistics {
@Override
public String toString() {
return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]", count, min, max, getMean(),
getStandardDeviation());
return String.format("[N=%d, min=%f, max=%f, mean=%f, sigma=%f]",
count, min, max, getMean(), getStandardDeviation());
}
}