INT-3637: JMX Improvements Msg Sources/Handlers

JIRA: https://jira.spring.io/browse/INT-3637

- Message Sources
- Message Handlers

Polishing; Final Review and PR Comments

- avoid second call to System.currentTimeMillis()
- fix elapsed time in handler metrics
- expose TrackableComponent when available

INT-3637: Add @IntergrationManagedResource

Prevents beans (channels, etc), which were previously picked
up via a proxy, from being exported by a standard context
MBeanExporter.

I looked at completely eliminating the MBeanExporterHelper,
which suppresses o.s.integration beans from being exported by
a standard MBeanExporter when there is an IMBE present, but I feel
this is too much of a breaking change. There are a number of
standard beans (such as WireTap) that are @ManagedResources and
these would disappear for users that don't have an IMBE.

That said, such beans previously disappear completely when there
*is* an IMBE so now they are now annotated with both so that
they are exported by at most one of the exporters.

Polishing; Use MBE.addExludedBean instead of DFA

INT-3639: JMX (AMQP/JMS) Channel Stats

JIRA: https://jira.spring.io/browse/INT-3639

Module channels inherit `ChannelSendMetrics`; add
`PollableChannelManagement` for polled channels.

INT-3638: JMX Initial Stats/Counts Settings

JIRA: https://jira.spring.io/browse/INT-3638

Add the ability to specify the initial settings for
'enableStats' and 'enableCounts' for MBeans that
support those statistics.

Polishing; PR Comments; Add Handler Metrics Test

Also found a problem using an inner MessageSource bean in an
inbound-channel-adapter - a BeanComponentDefinition was returned
instead of a BeanDefinition.

Add @DirtiesContext to JMX Tests

The new MonitorTests failed with InstanceAlreadyExistsException for the
errorChannel.

Add @DirtiesContext to all tests using the Spring test runner to avoid caching any
contexts after the test class completes.

INT-3637: JMX Support Negated Name Match Patterns

Note: with negated matches, order matters.

Polishing for `MonitorTests`
Fix `EnableIntegrationMBeanExport` JavaDocs
This commit is contained in:
Gary Russell
2015-02-19 21:49:20 +02:00
committed by Artem Bilan
parent b56a56e8e6
commit 09c203dc6d
85 changed files with 1762 additions and 650 deletions

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -24,6 +24,8 @@ 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;
import org.springframework.messaging.support.ChannelInterceptor;
@@ -36,9 +38,11 @@ import org.springframework.util.Assert;
*
* @author Mark Fisher
* @author Artem Bilan
* @author Gary Russell
* @since 2.1
*/
public class PollableAmqpChannel extends AbstractAmqpChannel implements PollableChannel {
public class PollableAmqpChannel extends AbstractAmqpChannel implements PollableChannel,
PollableChannelManagement {
private final String channelName;
@@ -54,6 +58,16 @@ 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
@@ -76,8 +90,34 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
this.amqpAdmin = amqpAdmin;
}
@Override
public int getReceiveCount() {
return getMetrics().getReceiveCount();
}
@Override
public long getReceiveCountLong() {
return getMetrics().getReceiveCountLong();
}
@Override
public int getReceiveErrorCount() {
return getMetrics().getReceiveErrorCount();
}
@Override
public long getReceiveErrorCountLong() {
return getMetrics().getReceiveErrorCountLong();
}
@Override
protected String getRoutingKey() {
return this.queueName;
}
@Override
protected void onInit() throws Exception {
super.onInit();
AmqpTemplate amqpTemplate = this.getAmqpTemplate();
if (this.queueName == null) {
if (this.amqpAdmin == null && amqpTemplate instanceof RabbitTemplate) {
@@ -91,15 +131,11 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
}
}
@Override
protected String getRoutingKey() {
return this.queueName;
}
@Override
public Message<?> receive() {
ChannelInterceptorList interceptorList = getInterceptors();
Deque<ChannelInterceptor> interceptorStack = null;
boolean counted = false;
try {
if (interceptorList.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -112,6 +148,10 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
if (object == null) {
return null;
}
if (isCountsEnabled()) {
getMetrics().afterReceive();
counted = true;
}
Message<?> message = null;
if (object instanceof Message<?>) {
message = (Message<?>) object;
@@ -126,6 +166,9 @@ public class PollableAmqpChannel extends AbstractAmqpChannel implements Pollable
return message;
}
catch (RuntimeException e) {
if (isCountsEnabled() && !counted) {
getMetrics().afterError();
}
if (interceptorStack != null) {
interceptorList.afterReceiveCompletion(null, this, e, interceptorStack);
}

View File

@@ -33,8 +33,8 @@ 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.IntegrationManagedResource;
import org.springframework.integration.support.management.Statistics;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageDeliveryException;
@@ -42,7 +42,6 @@ import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.ChannelInterceptor;
import org.springframework.util.Assert;
import org.springframework.util.StopWatch;
import org.springframework.util.StringUtils;
/**
@@ -56,7 +55,7 @@ import org.springframework.util.StringUtils;
* @author Gary Russell
* @author Artem Bilan
*/
@ManagedResource
@IntegrationManagedResource
public abstract class AbstractMessageChannel extends IntegrationObjectSupport
implements MessageChannel, TrackableComponent, ChannelInterceptorAware, MessageChannelMetrics {
@@ -72,6 +71,8 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
private volatile MessageConverter messageConverter;
private volatile boolean countsEnabled;
private volatile boolean statsEnabled;
private volatile ChannelSendMetrics channelMetrics;
@@ -86,9 +87,28 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
this.shouldTrack = shouldTrack;
}
@Override
public void enableCounts(boolean countsEnabled) {
this.countsEnabled = countsEnabled;
if (!countsEnabled) {
this.statsEnabled = false;
}
}
@Override
public boolean isCountsEnabled() {
return countsEnabled;
}
@Override
public void enableStats(boolean statsEnabled) {
if (statsEnabled) {
this.countsEnabled = true;
}
this.statsEnabled = statsEnabled;
if (this.channelMetrics != null) {
this.channelMetrics.setFullStatsEnabled(statsEnabled);
}
}
@Override
@@ -314,6 +334,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
protected void setChannelMetrics(ChannelSendMetrics channelMetrics) {
this.channelMetrics = channelMetrics;
this.channelMetrics.setFullStatsEnabled(this.statsEnabled);
}
/**
@@ -373,7 +394,7 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
Deque<ChannelInterceptor> interceptorStack = null;
boolean sent = false;
boolean statsProcessed = false;
StopWatch timer = null;
long start = 0;
try {
if (this.datatypes.length > 0) {
message = this.convertPayloadIfNecessary(message);
@@ -385,12 +406,12 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
return false;
}
}
if (this.statsEnabled) {
timer = this.channelMetrics.beforeSend();
if (this.countsEnabled) {
start = this.channelMetrics.beforeSend();
}
sent = this.doSend(message, timeout);
if (this.statsEnabled) {
this.channelMetrics.afterSend(timer, sent);
if (this.countsEnabled) {
this.channelMetrics.afterSend(start, sent);
statsProcessed = true;
}
this.interceptors.postSend(message, this, sent);
@@ -400,8 +421,8 @@ public abstract class AbstractMessageChannel extends IntegrationObjectSupport
return sent;
}
catch (Exception e) {
if (this.statsEnabled && !statsProcessed) {
this.channelMetrics.afterSend(timer, false);
if (this.countsEnabled && !statsProcessed) {
this.channelMetrics.afterSend(start, false);
}
if (interceptorStack != null) {
this.interceptors.afterSendCompletion(message, this, sent, e, interceptorStack);

View File

@@ -95,6 +95,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
public final Message<?> receive(long timeout) {
ChannelInterceptorList interceptorList = getInterceptors();
Deque<ChannelInterceptor> interceptorStack = null;
boolean counted = false;
try {
if (interceptorList.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -104,8 +105,9 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
}
}
Message<?> message = this.doReceive(timeout);
if (isStatsEnabled()) {
if (isCountsEnabled()) {
getMetrics().afterReceive();
counted = true;
}
message = interceptorList.postReceive(message, this);
if (interceptorStack != null) {
@@ -114,7 +116,7 @@ public abstract class AbstractPollableChannel extends AbstractMessageChannel imp
return message;
}
catch (RuntimeException e) {
if (isStatsEnabled()) {
if (isCountsEnabled() && !counted) {
getMetrics().afterError();
}
if (interceptorStack != null) {

View File

@@ -23,8 +23,8 @@ import org.springframework.beans.factory.BeanNameAware;
import org.springframework.integration.channel.management.ChannelSendMetrics;
import org.springframework.integration.channel.management.MessageChannelMetrics;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.integration.support.management.Statistics;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.util.StringUtils;
@@ -38,12 +38,14 @@ import org.springframework.util.StringUtils;
* @author Mark Fisher
* @author Gary Russell
*/
@ManagedResource
@IntegrationManagedResource
public class NullChannel implements PollableChannel, MessageChannelMetrics, BeanNameAware, NamedComponent {
private final Log logger = LogFactory.getLog(this.getClass());
private volatile ChannelSendMetrics metrics = new ChannelSendMetrics("nullChannel");
private volatile ChannelSendMetrics channelMetrics = new ChannelSendMetrics("nullChannel");
private volatile boolean countsEnabled;
private volatile boolean statsEnabled;
@@ -52,7 +54,7 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics, Bean
@Override
public void setBeanName(String beanName) {
this.beanName = beanName;
this.metrics = new ChannelSendMetrics(getComponentName());
this.channelMetrics = new ChannelSendMetrics(getComponentName());
}
@Override
@@ -67,12 +69,29 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics, Bean
@Override
public void reset() {
this.metrics.reset();
this.channelMetrics.reset();
}
@Override
public void enableCounts(boolean countsEnabled) {
this.countsEnabled = countsEnabled;
if (!countsEnabled) {
this.statsEnabled = false;
}
}
@Override
public boolean isCountsEnabled() {
return this.countsEnabled;
}
@Override
public void enableStats(boolean statsEnabled) {
if (statsEnabled) {
this.countsEnabled = true;
}
this.statsEnabled = statsEnabled;
this.channelMetrics.setFullStatsEnabled(statsEnabled);
}
@Override
@@ -82,77 +101,77 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics, Bean
@Override
public int getSendCount() {
return this.metrics.getSendCount();
return this.channelMetrics.getSendCount();
}
@Override
public long getSendCountLong() {
return this.metrics.getSendCountLong();
return this.channelMetrics.getSendCountLong();
}
@Override
public int getSendErrorCount() {
return this.metrics.getSendErrorCount();
return this.channelMetrics.getSendErrorCount();
}
@Override
public long getSendErrorCountLong() {
return this.metrics.getSendErrorCountLong();
return this.channelMetrics.getSendErrorCountLong();
}
@Override
public double getTimeSinceLastSend() {
return this.metrics.getTimeSinceLastSend();
return this.channelMetrics.getTimeSinceLastSend();
}
@Override
public double getMeanSendRate() {
return this.metrics.getMeanSendRate();
return this.channelMetrics.getMeanSendRate();
}
@Override
public double getMeanErrorRate() {
return this.metrics.getMeanErrorRate();
return this.channelMetrics.getMeanErrorRate();
}
@Override
public double getMeanErrorRatio() {
return this.metrics.getMeanErrorRatio();
return this.channelMetrics.getMeanErrorRatio();
}
@Override
public double getMeanSendDuration() {
return this.metrics.getMeanSendDuration();
return this.channelMetrics.getMeanSendDuration();
}
@Override
public double getMinSendDuration() {
return this.metrics.getMinSendDuration();
return this.channelMetrics.getMinSendDuration();
}
@Override
public double getMaxSendDuration() {
return this.metrics.getMaxSendDuration();
return this.channelMetrics.getMaxSendDuration();
}
@Override
public double getStandardDeviationSendDuration() {
return this.metrics.getStandardDeviationSendDuration();
return this.channelMetrics.getStandardDeviationSendDuration();
}
@Override
public Statistics getSendDuration() {
return this.metrics.getSendDuration();
return this.channelMetrics.getSendDuration();
}
@Override
public Statistics getSendRate() {
return this.metrics.getSendRate();
return this.channelMetrics.getSendRate();
}
@Override
public Statistics getErrorRate() {
return this.metrics.getErrorRate();
return this.channelMetrics.getErrorRate();
}
@Override
@@ -160,8 +179,8 @@ public class NullChannel implements PollableChannel, MessageChannelMetrics, Bean
if (logger.isDebugEnabled()) {
logger.debug("message sent to null channel: " + message);
}
if (this.statsEnabled) {
this.metrics.afterSend(this.metrics.beforeSend(), true);
if (this.countsEnabled) {
this.channelMetrics.afterSend(this.channelMetrics.beforeSend(), true);
}
return true;
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,6 +22,7 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelInterceptorAware;
import org.springframework.integration.core.MessageSelector;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
@@ -39,6 +40,7 @@ import org.springframework.util.Assert;
* @author Gary Russell
*/
@ManagedResource
@IntegrationManagedResource
public class WireTap extends ChannelInterceptorAdapter implements Lifecycle, VetoCapableInterceptor {
private static final Log logger = LogFactory.getLog(WireTap.class);

View File

@@ -22,8 +22,6 @@ import org.springframework.integration.support.management.ExponentialMovingAvera
import org.springframework.integration.support.management.ExponentialMovingAverageRate;
import org.springframework.integration.support.management.ExponentialMovingAverageRatio;
import org.springframework.integration.support.management.Statistics;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.util.StopWatch;
/**
* Registers all message channels, and accumulates statistics about their performance. The statistics are then published
@@ -34,7 +32,6 @@ import org.springframework.util.StopWatch;
* @author Gary Russell
* @since 2.0
*/
@ManagedResource
public class ChannelSendMetrics {
protected final Log logger = LogFactory.getLog(getClass());
@@ -63,11 +60,16 @@ public class ChannelSendMetrics {
private final String name;
private volatile boolean fullStatsEnabled;
public ChannelSendMetrics(String name) {
this.name = name;
}
public void setFullStatsEnabled(boolean fullStatsEnabled) {
this.fullStatsEnabled = fullStatsEnabled;
}
public void destroy() {
if (logger.isDebugEnabled()) {
@@ -79,33 +81,38 @@ public class ChannelSendMetrics {
return name;
}
public StopWatch beforeSend() {
public long beforeSend() {
if (logger.isTraceEnabled()) {
logger.trace("Recording send on channel(" + this.name + ")");
}
final StopWatch timer = new StopWatch(this.name + ".send:execution");
timer.start();
long start = 0;
if (this.fullStatsEnabled) {
start = System.currentTimeMillis();
this.sendRate.increment();
}
this.sendCount.incrementAndGet();
this.sendRate.increment();
return timer;
return start;
}
public void afterSend(StopWatch timer, boolean result) {
if (timer != null) {
timer.stop();
if (result) {
sendSuccessRatio.success();
sendDuration.append(timer.getTotalTimeMillis());
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 {
sendSuccessRatio.failure();
sendErrorCount.incrementAndGet();
if (this.fullStatsEnabled) {
sendSuccessRatio.failure(now);
sendErrorCount.incrementAndGet();
}
sendErrorRate.increment();
}
if (logger.isTraceEnabled()) {
logger.trace(timer);
logger.trace("Elapsed: " + this.name + ": " + elapsed);
}
}
}

View File

@@ -16,10 +16,9 @@
package org.springframework.integration.channel.management;
import org.springframework.integration.support.management.MetricsEnablement;
import org.springframework.integration.support.management.Statistics;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.support.MetricType;
/**
@@ -30,16 +29,7 @@ import org.springframework.jmx.support.MetricType;
* @author Gary Russell
* @since 2.0
*/
public interface MessageChannelMetrics {
@ManagedOperation
void reset();
@ManagedOperation
void enableStats(boolean statsEnabled);
@ManagedAttribute
boolean isStatsEnabled();
public interface MessageChannelMetrics extends MetricsEnablement {
/**
* @return the number of successful sends

View File

@@ -0,0 +1,4 @@
/**
* Provides classes related to channel management.
*/
package org.springframework.integration.channel.management;

View File

@@ -150,7 +150,7 @@ public abstract class AbstractMethodAnnotationPostProcessor<T extends Annotation
((Advised) handler).addAdvisor(advisor);
}
else {
ProxyFactory proxyFactory = new ProxyFactory(bean);
ProxyFactory proxyFactory = new ProxyFactory(handler);
proxyFactory.addAdvisor(advisor);
handler = (MessageHandler) proxyFactory.getProxy(this.beanFactory.getBeanClassLoader());
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -42,6 +42,7 @@ import org.springframework.util.xml.DomUtils;
*
* @author Mark Fisher
* @author Artem Bilan
* @author Gary Russell
*/
public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundChannelAdapterParser {
@@ -79,7 +80,7 @@ public class DefaultInboundChannelAdapterParser extends AbstractPollingInboundCh
result = this.parseMethodInvokingSource(innnerBeanDef, methodName, element, parserContext);
}
else {
result = innnerBeanDef;
result = innnerBeanDef.getBeanDefinition();
}
}
else if (hasScriptElement) {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -19,12 +19,15 @@ package org.springframework.integration.endpoint;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.expression.Expression;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.management.MessageSourceMetrics;
import org.springframework.integration.support.AbstractIntegrationMessageBuilder;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.integration.util.AbstractExpressionEvaluator;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
@@ -33,15 +36,25 @@ import org.springframework.util.CollectionUtils;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
* @since 2.0
*/
@IntegrationManagedResource
public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluator implements MessageSource<T>,
NamedComponent, BeanNameAware {
MessageSourceMetrics, NamedComponent, BeanNameAware {
private final AtomicLong messageCount = new AtomicLong();
private volatile Map<String, Expression> headerExpressions = Collections.emptyMap();
private volatile String beanName;
private volatile String managedType;
private volatile String managedName;
private volatile boolean countsEnabled;
public void setHeaderExpressions(Map<String, Expression> headerExpressions) {
this.headerExpressions = (headerExpressions != null)
? headerExpressions : Collections.<String, Expression>emptyMap();
@@ -52,11 +65,56 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
this.beanName = name;
}
@Override
public void setManagedType(String managedType) {
this.managedType = managedType;
}
@Override
public String getManagedType() {
return this.managedType;
}
@Override
public void setManagedName(String managedName) {
this.managedName = managedName;
}
@Override
public String getManagedName() {
return this.managedName;
}
@Override
public String getComponentName() {
return this.beanName;
}
@Override
public boolean isCountsEnabled() {
return this.countsEnabled;
}
@Override
public void enableCounts(boolean countsEnabled) {
this.countsEnabled = countsEnabled;
}
@Override
public void reset() {
this.messageCount.set(0);
}
@Override
public int getMessageCount() {
return (int) this.messageCount.get();
}
@Override
public long getMessageCountLong() {
return this.messageCount.get();
}
@Override
@SuppressWarnings("unchecked")
public final Message<T> receive() {
@@ -91,6 +149,9 @@ public abstract class AbstractMessageSource<T> extends AbstractExpressionEvaluat
}
message = builder.build();
}
if (this.countsEnabled && message != null) {
this.messageCount.incrementAndGet();
}
return message;
}

View File

@@ -1,46 +1,48 @@
/*
* Copyright 2002-2014 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.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
* @since 2.0
*/
public interface MessageSourceMetrics {
@ManagedOperation
void reset();
/**
* @return the number of successful handler calls
*/
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Message Source Message Count")
int getMessageCount();
/**
* @return the number of successful handler calls
* @since 3.0
*/
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Message Source Message Count")
long getMessageCountLong();
String getName();
String getSource();
}
/*
* 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.endpoint.management;
import org.springframework.integration.support.management.CountsEnablement;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
public interface MessageSourceMetrics extends CountsEnablement {
/**
* @return the number of successful handler calls
*/
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Message Source Message Count")
int getMessageCount();
/**
* @return the number of successful handler calls
* @since 3.0
*/
@ManagedMetric(metricType = MetricType.COUNTER, displayName = "Message Source Message Count")
long getMessageCountLong();
void setManagedName(String name);
String getManagedName();
void setManagedType(String source);
String getManagedType();
}

View File

@@ -0,0 +1,4 @@
/**
* Provides classes related to endpoint management.
*/
package org.springframework.integration.endpoint.management;

View File

@@ -19,8 +19,12 @@ 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.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.IntegrationManagedResource;
import org.springframework.integration.support.management.Statistics;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessageHandlingException;
@@ -37,12 +41,23 @@ import org.springframework.util.Assert;
* @author Oleg Zhurakousky
* @author Gary Russell
*/
public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler, TrackableComponent, Orderable {
@IntegrationManagedResource
public abstract class AbstractMessageHandler extends IntegrationObjectSupport implements MessageHandler,
MessageHandlerMetrics, TrackableComponent, Orderable {
private volatile boolean shouldTrack = false;
private volatile int order = Ordered.LOWEST_PRECEDENCE;
private final SimpleMessageHandlerMetrics handlerMetrics = new SimpleMessageHandlerMetrics();
private volatile boolean statsEnabled;
private volatile boolean countsEnabled;
private volatile String managedName;
private volatile String managedType;
@Override
public void setOrder(int order) {
@@ -71,13 +86,19 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
if (this.logger.isDebugEnabled()) {
this.logger.debug(this + " received message: " + message);
}
long start = 0;
try {
if (message != null && this.shouldTrack) {
message = MessageHistory.write(message, this, this.getMessageBuilderFactory());
}
if (this.countsEnabled) {
start = this.handlerMetrics.beforeHandle(message);
}
this.handleMessageInternal(message);
this.handlerMetrics.afterHandle(start, true);
}
catch (Exception e) {
this.handlerMetrics.afterHandle(start, false);
if (e instanceof MessagingException) {
throw (MessagingException) e;
}
@@ -87,4 +108,113 @@ public abstract class AbstractMessageHandler extends IntegrationObjectSupport im
protected abstract void handleMessageInternal(Message<?> message) throws Exception;
@Override
public void reset() {
this.handlerMetrics.reset();
}
@Override
public long getHandleCountLong() {
return this.handlerMetrics.getHandleCountLong();
}
@Override
public int getHandleCount() {
return this.handlerMetrics.getHandleCount();
}
@Override
public int getErrorCount() {
return this.handlerMetrics.getErrorCount();
}
@Override
public long getErrorCountLong() {
return this.handlerMetrics.getErrorCountLong();
}
@Override
public double getMeanDuration() {
return this.handlerMetrics.getMeanDuration();
}
@Override
public double getMinDuration() {
return this.handlerMetrics.getMinDuration();
}
@Override
public double getMaxDuration() {
return this.handlerMetrics.getMaxDuration();
}
@Override
public double getStandardDeviationDuration() {
return this.handlerMetrics.getStandardDeviationDuration();
}
@Override
public int getActiveCount() {
return this.handlerMetrics.getActiveCount();
}
@Override
public long getActiveCountLong() {
return this.handlerMetrics.getActiveCountLong();
}
@Override
public Statistics getDuration() {
return this.handlerMetrics.getDuration();
}
@Override
public void enableStats(boolean statsEnabled) {
if (statsEnabled) {
this.countsEnabled = true;
}
this.statsEnabled = statsEnabled;
if (this.handlerMetrics != null) {
this.handlerMetrics.setFullStatsEnabled(statsEnabled);
}
}
@Override
public boolean isStatsEnabled() {
return this.statsEnabled;
}
@Override
public void enableCounts(boolean countsEnabled) {
this.countsEnabled = countsEnabled;
if (!countsEnabled) {
this.statsEnabled = false;
}
}
@Override
public boolean isCountsEnabled() {
return this.countsEnabled;
}
@Override
public void setManagedName(String managedName) {
this.managedName = managedName;
}
@Override
public String getManagedName() {
return this.managedName;
}
@Override
public void setManagedType(String managedType) {
this.managedType = managedType;
}
@Override
public String getManagedType() {
return this.managedType;
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -38,6 +38,7 @@ import org.springframework.integration.store.MessageGroupStore;
import org.springframework.integration.store.MessageStore;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
@@ -74,10 +75,12 @@ import org.springframework.util.CollectionUtils;
*
* @author Mark Fisher
* @author Artem Bilan
* @author Gary Russell
* @since 1.0.3
*/
@ManagedResource
@IntegrationManagedResource
public class DelayHandler extends AbstractReplyProducingMessageHandler implements DelayHandlerManagement,
ApplicationListener<ContextRefreshedEvent> {

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -14,21 +14,19 @@
* limitations under the License.
*/
package org.springframework.integration.monitor;
package org.springframework.integration.handler.management;
import org.springframework.integration.support.management.MetricsEnablement;
import org.springframework.integration.support.management.Statistics;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.support.MetricType;
/**
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
public interface MessageHandlerMetrics {
@ManagedOperation
void reset();
public interface MessageHandlerMetrics extends MetricsEnablement {
/**
* @return the number of successful handler calls
@@ -91,8 +89,12 @@ public interface MessageHandlerMetrics {
*/
Statistics getDuration();
String getName();
void setManagedName(String name);
String getSource();
String getManagedName();
void setManagedType(String source);
String getManagedType();
}

View File

@@ -14,36 +14,29 @@
* limitations under the License.
*/
package org.springframework.integration.monitor;
package org.springframework.integration.handler.management;
import java.util.concurrent.atomic.AtomicLong;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.support.management.ExponentialMovingAverage;
import org.springframework.integration.support.management.Statistics;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.util.StopWatch;
/**
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
@ManagedResource
public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHandlerMetrics {
public class SimpleMessageHandlerMetrics {
private static final Log logger = LogFactory.getLog(SimpleMessageHandlerMetrics.class);
private static final int DEFAULT_MOVING_AVERAGE_WINDOW = 10;
private final MessageHandler handler;
private final AtomicLong activeCount = new AtomicLong();
private final AtomicLong handleCount = new AtomicLong();
@@ -54,83 +47,45 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa
private volatile String name;
private volatile String source;
public SimpleMessageHandlerMetrics(MessageHandler handler) {
this.handler = handler;
}
private volatile boolean fullStatsEnabled;
public void setName(String name) {
this.name = name;
}
@Override
public String getName() {
return this.name;
public void setFullStatsEnabled(boolean fullStatsEnabled) {
this.fullStatsEnabled = fullStatsEnabled;
}
public void setSource(String source) {
this.source = source;
}
@Override
public String getSource() {
return this.source;
}
public MessageHandler getMessageHandler() {
return this.handler;
}
@Override
public Object invoke(MethodInvocation invocation) throws Throwable {
String method = invocation.getMethod().getName();
if ("handleMessage".equals(method)) {
Message<?> message = (Message<?>) invocation.getArguments()[0];
handleMessage(invocation, message);
return null;
}
return invocation.proceed();
}
private void handleMessage(MethodInvocation invocation, Message<?> message) throws Throwable {
public long beforeHandle(Message<?> message) throws Exception {
if (logger.isTraceEnabled()) {
logger.trace("messageHandler(" + this.handler + ") message(" + message + ") :");
logger.trace("messageHandler(" + this.name + ") message(" + message + ") :");
}
String name = this.name;
if (name == null) {
name = this.handler.toString();
long start = 0;
if (this.fullStatsEnabled) {
start = System.currentTimeMillis();
}
StopWatch timer = new StopWatch(name + ".handle:execution");
try {
timer.start();
this.handleCount.incrementAndGet();
this.activeCount.incrementAndGet();
this.handleCount.incrementAndGet();
this.activeCount.incrementAndGet();
return start;
}
invocation.proceed();
timer.stop();
this.duration.append(timer.getTotalTimeMillis());
public void afterHandle(long start, boolean success) {
this.activeCount.decrementAndGet();
if (this.fullStatsEnabled && success) {
this.duration.append(System.currentTimeMillis() - start);
}
catch (Throwable e) {//NOSONAR - rethrown below
else if (!success) {
this.errorCount.incrementAndGet();
throw e;
}
finally {
this.activeCount.decrementAndGet();
}
}
@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);
@@ -138,59 +93,44 @@ public class SimpleMessageHandlerMetrics implements MethodInterceptor, MessageHa
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();
}
@Override
public String toString() {
return String.format("MessageHandlerMonitor: [name=%s, source=%s, duration=%s]", name, source, duration);
}
}

View File

@@ -0,0 +1,4 @@
/**
* Provides classes related to handler management.
*/
package org.springframework.integration.handler.management;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -31,6 +31,7 @@ import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.context.SmartLifecycle;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
@@ -46,6 +47,7 @@ import org.springframework.util.StringUtils;
* @since 2.0
*/
@ManagedResource
@IntegrationManagedResource
public class MessageHistoryConfigurer implements SmartLifecycle, BeanFactoryAware {
private final Log logger = LogFactory.getLog(this.getClass());

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,14 +17,15 @@
package org.springframework.integration.history;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
/**
* @author Mark Fisher
* @author Gary Russell
* @since 2.0
*/
@ManagedResource
@IntegrationManagedResource
public interface TrackableComponent extends NamedComponent {
@ManagedOperation

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -16,6 +16,7 @@
package org.springframework.integration.metadata;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedResource;
@@ -26,9 +27,11 @@ import org.springframework.jmx.export.annotation.ManagedResource;
* @author Josh Long
* @author Oleg Zhurakousky
* @author Mark Fisher
* @author Gary Russell
* @since 2.0
*/
@ManagedResource
@IntegrationManagedResource
public interface MetadataStore {
/**

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -24,6 +24,7 @@ import org.springframework.core.convert.support.DefaultConversionService;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -43,6 +44,7 @@ import org.springframework.util.Assert;
* @author Artem Bilan
*/
@ManagedResource
@IntegrationManagedResource
public abstract class AbstractMessageRouter extends AbstractMessageHandler {
private volatile MessageChannel defaultOutputChannel;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -20,6 +20,7 @@ import java.util.Map;
import java.util.Properties;
import org.springframework.integration.router.RecipientListRouter.Recipient;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
@@ -29,10 +30,12 @@ import org.springframework.jmx.export.annotation.ManagedResource;
* RecipientListRouter. This can be used with a control-bus and JMX.
*
* @author Liujiong
* @author Gary Russell
* @since 4.1
*
*/
@ManagedResource
@IntegrationManagedResource
public interface RecipientListRouterManagement {
/**

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -23,6 +23,7 @@ import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedResource;
@@ -37,6 +38,7 @@ import org.springframework.messaging.Message;
*
*/
@ManagedResource
@IntegrationManagedResource
public abstract class AbstractMessageGroupStore implements MessageGroupStore, Iterable<MessageGroup>,
BeanFactoryAware {

View File

@@ -0,0 +1,39 @@
/*
* 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;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
/**
* Base interface containing methods to control basic statistics gathering.
*
* @author Gary Russell
* @since 4.2
*
*/
public interface CountsEnablement {
@ManagedOperation
void reset();
@ManagedOperation(description = "Enable message counting statistics")
void enableCounts(boolean countsEnabled);
@ManagedAttribute
boolean isCountsEnabled();
}

View File

@@ -57,14 +57,30 @@ public class ExponentialMovingAverageRatio {
* Add a new event with successful outcome.
*/
public void success() {
append(1);
append(1, System.currentTimeMillis());
}
/**
* Add a new event with successful outcome.
* @param t The current timestamp.
*/
public void success(long t) {
append(1, t);
}
/**
* Add a new event with failed outcome.
*/
public void failure() {
append(0);
append(0, System.currentTimeMillis());
}
/**
* Add a new event with failed outcome.
* @param t the current timestamp.
*/
public void failure(long t) {
append(0, t);
}
public synchronized void reset() {
@@ -74,8 +90,7 @@ public class ExponentialMovingAverageRatio {
cumulative.reset();
}
private synchronized void append(int value) {
long t = System.currentTimeMillis();
private synchronized void append(int value, long t) {
double alpha = Math.exp((t0 - t) * lapse);
t0 = t;
sum = alpha * sum + value;

View File

@@ -0,0 +1,66 @@
/*
* 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;
import java.lang.annotation.Documented;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.springframework.jmx.export.annotation.ManagedResource;
/**
* Clone of {@link ManagedResource} limiting beans thus annoated so that they
* will only be exported by the {@code IntegrationMBeanExporter}.
*
* @author Gary Russell
* @since 4.2
*
*/
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface IntegrationManagedResource {
/**
* The annotation value is equivalent to the {@code objectName}
* attribute, for simple default usage.
* @return the value.
*/
String value() default "";
String objectName() default "";
String description() default "";
int currencyTimeLimit() default -1;
boolean log() default false;
String logFile() default "";
String persistPolicy() default "";
int persistPeriod() default -1;
String persistName() default "";
String persistLocation() default "";
}

View File

@@ -0,0 +1,37 @@
/*
* 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;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
/**
* Base interface containing methods to control complete statistics gathering.
*
* @author Gary Russell
* @since 4.2
*
*/
public interface MetricsEnablement extends CountsEnablement {
@ManagedOperation(description = "Enable all statistics")
void enableStats(boolean statsEnabled);
@ManagedAttribute
boolean isStatsEnabled();
}

View File

@@ -0,0 +1,4 @@
/**
* Provides classes related to management.
*/
package org.springframework.integration.support.management;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -16,22 +16,27 @@ package org.springframework.integration.groovy.config;
import java.util.HashMap;
import java.util.Map;
import groovy.lang.Binding;
import groovy.lang.MissingPropertyException;
import org.springframework.beans.factory.*;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanCreationNotAllowedException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.context.Lifecycle;
import org.springframework.core.annotation.AnnotationUtils;
import org.springframework.messaging.Message;
import org.springframework.integration.config.AbstractSimpleMessageHandlerFactoryBean;
import org.springframework.messaging.MessageHandler;
import org.springframework.integration.groovy.GroovyCommandMessageProcessor;
import org.springframework.integration.handler.ServiceActivatingHandler;
import org.springframework.integration.scripting.ScriptVariableGenerator;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.scripting.groovy.GroovyObjectCustomizer;
import org.springframework.util.CustomizableThreadCreator;
import groovy.lang.Binding;
import groovy.lang.MissingPropertyException;
/**
* FactoryBean for creating {@link MessageHandler} instances to handle a message as a Groovy Script.
*
@@ -40,6 +45,7 @@ import org.springframework.util.CustomizableThreadCreator;
* @author Mark Fisher
* @author Artem Bilan
* @author Stefan Reuter
* @author Gary Russell
* @since 2.0
*/
public class GroovyControlBusFactoryBean extends AbstractSimpleMessageHandlerFactoryBean<MessageHandler> implements BeanClassLoaderAware {
@@ -67,11 +73,14 @@ public class GroovyControlBusFactoryBean extends AbstractSimpleMessageHandlerFac
protected MessageHandler createHandler() {
Binding binding = new ManagedBeansBinding(this.getBeanFactory());
GroovyCommandMessageProcessor processor = new GroovyCommandMessageProcessor(binding, new ScriptVariableGenerator() {
@Override
public Map<String, Object> generateScriptVariables(Message<?> message) {
Map<String, Object> variables = new HashMap<String, Object>();
variables.put("headers", message.getHeaders());
return variables;
}
});
if (this.customizer != null) {
processor.setCustomizer(this.customizer);
@@ -131,7 +140,8 @@ public class GroovyControlBusFactoryBean extends AbstractSimpleMessageHandlerFac
if (bean instanceof Lifecycle ||
bean instanceof CustomizableThreadCreator ||
(AnnotationUtils.findAnnotation(bean.getClass(), ManagedResource.class) != null)) {
(AnnotationUtils.findAnnotation(bean.getClass(), ManagedResource.class) != null) ||
(AnnotationUtils.findAnnotation(bean.getClass(), IntegrationManagedResource.class) != null)) {
return bean;
}
throw new BeanCreationNotAllowedException(name, "Only beans with @ManagedResource or beans which implement " +

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -54,7 +54,6 @@ import org.springframework.jdbc.core.SingleColumnRowMapper;
import org.springframework.jdbc.support.lob.DefaultLobHandler;
import org.springframework.jdbc.support.lob.LobHandler;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHeaders;
@@ -77,10 +76,10 @@ import org.springframework.util.StringUtils;
* @author Matt Stine
* @author Gunnar Hillert
* @author Will Schipp
* @author Gary Russell
*
* @since 2.0
*/
@ManagedResource
public class JdbcMessageStore extends AbstractMessageGroupStore implements MessageStore, InitializingBean {
private static final Log logger = LogFactory.getLog(JdbcMessageStore.class);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -32,6 +32,7 @@ import org.springframework.expression.common.LiteralExpression;
import org.springframework.expression.spel.support.StandardEvaluationContext;
import org.springframework.integration.expression.ExpressionUtils;
import org.springframework.integration.jdbc.storedproc.ProcedureParameter;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowMapper;
import org.springframework.jdbc.core.SqlInOutParameter;
@@ -65,6 +66,7 @@ import com.google.common.cache.LoadingCache;
*
*/
@ManagedResource
@IntegrationManagedResource
public class StoredProcExecutor implements BeanFactoryAware, InitializingBean {
private volatile EvaluationContext evaluationContext;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -55,6 +55,7 @@ import org.springframework.integration.store.PriorityCapableChannelMessageStore;
import org.springframework.integration.store.SimpleMessageGroup;
import org.springframework.integration.support.DefaultMessageBuilderFactory;
import org.springframework.integration.support.MessageBuilderFactory;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.integration.support.utils.IntegrationUtils;
import org.springframework.integration.transaction.TransactionSynchronizationFactory;
import org.springframework.integration.util.UUIDConverter;
@@ -92,9 +93,11 @@ import org.springframework.util.StringUtils;
*
* @author Gunnar Hillert
* @author Artem Bilan
* @author Gary Russell
* @since 2.2
*/
@ManagedResource
@IntegrationManagedResource
public class JdbcChannelMessageStore implements PriorityCapableChannelMessageStore, InitializingBean, BeanFactoryAware {
private static final Log logger = LogFactory.getLog(JdbcChannelMessageStore.class);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -19,6 +19,8 @@ package org.springframework.integration.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;
import org.springframework.messaging.PollableChannel;
@@ -31,7 +33,8 @@ import org.springframework.messaging.support.ChannelInterceptor;
* @author Artem Bilan
* @since 2.0
*/
public class PollableJmsChannel extends AbstractJmsChannel implements PollableChannel {
public class PollableJmsChannel extends AbstractJmsChannel implements PollableChannel,
PollableChannelManagement {
private volatile String messageSelector;
@@ -43,9 +46,41 @@ 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();
}
@Override
public long getReceiveCountLong() {
return getMetrics().getReceiveCountLong();
}
@Override
public int getReceiveErrorCount() {
return getMetrics().getReceiveErrorCount();
}
@Override
public long getReceiveErrorCountLong() {
return getMetrics().getReceiveErrorCountLong();
}
@Override
public Message<?> receive() {
ChannelInterceptorList interceptorList = getInterceptors();
Deque<ChannelInterceptor> interceptorStack = null;
boolean counted = false;
try {
if (interceptorList.getInterceptors().size() > 0) {
interceptorStack = new ArrayDeque<ChannelInterceptor>();
@@ -65,6 +100,10 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
if (object == null) {
return null;
}
if (isCountsEnabled()) {
getMetrics().afterReceive();
counted = true;
}
Message<?> message = null;
if (object instanceof Message<?>) {
message = (Message<?>) object;
@@ -79,6 +118,9 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
return message;
}
catch (RuntimeException e) {
if (isCountsEnabled() && !counted) {
getMetrics().afterError();
}
if (interceptorStack != null) {
interceptorList.afterReceiveCompletion(null, this, e, interceptorStack);
}
@@ -86,6 +128,7 @@ public class PollableJmsChannel extends AbstractJmsChannel implements PollableCh
}
}
@Override
public Message<?> receive(long timeout) {
try {
DynamicJmsTemplateProperties.setReceiveTimeout(timeout);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -29,6 +29,7 @@ import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.MBeanExporter;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.jmx.export.notification.NotificationPublisher;
@@ -40,6 +41,7 @@ import org.springframework.util.Assert;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gary Russell
* @since 2.0
*/
public class NotificationPublishingMessageHandler extends AbstractMessageHandler implements BeanFactoryAware, InitializingBean {
@@ -128,8 +130,10 @@ public class NotificationPublishingMessageHandler extends AbstractMessageHandler
/**
* Simple class used for the actual MBean instances to be registered.
* Exposed to standard MBEs as well as the IMBE for backwards compatibility.
*/
@ManagedResource
@IntegrationManagedResource
private static class PublisherDelegate implements NotificationPublisherAware {
private volatile NotificationPublisher notificationPublisher;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors.
* Copyright 2014-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -37,6 +37,7 @@ import org.springframework.jmx.support.RegistrationPolicy;
* {@code <int-jmx:mbean-export/>} element.
*
* @author Artem Bilan
* @author Gary Russell
* @since 4.0
*/
@Target(ElementType.TYPE)
@@ -70,14 +71,52 @@ public @interface EnableIntegrationMBeanExport {
RegistrationPolicy registration() default RegistrationPolicy.FAIL_ON_EXISTING;
/**
* List of simple patterns for component names to register (defaults to '*').
* The pattern is applied to all components before they are registered, looking for a match on
* the 'name' property of the ObjectName. A MessageChannel and a MessageHandler (for instance)
* can share a name because they have a different type, so in that case they would either both
* be included or both excluded.
* Supports property placeholders (e.g. {@code ${managed.components}}). Can be applied for each element.
* A list of simple patterns for component names to register (defaults to '*'). The
* pattern is applied to all components before they are registered, looking for a
* match on the 'name' property of the ObjectName. A MessageChannel and a
* MessageHandler (for instance) can share a name because they have a different type,
* so in that case they would either both be included or both excluded. Since version
* 4.2, a leading '!' negates the pattern match ('!foo*' means don't export components
* where the name matches the pattern 'foo*'). For components with names that match
* multiple patterns, the first pattern wins. Supports property placeholders (e.g.
* {@code $ managed.components}}). Can be applied for each element.
* @return the patterns.
*/
String[] managedComponents() default "*";
/**
* A list of simple patterns for component names for which message counts will be
* enabled (defaults to '*'). Only patterns that also match
* {@link #managedComponents() managedComponents} will be considered. Enables message
* counting (`sendCount`, `errorCount`, `receiveCount`) for those components that
* support counters (channels, message handlers, etc). This is the initial setting
* only, individual components can have counts enabled/disabled at runtime. May be
* overridden by an entry in {@link #statsEnabled() statsEnabled} which is additional
* functionality over simple counts. If a pattern starts with `!`, counts are disabled
* for matches. For components that match multiple patterns, the first pattern wins.
* Disabling counts at runtime also disables stats.
* @return the patterns.
* @since 4.2
*/
String[] countsEnabled() default "*";
/**
* A list of simple patterns for component names for which message statistics will be
* enabled (response times, rates etc), as well as counts (a positive match here
* overrides {@link #countsEnabled() countsEnabled}, you can't have statistics without
* counts). (defaults to '*'). Only patterns that also match
* {@link #managedComponents() managedComponents} will be considered. Enables
* statistics for those components that support statistics (channels - when sending,
* message handlers, etc). This is the initial setting only, individual components can
* have stats enabled/disabled at runtime. If a pattern starts with `!`, stats (and
* counts) are disabled for matches. Note: this means that '!foo' here will disable
* stats and counts for 'foo' even if counts are enabled for 'foo' in
* {@link #countsEnabled() countsEnabled}. For components
* that match multiple patterns, the first pattern wins. Enabling stats at runtime
* also enables counts.
* @return the patterns.
* @since 4.2
*/
String[] statsEnabled() default "*";
}

View File

@@ -17,6 +17,7 @@
package org.springframework.integration.jmx.config;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
@@ -98,6 +99,8 @@ public class IntegrationMBeanExportConfiguration implements ImportAware, Environ
setupDomain(exporter);
setupServer(exporter);
setupComponentNamePatterns(exporter);
setupCountsEnabledNamePatterns(exporter);
setupStatsEnabledNamePatterns(exporter);
return exporter;
}
@@ -136,9 +139,29 @@ public class IntegrationMBeanExportConfiguration implements ImportAware, Environ
String[] managedComponents = this.attributes.getStringArray("managedComponents");
for (String managedComponent : managedComponents) {
String pattern = this.environment.resolvePlaceholders(managedComponent);
patterns.addAll(StringUtils.commaDelimitedListToSet(pattern));
patterns.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray(pattern)));
}
exporter.setComponentNamePatterns(patterns.toArray(new String[patterns.size()]));
}
private void setupCountsEnabledNamePatterns(IntegrationMBeanExporter exporter) {
List<String> patterns = new ArrayList<String>();
String[] countsEnabled = this.attributes.getStringArray("countsEnabled");
for (String managedComponent : countsEnabled) {
String pattern = this.environment.resolvePlaceholders(managedComponent);
patterns.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray(pattern)));
}
exporter.setEnabledCountsPatterns(patterns.toArray(new String[patterns.size()]));
}
private void setupStatsEnabledNamePatterns(IntegrationMBeanExporter exporter) {
List<String> patterns = new ArrayList<String>();
String[] statsEnabled = this.attributes.getStringArray("statsEnabled");
for (String managedComponent : statsEnabled) {
String pattern = this.environment.resolvePlaceholders(managedComponent);
patterns.addAll(Arrays.asList(StringUtils.commaDelimitedListToStringArray(pattern)));
}
exporter.setEnabledStatsPatterns(patterns.toArray(new String[patterns.size()]));
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014-2015 the original author or authors.
* Copyright 2014 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,12 +21,12 @@ import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.beans.factory.support.BeanDefinitionRegistry;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.integration.config.IntegrationConfigurationInitializer;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
/**
* The JMX Integration infrastructure {@code beanFactory} initializer.
*
* @author Artem Bilan
* @author Gary Russell
* @since 4.0
*/
public class JmxIntegrationConfigurationInitializer implements IntegrationConfigurationInitializer {
@@ -39,8 +39,10 @@ public class JmxIntegrationConfigurationInitializer implements IntegrationConfig
}
private void registerMBeanExporterHelperIfNecessary(ConfigurableListableBeanFactory beanFactory) {
((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(MBEAN_EXPORTER_HELPER_BEAN_NAME,
new RootBeanDefinition(MBeanExporterHelper.class));
if (beanFactory.getBeanNamesForType(IntegrationMBeanExporter.class).length > 0) {
((BeanDefinitionRegistry) beanFactory).registerBeanDefinition(MBEAN_EXPORTER_HELPER_BEAN_NAME,
new RootBeanDefinition(MBeanExporterHelper.class));
}
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2015 the original author or authors.
* Copyright 2002-2014 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
@@ -26,6 +26,7 @@ import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.core.Ordered;
import org.springframework.core.type.StandardMethodMetadata;
import org.springframework.integration.config.IntegrationConfigUtils;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
import org.springframework.jmx.export.MBeanExporter;
import org.springframework.util.Assert;
@@ -37,36 +38,19 @@ import org.springframework.util.StringUtils;
* of bean names that will be exported by the IntegrationMBeanExporter and merges it with the list
* of 'excludedBeans' of MBeanExporter so it will not attempt to export them again.
*
* Since 4.2, we unconditionally exclude all channels from standard context exporter (channels
* are now managed resources and would otherwise be picked up).
*
* @author Oleg Zhurakousky
* @author Artem Bilan
* @author Gary Russell
* @since 2.1
*
*/
class MBeanExporterHelper implements BeanPostProcessor, Ordered, BeanFactoryAware, InitializingBean {
private final static String EXCLUDED_BEANS_PROPERTY_NAME = "excludedBeans";
private final static String SI_ROOT_PACKAGE = "org.springframework.integration.";
private final Set<String> siBeanNames = new HashSet<String>();
/**
* For backwards compatibility, we don't want a context MBeanExporter to export
* integration components that are now managed resources. When there *is* an IMBE,
* *all* managed resources in org.springframework.integration are exluded.
*/
private final Set<String> siBeansToExcludeWhenNoIMBE = new HashSet<String>();
private volatile DefaultListableBeanFactory beanFactory;
private volatile boolean capturedAutoChannelCandidates;
private volatile boolean hasIntegrationExporter;
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
Assert.isInstanceOf(DefaultListableBeanFactory.class, beanFactory);
@@ -84,16 +68,9 @@ class MBeanExporterHelper implements BeanPostProcessor, Ordered, BeanFactoryAwar
className = ((StandardMethodMetadata) def.getSource()).getIntrospectedMethod().getReturnType().getName();
}
if (StringUtils.hasText(className)){
if (className.startsWith(SI_ROOT_PACKAGE)
if (className.startsWith(IntegrationConfigUtils.BASE_PACKAGE)
&& !(className.endsWith(IntegrationMBeanExporter.class.getName()))){
this.siBeanNames.add(beanName);
}
if (className.startsWith(SI_ROOT_PACKAGE + "channel.")
&& !(className.endsWith(IntegrationMBeanExporter.class.getName()))){
this.siBeansToExcludeWhenNoIMBE.add(beanName);
}
else if (className.equals(IntegrationMBeanExporter.class.getName())) {
this.hasIntegrationExporter = true;
siBeanNames.add(beanName);
}
}
}
@@ -109,29 +86,14 @@ class MBeanExporterHelper implements BeanPostProcessor, Ordered, BeanFactoryAwar
Collection<String> autoCreateChannelCandidatesNames =
(Collection<String>) new DirectFieldAccessor(autoCreateChannelCandidates).getPropertyValue("channelNames");
this.siBeanNames.addAll(autoCreateChannelCandidatesNames);
this.siBeansToExcludeWhenNoIMBE.addAll(autoCreateChannelCandidatesNames);
}
this.capturedAutoChannelCandidates = true;
}
if (bean instanceof MBeanExporter && !(bean instanceof IntegrationMBeanExporter)) {
MBeanExporter mbeanExporter = (MBeanExporter) bean;
DirectFieldAccessor mbeDfa = new DirectFieldAccessor(mbeanExporter);
@SuppressWarnings("unchecked")
Set<String> excludedNames = (Set<String>) mbeDfa.getPropertyValue(EXCLUDED_BEANS_PROPERTY_NAME);
if (excludedNames != null) {
excludedNames = new HashSet<String>(excludedNames);
for (String siBean : this.siBeanNames) {
mbeanExporter.addExcludedBean(siBean);
}
else {
excludedNames = new HashSet<String>();
}
if (!this.hasIntegrationExporter) {
excludedNames.addAll(this.siBeansToExcludeWhenNoIMBE);
}
else {
excludedNames.addAll(this.siBeanNames);
}
//TODO: SF 4.1.5 and above now has additive exclusions (SPR-12686)
mbeDfa.setPropertyValue(EXCLUDED_BEANS_PROPERTY_NAME, excludedNames);
}
return bean;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -58,6 +58,8 @@ public class MBeanExporterParser extends AbstractSingleBeanDefinitionParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "default-domain");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "object-name-static-properties");
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, "object-naming-strategy", "namingStrategy");
builder.addPropertyValue("server", mbeanServer);

View File

@@ -31,38 +31,41 @@ import javax.management.JMException;
import javax.management.ObjectName;
import javax.management.modelmbean.ModelMBean;
import org.aopalliance.aop.Advice;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.aop.Advisor;
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.aop.support.NameMatchMethodPointcutAdvisor;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.annotation.AnnotationBeanUtils;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.EmbeddedValueResolverAware;
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.MessageChannelMetrics;
import org.springframework.integration.channel.management.PollableChannelManagement;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.management.MessageSourceMetrics;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
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.IntegrationManagedResource;
import org.springframework.integration.support.management.Statistics;
import org.springframework.jmx.export.MBeanExporter;
import org.springframework.jmx.export.UnableToRegisterMBeanException;
@@ -72,6 +75,7 @@ import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.jmx.export.assembler.MetadataMBeanInfoAssembler;
import org.springframework.jmx.export.metadata.InvalidMetadataException;
import org.springframework.jmx.export.naming.MetadataNamingStrategy;
import org.springframework.jmx.support.MetricType;
import org.springframework.messaging.MessageChannel;
@@ -81,6 +85,8 @@ import org.springframework.util.PatternMatchUtils;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.FieldCallback;
import org.springframework.util.ReflectionUtils.FieldFilter;
import org.springframework.util.StringUtils;
import org.springframework.util.StringValueResolver;
/**
* <p>
@@ -110,14 +116,15 @@ import org.springframework.util.ReflectionUtils.FieldFilter;
* @author Artem Bilan
*/
@ManagedResource
@IntegrationManagedResource
public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostProcessor, BeanFactoryAware,
ApplicationContextAware, BeanClassLoaderAware, SmartLifecycle {
ApplicationContextAware, EmbeddedValueResolverAware, 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 final AnnotationJmxAttributeSource attributeSource = new IntegrationJmxAttributeSource();
private ListableBeanFactory beanFactory;
@@ -127,9 +134,9 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
private final Map<Object, AtomicLong> anonymousSourceCounters = new HashMap<Object, AtomicLong>();
private final Set<SimpleMessageHandlerMetrics> handlers = new HashSet<SimpleMessageHandlerMetrics>();
private final Set<MessageHandlerMetrics> handlers = new HashSet<MessageHandlerMetrics>();
private final Set<SimpleMessageSourceMetrics> sources = new HashSet<SimpleMessageSourceMetrics>();
private final Set<MessageSourceMetrics> sources = new HashSet<MessageSourceMetrics>();
private final Set<Lifecycle> inboundLifecycleMessageProducers = new HashSet<Lifecycle>();
@@ -151,8 +158,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
private final Map<String, String> beansByEndpointName = new HashMap<String, String>();
private ClassLoader beanClassLoader;
private volatile boolean autoStartup = true;
private volatile int phase = 0;
@@ -171,12 +176,19 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
private String[] componentNamePatterns = { "*" };
private String[] enabledCountsPatterns = { "*" };
private String[] enabledStatsPatterns = { "*" };
private volatile long shutdownDeadline;
private final AtomicBoolean shuttingDown = new AtomicBoolean();
private MessageHistoryConfigurer messageHistoryConfigurer;
private StringValueResolver embeddedValueResolver;
public IntegrationMBeanExporter() {
super();
// Shouldn't be necessary, but to be on the safe side...
@@ -185,12 +197,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
setAssembler(assembler);
}
@Override
public void setBeanClassLoader(ClassLoader classLoader) {
this.beanClassLoader = classLoader;
super.setBeanClassLoader(classLoader);
}
/**
* Static properties that will be added to all object names.
*
@@ -211,11 +217,64 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
this.defaultNamingStrategy.setDefaultDomain(domain);
}
/**
* Set the array of simple patterns for component names to register (defaults to '*').
* The pattern is applied to all components before they are registered, looking for a
* match on the 'name' property of the ObjectName. A MessageChannel and a
* MessageHandler (for instance) can share a name because they have a different type,
* so in that case they would either both be included or both excluded. Since version
* 4.2, a leading '!' negates the pattern match ('!foo*' means don't export components
* where the name matches the pattern 'foo*'). For components with names that match
* multiple patterns, the first pattern wins.
* @param componentNamePatterns the patterns.
*/
public void setComponentNamePatterns(String[] componentNamePatterns) {
Assert.notEmpty(componentNamePatterns, "componentNamePatterns must not be empty");
this.componentNamePatterns = Arrays.copyOf(componentNamePatterns, componentNamePatterns.length);
}
/**
* Set the array of simple patterns for component names for which message counts will
* be enabled (defaults to '*'). Only patterns that also match
* {@link #setComponentNamePatterns(String[]) componentNamePatterns} will be
* considered. Enables message counting (`sendCount`, `errorCount`, `receiveCount`)
* for those components that support counters (channels, message handlers, etc).
* This is the initial setting only, individual components can have counts
* enabled/disabled at runtime. May be overridden by an entry in
* {@link #setEnabledStatsPatterns(String[]) enabledStatsPatterns} which is additional
* functionality over simple counts. If a pattern starts with `!`, counts are disabled
* for matches. For components that match multiple patterns, the first pattern wins.
* Disabling counts at runtime also disables stats.
* @param enabledCountsPatterns the patterns.
* @since 4.2
*/
public void setEnabledCountsPatterns(String[] enabledCountsPatterns) {
Assert.notEmpty(enabledCountsPatterns, "enabledCountsPatterns must not be empty");
this.enabledCountsPatterns = Arrays.copyOf(enabledCountsPatterns, enabledCountsPatterns.length);
}
/**
* Set the array of simple patterns for component names for which message statistics
* will be enabled (response times, rates etc), as well as counts (a positive match
* here overrides {@link #setEnabledCountsPatterns(String[]) enabledCountsPatterns},
* you can't have statistics without counts). (defaults to '*'). Only patterns that
* also match {@link #setComponentNamePatterns(String[]) componentNamePatterns} will
* be considered. Enables statistics for those components that support statistics
* (channels - when sending, message handlers, etc). This is the initial setting only,
* individual components can have stats enabled/disabled at runtime. If a pattern
* starts with `!`, stats (and counts) are disabled for matches. Note: this means that
* '!foo' here will disable stats and counts for 'foo' even if counts are enabled for
* 'foo' in {@link #setEnabledCountsPatterns(String[]) enabledCountsPatterns}. For
* components that match multiple patterns, the first pattern wins. Enabling stats at
* runtime also enables counts.
* @param enabledStatsPatterns the patterns.
* @since 4.2
*/
public void setEnabledStatsPatterns(String[] enabledStatsPatterns) {
Assert.notEmpty(enabledStatsPatterns, "componentNamePatterns must not be empty");
this.enabledStatsPatterns = Arrays.copyOf(enabledStatsPatterns, enabledStatsPatterns.length);
}
@Override
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
super.setBeanFactory(beanFactory);
@@ -231,18 +290,12 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
public void setEmbeddedValueResolver(StringValueResolver resolver) {
this.embeddedValueResolver = resolver;
}
if (bean instanceof Advised) {
for (Advisor advisor : ((Advised) bean).getAdvisors()) {
Advice advice = advisor.getAdvice();
if (advice instanceof MessageHandlerMetrics || advice instanceof MessageSourceMetrics
|| advice instanceof MessageChannelMetrics) {
// Already advised - so probably a factory bean product
return bean;
}
}
}
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (IntegrationContextUtils.INTEGRATION_MESSAGE_HISTORY_CONFIGURER_BEAN_NAME.equals(beanName)
&& bean instanceof MessageHistoryConfigurer) {
@@ -250,30 +303,31 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
return bean;
}
if (bean instanceof MessageHandler) {
if (bean instanceof MessageHandlerMetrics) {
if (this.handlerInAnonymousWrapper(bean) != null) {
if (logger.isDebugEnabled()) {
logger.debug("Skipping " + beanName + " because it wraps another handler");
}
return bean;
}
SimpleMessageHandlerMetrics monitor = new SimpleMessageHandlerMetrics((MessageHandler) bean);
Object advised = applyHandlerInterceptor(bean, monitor, beanClassLoader);
// If the handler is proxied, we have to extract the target to expose as an MBean.
// The MetadataMBeanInfoAssembler does not support JDK dynamic proxies.
MessageHandlerMetrics monitor = (MessageHandlerMetrics) extractTarget(bean);
handlers.add(monitor);
bean = advised;
}
if (bean instanceof MessageSource<?>) {
SimpleMessageSourceMetrics monitor = new SimpleMessageSourceMetrics((MessageSource<?>) bean);
Object advised = applySourceInterceptor(bean, monitor, beanClassLoader);
if (bean instanceof MessageSourceMetrics) {
// If the source is proxied, we have to extract the target to expose as an MBean.
// The MetadataMBeanInfoAssembler does not support JDK dynamic proxies.
MessageSourceMetrics monitor = (MessageSourceMetrics) extractTarget(bean);
sources.add(monitor);
bean = advised;
}
if (bean instanceof MessageChannel && bean instanceof MessageChannelMetrics
&& bean instanceof NamedComponent) {
// If the channel is proxied, we have to extract the target to expose as an MBean.
// The MetadataMBeanInfoAssembler does not support JDK dynamic proxies.
MessageChannelMetrics monitor = (MessageChannelMetrics) extractTarget(bean);
monitor.enableStats(true);//TODO: INT-3638
channels.add(monitor);
}
@@ -461,20 +515,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
}
/**
* Shutdown active components.
*
* @param force No longer used.
* @param howLong The time to wait in total for all activities to complete
* in milliseconds.
* @deprecated Use {@link #stopActiveComponents(long)}.
*/
@Deprecated
@ManagedOperation
public void stopActiveComponents(boolean force, long howLong) {
stopActiveComponents(howLong);
}
/**
* Shutdown active components.
*
@@ -532,11 +572,11 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
public void stopMessageSources() {
for (Entry<String, MessageSourceMetrics> entry : this.allSourcesByName.entrySet()) {
MessageSourceMetrics sourceMetrics = entry.getValue();
if (sourceMetrics instanceof LifecycleMessageSourceMetrics) {
if (sourceMetrics instanceof Lifecycle) {
if (logger.isInfoEnabled()) {
logger.info("Stopping message source " + sourceMetrics);
}
((LifecycleMessageSourceMetrics) sourceMetrics).stop();
((Lifecycle) sourceMetrics).stop();
}
else {
if (logger.isInfoEnabled()) {
@@ -713,7 +753,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
for (MessageChannelMetrics monitor : channels) {
String name = ((NamedComponent) monitor).getComponentName();
this.allChannelsByName.put(name, monitor);
if (!PatternMatchUtils.simpleMatch(this.componentNamePatterns, name)) {
if (!matches(this.componentNamePatterns, name)) {
continue;
}
// Only register once...
@@ -723,17 +763,25 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
if (name != null) {
channelsByName.put(name, monitor);
}
Boolean enabled = smartMatch(this.enabledCountsPatterns, name);
if (enabled != null) {
monitor.enableCounts(enabled);
}
enabled = smartMatch(this.enabledStatsPatterns, name);
if (enabled != null) {
monitor.enableStats(enabled);
}
registerBeanNameOrInstance(monitor, beanKey);
}
}
}
private void registerHandlers() {
for (SimpleMessageHandlerMetrics source : handlers) {
MessageHandlerMetrics monitor = enhanceHandlerMonitor(source);
String name = monitor.getName();
for (MessageHandlerMetrics handler : handlers) {
MessageHandlerMetrics monitor = enhanceHandlerMonitor(handler);
String name = monitor.getManagedName();
this.allHandlersByName.put(name, monitor);
if (!PatternMatchUtils.simpleMatch(this.componentNamePatterns, name)) {
if (!matches(this.componentNamePatterns, name)) {
continue;
}
// Only register once...
@@ -742,23 +790,25 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
if (name != null) {
handlersByName.put(name, monitor);
}
registerBeanNameOrInstance(monitor, beanKey);
// Expose the raw bean if it is managed
MessageHandler bean = source.getMessageHandler();
if (assembler.includeBean(bean.getClass(), source.getName())) {
registerBeanInstance(bean,
this.getMonitoredIntegrationObjectBeanKey(bean, name));
Boolean enabled = smartMatch(this.enabledCountsPatterns, name);
if (enabled != null) {
monitor.enableCounts(enabled);
}
enabled = smartMatch(this.enabledStatsPatterns, name);
if (enabled != null) {
monitor.enableStats(enabled);
}
registerBeanNameOrInstance(monitor, beanKey);
}
}
}
private void registerSources() {
for (SimpleMessageSourceMetrics source : sources) {
for (MessageSourceMetrics source : sources) {
MessageSourceMetrics monitor = enhanceSourceMonitor(source);
String name = monitor.getName();
String name = monitor.getManagedName();
this.allSourcesByName.put(name, monitor);
if (!PatternMatchUtils.simpleMatch(this.componentNamePatterns, name)) {
if (!matches(this.componentNamePatterns, name)) {
continue;
}
// Only register once...
@@ -767,13 +817,11 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
if (name != null) {
sourcesByName.put(name, monitor);
}
registerBeanNameOrInstance(monitor, beanKey);
// Expose the raw bean if it is managed
MessageSource<?> bean = source.getMessageSource();
if (assembler.includeBean(bean.getClass(), source.getName())) {
registerBeanInstance(bean,
this.getMonitoredIntegrationObjectBeanKey(bean, name));
Boolean enabled = smartMatch(this.enabledCountsPatterns, name);
if (enabled != null) {
monitor.enableCounts(enabled);
}
registerBeanNameOrInstance(monitor, beanKey);
}
}
}
@@ -795,7 +843,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
name = endpoint.getComponentName();
source = "endpoint";
}
if (!PatternMatchUtils.simpleMatch(this.componentNamePatterns, name)) {
if (!matches(this.componentNamePatterns, name)) {
continue;
}
if (endpointNames.contains(name)) {
@@ -814,18 +862,40 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
}
}
private Object applyHandlerInterceptor(Object bean, SimpleMessageHandlerMetrics interceptor,
ClassLoader beanClassLoader) {
NameMatchMethodPointcutAdvisor handlerAdvice = new NameMatchMethodPointcutAdvisor(interceptor);
handlerAdvice.addMethodName("handleMessage");
return applyAdvice(bean, handlerAdvice, beanClassLoader);
/**
* Simple pattern match against the supplied patterns; also supports negated ('!')
* patterns. First match wins (positive or negative).
* @param patterns the patterns.
* @param name the name to match.
* @return true if positive match, false if no match or negative match.
*/
private boolean matches(String[] patterns, String name) {
Boolean match = smartMatch(patterns, name);
return match == null ? false : match;
}
private Object applySourceInterceptor(Object bean, SimpleMessageSourceMetrics interceptor,
ClassLoader beanClassLoader) {
NameMatchMethodPointcutAdvisor sourceAdvice = new NameMatchMethodPointcutAdvisor(interceptor);
sourceAdvice.addMethodName("receive");
return applyAdvice(bean, sourceAdvice, beanClassLoader);
/**
* Simple pattern match against the supplied patterns; also supports negated ('!')
* patterns. First match wins (positive or negative).
* @param patterns the patterns.
* @param name the name to match.
* @return null if no match; true for positive match; false for negative match.
*/
private Boolean smartMatch(String[] patterns, String name) {
if (patterns != null) {
for (String pattern : patterns) {
boolean reverse = false;
String patternToUse = pattern;
if (pattern.startsWith("!")) {
reverse = true;
patternToUse = pattern.substring(1);
}
if (PatternMatchUtils.simpleMatch(patternToUse, name)) {
return !reverse;
}
}
}
return null;
}
private Object extractTarget(Object bean) {
@@ -876,14 +946,14 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
private String getHandlerBeanKey(MessageHandlerMetrics handler) {
// This ordering of keys seems to work with default settings of JConsole
return String.format(domain + ":type=MessageHandler,name=%s,bean=%s" + getStaticNames(), handler.getName(),
handler.getSource());
return String.format(domain + ":type=MessageHandler,name=%s,bean=%s" + getStaticNames(),
handler.getManagedName(), handler.getManagedType());
}
private String getSourceBeanKey(MessageSourceMetrics handler) {
private String getSourceBeanKey(MessageSourceMetrics source) {
// This ordering of keys seems to work with default settings of JConsole
return String.format(domain + ":type=MessageSource,name=%s,bean=%s" + getStaticNames(), handler.getName(),
handler.getSource());
return String.format(domain + ":type=MessageSource,name=%s,bean=%s" + getStaticNames(),
source.getManagedName(), source.getManagedType());
}
private String getEndpointBeanKey(AbstractEndpoint endpoint, String name, String source) {
@@ -891,11 +961,6 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
return String.format(domain + ":type=ManagedEndpoint,name=%s,bean=%s" + getStaticNames(), name, source);
}
private String getMonitoredIntegrationObjectBeanKey(Object object, String name) {
// This ordering of keys seems to work with default settings of JConsole
return String.format(domain + ":type=" + object.getClass().getSimpleName() + ",name=%s" + getStaticNames(), name);
}
private String getStaticNames() {
if (objectNameStaticProperties.isEmpty()) {
return "";
@@ -908,11 +973,11 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
return builder.toString();
}
private MessageHandlerMetrics enhanceHandlerMonitor(SimpleMessageHandlerMetrics monitor) {
private MessageHandlerMetrics enhanceHandlerMonitor(MessageHandlerMetrics monitor) {
MessageHandlerMetrics result = monitor;
if (monitor.getName() != null && monitor.getSource() != null) {
if (monitor.getManagedName() != null && monitor.getManagedType() != null) {
return monitor;
}
@@ -924,14 +989,12 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
String source = "endpoint";
Object endpoint = null;
MessageHandler messageHandler = monitor.getMessageHandler();
for (String beanName : names) {
endpoint = beanFactory.getBean(beanName);
try {
Object field = extractTarget(getField(endpoint, "handler"));
if (field == messageHandler ||
this.extractTarget(this.handlerInAnonymousWrapper(field)) == messageHandler) {
if (field == monitor ||
this.extractTarget(this.handlerInAnonymousWrapper(field)) == monitor) {
name = beanName;
endpointName = beanName;
break;
@@ -979,15 +1042,30 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
if (endpoint instanceof Lifecycle) {
// Wrap the monitor in a lifecycle so it exposes the start/stop operations
result = new LifecycleMessageHandlerMetrics((Lifecycle) endpoint, monitor);
if (monitor instanceof MappingMessageRouterManagement) {
if (monitor instanceof TrackableComponent) {
result = new TrackableRouterMetrics((Lifecycle) endpoint, (MappingMessageRouterManagement) monitor);
}
else {
result = new RouterMetrics((Lifecycle) endpoint, (MappingMessageRouterManagement) monitor);
}
}
else {
if (monitor instanceof TrackableComponent) {
result = new LifecycleTrackableMessageHandlerMetrics((Lifecycle) endpoint, monitor);
}
else {
result = new LifecycleMessageHandlerMetrics((Lifecycle) endpoint, monitor);
}
}
}
if (name == null) {
if (messageHandler instanceof NamedComponent) {
name = ((NamedComponent) messageHandler).getComponentName();
if (monitor instanceof NamedComponent) {
name = ((NamedComponent) monitor).getComponentName();
}
if (name == null) {
name = messageHandler.toString();
name = monitor.toString();
}
source = "handler";
}
@@ -996,8 +1074,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
beansByEndpointName.put(name, endpointName);
}
monitor.setSource(source);
monitor.setName(name);
monitor.setManagedType(source);
monitor.setManagedName(name);
return result;
@@ -1007,11 +1085,11 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
return name.substring("_org.springframework.integration".length() + 1);
}
private MessageSourceMetrics enhanceSourceMonitor(SimpleMessageSourceMetrics monitor) {
private MessageSourceMetrics enhanceSourceMonitor(MessageSourceMetrics monitor) {
MessageSourceMetrics result = monitor;
if (monitor.getName() != null && monitor.getSource() != null) {
if (monitor.getManagedName() != null) {
return monitor;
}
@@ -1032,7 +1110,7 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
catch (Exception e) {
logger.trace("Could not get source from bean = " + beanName);
}
if (field == monitor.getMessageSource()) {
if (field == monitor) {
name = beanName;
endpointName = beanName;
break;
@@ -1076,11 +1154,16 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
if (endpoint instanceof Lifecycle) {
// Wrap the monitor in a lifecycle so it exposes the start/stop operations
result = new LifecycleMessageSourceMetrics((Lifecycle) endpoint, monitor);
if (endpoint instanceof TrackableComponent) {
result = new LifecycleTrackableMessageSourceMetrics((Lifecycle) endpoint, monitor);
}
else {
result = new LifecycleMessageSourceMetrics((Lifecycle) endpoint, monitor);
}
}
if (name == null) {
name = monitor.getMessageSource().toString();
name = monitor.toString();
source = "handler";
}
@@ -1088,8 +1171,8 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
beansByEndpointName.put(name, endpointName);
}
monitor.setSource(source);
monitor.setName(name);
monitor.setManagedType(source);
monitor.setManagedName(name);
return result;
}
@@ -1108,4 +1191,30 @@ public class IntegrationMBeanExporter extends MBeanExporter implements BeanPostP
return ReflectionUtils.getField(field, target);
}
private class IntegrationJmxAttributeSource extends AnnotationJmxAttributeSource {
@Override
public org.springframework.jmx.export.metadata.ManagedResource getManagedResource(Class<?> beanClass)
throws InvalidMetadataException {
IntegrationManagedResource ann =
AnnotationUtils.getAnnotation(beanClass, IntegrationManagedResource.class);
if (ann == null) {
return null;
}
org.springframework.jmx.export.metadata.ManagedResource managedResource =
new org.springframework.jmx.export.metadata.ManagedResource();
AnnotationBeanUtils.copyPropertiesToBean(ann, managedResource,
IntegrationMBeanExporter.this.embeddedValueResolver);
if (!"".equals(ann.value()) && !StringUtils.hasLength(managedResource.getObjectName())) {
String value = ann.value();
if (IntegrationMBeanExporter.this.embeddedValueResolver != null) {
value = IntegrationMBeanExporter.this.embeddedValueResolver.resolveStringValue(value);
}
managedResource.setObjectName(value);
}
return managedResource;
}
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,24 +17,26 @@
package org.springframework.integration.monitor;
import org.springframework.context.Lifecycle;
import org.springframework.integration.handler.management.MessageHandlerMetrics;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.integration.support.management.Statistics;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
/**
* A {@link MessageHandlerMetrics} 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.
*
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
@ManagedResource
@IntegrationManagedResource
public class LifecycleMessageHandlerMetrics implements MessageHandlerMetrics, Lifecycle {
private final Lifecycle lifecycle;
private final MessageHandlerMetrics delegate;
protected final MessageHandlerMetrics delegate;
public LifecycleMessageHandlerMetrics(Lifecycle lifecycle, MessageHandlerMetrics delegate) {
@@ -101,13 +103,13 @@ public class LifecycleMessageHandlerMetrics implements MessageHandlerMetrics, Li
}
@Override
public String getName() {
return this.delegate.getName();
public String getManagedName() {
return this.delegate.getManagedName();
}
@Override
public String getSource() {
return this.delegate.getSource();
public String getManagedType() {
return this.delegate.getManagedType();
}
@Override
@@ -130,4 +132,34 @@ public class LifecycleMessageHandlerMetrics implements MessageHandlerMetrics, Li
return this.delegate.getActiveCountLong();
}
@Override
public void enableStats(boolean statsEnabled) {
this.delegate.enableStats(statsEnabled);
}
@Override
public void enableCounts(boolean countsEnabled) {
this.delegate.enableCounts(countsEnabled);
}
@Override
public boolean isStatsEnabled() {
return this.delegate.isStatsEnabled();
}
@Override
public boolean isCountsEnabled() {
return this.delegate.isCountsEnabled();
}
@Override
public void setManagedName(String name) {
this.delegate.setManagedName(name);
}
@Override
public void setManagedType(String source) {
this.delegate.setManagedType(source);
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,18 +17,20 @@
package org.springframework.integration.monitor;
import org.springframework.context.Lifecycle;
import org.springframework.integration.endpoint.management.MessageSourceMetrics;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
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 start and stop polling endpoints, for instance, in a live system.
*
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
@ManagedResource
@IntegrationManagedResource
public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Lifecycle {
private final Lifecycle lifecycle;
@@ -42,38 +44,41 @@ public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Life
}
@Override
@ManagedOperation
public void reset() {
this.delegate.reset();
}
@Override
@ManagedAttribute
public boolean isRunning() {
return this.lifecycle.isRunning();
}
@Override
@ManagedOperation
public void start() {
this.lifecycle.start();
}
@Override
@ManagedOperation
public void stop() {
this.lifecycle.stop();
}
public String getName() {
return this.delegate.getName();
@Override
public String getManagedName() {
return this.delegate.getManagedName();
}
public String getSource() {
return this.delegate.getSource();
@Override
public String getManagedType() {
return this.delegate.getManagedType();
}
/**
* @return int
* @see org.springframework.integration.monitor.MessageSourceMetrics#getMessageCount()
*/
@Override
public int getMessageCount() {
return this.delegate.getMessageCount();
}
@@ -83,4 +88,28 @@ public class LifecycleMessageSourceMetrics implements MessageSourceMetrics, Life
return this.delegate.getMessageCountLong();
}
@Override
public void enableCounts(boolean countsEnabled) {
delegate.enableCounts(countsEnabled);
}
@Override
public boolean isCountsEnabled() {
return delegate.isCountsEnabled();
}
@Override
public void setManagedName(String name) {
delegate.setManagedName(name);
}
@Override
public void setManagedType(String source) {
delegate.setManagedType(source);
}
}

View File

@@ -0,0 +1,58 @@
/*
* 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.context.Lifecycle;
import org.springframework.integration.handler.management.MessageHandlerMetrics;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.util.Assert;
/**
* Adds {@link TrackableComponent}.
*
* @author Gary Russell
* @since 4.2
*/
@IntegrationManagedResource
public class LifecycleTrackableMessageHandlerMetrics extends LifecycleMessageHandlerMetrics
implements TrackableComponent {
private final TrackableComponent trackable;
public LifecycleTrackableMessageHandlerMetrics(Lifecycle lifecycle, MessageHandlerMetrics delegate) {
super(lifecycle, delegate);
Assert.isInstanceOf(TrackableComponent.class, delegate);
this.trackable = (TrackableComponent) delegate;
}
@Override
public String getComponentName() {
return this.trackable.getComponentName();
}
@Override
public String getComponentType() {
return this.trackable.getComponentType();
}
@Override
public void setShouldTrack(boolean shouldTrack) {
this.trackable.setShouldTrack(shouldTrack);
}
}

View File

@@ -0,0 +1,58 @@
/*
* 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.context.Lifecycle;
import org.springframework.integration.endpoint.management.MessageSourceMetrics;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.util.Assert;
/**
* Adds {@link TrackableComponent}.
*
* @author Gary Russell
* @since 2.0
*/
@IntegrationManagedResource
public class LifecycleTrackableMessageSourceMetrics extends LifecycleMessageSourceMetrics
implements TrackableComponent {
private final TrackableComponent trackable;
public LifecycleTrackableMessageSourceMetrics(Lifecycle lifecycle, MessageSourceMetrics delegate) {
super(lifecycle, delegate);
Assert.isInstanceOf(TrackableComponent.class, lifecycle);
this.trackable = (TrackableComponent) lifecycle;
}
@Override
public String getComponentName() {
return this.trackable.getComponentName();
}
@Override
public String getComponentType() {
return this.trackable.getComponentType();
}
@Override
public void setShouldTrack(boolean shouldTrack) {
this.trackable.setShouldTrack(shouldTrack);
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,17 +17,18 @@ package org.springframework.integration.monitor;
import org.springframework.context.Lifecycle;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
/**
* Wrapper for an {@link AbstractEndpoint} that exposes a management interface.
*
*
* @author Dave Syer
*
* @author Gary Russell
*
*/
@ManagedResource
@IntegrationManagedResource
public class ManagedEndpoint implements Lifecycle {
private final AbstractEndpoint delegate;
@@ -36,16 +37,19 @@ public class ManagedEndpoint implements Lifecycle {
this.delegate = delegate;
}
@Override
@ManagedAttribute
public final boolean isRunning() {
return delegate.isRunning();
}
@Override
@ManagedOperation
public final void start() {
delegate.start();
}
@Override
@ManagedOperation
public final void stop() {
delegate.stop();

View File

@@ -0,0 +1,66 @@
/*
* 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 java.util.Map;
import java.util.Properties;
import org.springframework.context.Lifecycle;
import org.springframework.integration.handler.management.MessageHandlerMetrics;
import org.springframework.integration.router.MappingMessageRouterManagement;
/**
* Allows Router operations to appear in the same MBean as statistics.
*
* @author Gary Russell
* @since 4.2
*
*/
public class RouterMetrics extends LifecycleMessageHandlerMetrics implements MappingMessageRouterManagement {
private final MappingMessageRouterManagement router;
public RouterMetrics(Lifecycle lifecycle, MappingMessageRouterManagement delegate) {
super(lifecycle, (MessageHandlerMetrics) delegate);
this.router = delegate;
}
@Override
public void setChannelMapping(String key, String channelName) {
this.router.setChannelMapping(key, channelName);
}
@Override
public void removeChannelMapping(String key) {
this.router.removeChannelMapping(key);
}
@Override
public void replaceChannelMappings(Properties channelMappings) {
this.router.replaceChannelMappings(channelMappings);
}
@Override
public Map<String, String> getChannelMappings() {
return this.router.getChannelMappings();
}
@Override
public void setChannelMappings(Map<String, String> channelMappings) {
this.router.setChannelMappings(channelMappings);
}
}

View File

@@ -1,89 +0,0 @@
/*
* Copyright 2002-2014 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 java.util.concurrent.atomic.AtomicLong;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.integration.core.MessageSource;
/**
* @author Dave Syer
* @since 2.0
*/
public class SimpleMessageSourceMetrics implements MethodInterceptor, MessageSourceMetrics {
private final AtomicLong messageCount = new AtomicLong();
private final MessageSource<?> messageSource;
private volatile String source;
private volatile String name;
public SimpleMessageSourceMetrics(MessageSource<?> messageSource) {
this.messageSource = messageSource;
}
public void setName(String name) {
this.name = name;
}
public String getName() {
return this.name;
}
public void setSource(String source) {
this.source = source;
}
public String getSource() {
return this.source;
}
public MessageSource<?> getMessageSource() {
return this.messageSource;
}
public void reset() {
this.messageCount.set(0);
}
public int getMessageCount() {
return (int) this.messageCount.get();
}
public long getMessageCountLong() {
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) {
this.messageCount.incrementAndGet();
}
return result;
}
@Override
public String toString() {
return String.format("MessageSourceMonitor: [name=%s, source=%s, count=%d]", name, source, messageCount.get());
}
}

View File

@@ -0,0 +1,55 @@
/*
* 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.context.Lifecycle;
import org.springframework.integration.history.TrackableComponent;
import org.springframework.integration.router.MappingMessageRouterManagement;
import org.springframework.util.Assert;
/**
* Adds {@link TrackableComponent}.
*
* @author Gary Russell
* @since 2.0
*/
public class TrackableRouterMetrics extends RouterMetrics implements TrackableComponent {
private final TrackableComponent trackable;
public TrackableRouterMetrics(Lifecycle lifecycle, MappingMessageRouterManagement delegate) {
super(lifecycle, delegate);
Assert.isInstanceOf(TrackableComponent.class, delegate);
this.trackable = (TrackableComponent) delegate;
}
@Override
public String getComponentName() {
return this.trackable.getComponentName();
}
@Override
public String getComponentType() {
return this.trackable.getComponentType();
}
@Override
public void setShouldTrack(boolean shouldTrack) {
this.trackable.setShouldTrack(shouldTrack);
}
}

View File

@@ -182,7 +182,42 @@
The pattern is applied to all components before they are registered, looking for a match on
the 'name' property of the ObjectName. A MessageChannel and a MessageHandler (for instance)
can share a name because they have a different type, so in that case they would either both
be included or both excluded.
be included or both excluded. Since version 4.2, a leading '!' negates the pattern match
('!foo*' means don't export components where the name matches the pattern 'foo*').
For components with names that match multiple patterns, the first pattern wins.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="counts-enabled" use="optional">
<xsd:annotation>
<xsd:documentation>
Comma separated list of simple patterns for component names for which message counts
will be enabled (defaults to '*'). Only patterns that also match 'managed-components'
will be considered. Enables message counting (`sendCount`, `errorCount`, `receiveCount`)
for those components that support counters (channels, message handlers, etc).
This is the initial setting only, individual components can have counts enabled/disabled
at runtime. May be overridden by an entry in 'stats-enabled' which is additional
functionality over simple counts. If a pattern starts with `!`, counts are disabled
for matches. For components with names that match multiple patterns, the first pattern wins.
Disabling counts at runtime also disables stats.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="stats-enabled" use="optional">
<xsd:annotation>
<xsd:documentation>
Comma separated list of simple patterns for component names for which message statistics
will be enabled (response times, rates etc), as well as counts (a positive match here
overrides `counts-enabled`, you can't have statistics without counts).
(defaults to '*'). Only patterns that also match 'managed-components'
will be considered. Enables statistics for those components that support
statistics (channels - when sending, message handlers, etc).
This is the initial setting only, individual components can have stats enabled/disabled
at runtime. If a pattern starts with `!`, stats (and counts) are disabled
for matches. Note: this means that '!foo' here will disable stats
and counts for 'foo' even if counts are enabled for 'foo' in 'counts-enabled'.
For components with names that match multiple patterns, the first pattern wins.
Enabling stats at runtime also enables counts.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>

View File

@@ -64,6 +64,7 @@
<value type="java.lang.String">SendCountLong</value>
<value type="java.lang.String">SendErrorCount</value>
<value type="java.lang.String">SendErrorCountLong</value>
<value type="java.lang.String">CountsEnabled</value>
<value type="java.lang.String">StatsEnabled</value>
</array>
</constructor-arg>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013 the original author or authors.
* Copyright 2013-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.
@@ -31,15 +31,18 @@ import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Stuart Williams
* @author Gary Russell
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@DirtiesContext
public class MBeanAttributeFilterTests {
@Autowired

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -40,6 +40,7 @@ import org.springframework.integration.util.StackTraceUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -53,6 +54,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class ServiceActivatorDefaultFrameworkMethodTests {
@Autowired
@@ -148,7 +150,7 @@ public class ServiceActivatorDefaultFrameworkMethodTests {
// INT-2399
@Test
public void testMessageProcessor() {
Object processor = TestUtils.getPropertyValue(processorTestService, "handler.h.advised.targetSource.target.processor");
Object processor = TestUtils.getPropertyValue(processorTestService, "handler.processor");
assertSame(testMessageProcessor, processor);
QueueChannel replyChannel = new QueueChannel();

View File

@@ -95,7 +95,7 @@ public class UpdateMappingsTests {
MessagingTemplate messagingTemplate = new MessagingTemplate();
messagingTemplate.setReceiveTimeout(1000);
Set<ObjectName> names = this.server.queryNames(ObjectName
.getInstance("update.mapping.domain:type=HeaderValueRouter,name=router"),
.getInstance("update.mapping.domain:type=MessageHandler,name=router,bean=endpoint"),
null);
assertEquals(1, names.size());
Map<String, String> map = new HashMap<String, String>();

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -22,13 +22,15 @@ import static org.junit.Assert.assertSame;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -39,6 +41,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class AttributePollingChannelAdapterParserTests {
@Autowired

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -20,18 +20,21 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.jmx.export.MBeanExporter;
import org.springframework.messaging.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Mark Fisher
* @author Gary Russell
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class ControlBusParserTests {
@Autowired

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -21,14 +21,15 @@ import static org.junit.Assert.assertEquals;
import java.util.HashMap;
import java.util.Map;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
@@ -41,6 +42,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class DynamicRouterTests {
@Autowired
@@ -53,15 +55,18 @@ public class DynamicRouterTests {
@Autowired
@Qualifier("processAChannel")
private QueueChannel processAChannel;
private PollableChannel processAChannel;
@Autowired
@Qualifier("processBChannel")
private QueueChannel processBChannel;
private PollableChannel processBChannel;
@Autowired
@Qualifier("processCChannel")
private QueueChannel processCChannel;
private PollableChannel processCChannel;
@Autowired
private MessageChannel nullChannel;
@Test @DirtiesContext
@@ -109,4 +114,12 @@ public class DynamicRouterTests {
assertEquals("123", processCChannel.receive(0).getPayload());
}
@Test @DirtiesContext @Ignore
public void testPerf() throws Exception {
for (int i = 0; i < 10000000; i++) {
this.nullChannel.send(null);
}
System.out.println("done");
}
}

View File

@@ -19,7 +19,9 @@
server="mbs"
default-domain="tests.MBeanExpoerterParser"
object-name-static-properties="appProperties"
object-naming-strategy="keyNamer"/>
object-naming-strategy="keyNamer"
counts-enabled="foo, !baz, ba*"
stats-enabled="fiz, buz" />
<util:properties id="appProperties">
<prop key="foo">foo</prop>
@@ -28,4 +30,16 @@
<bean id="keyNamer" class="org.springframework.jmx.export.naming.KeyNamingStrategy" />
<si:channel id="foo" />
<si:channel id="bar" />
<si:channel id="baz" />
<si:channel id="qux" />
<si:channel id="fiz" />
<si:channel id="buz" />
</beans>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
package org.springframework.integration.jmx.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
@@ -30,8 +31,10 @@ import org.junit.runner.RunWith;
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.test.util.TestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -39,10 +42,12 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Gunnar Hillert
* @author Gary Russell
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class MBeanExporterParserTests {
@Autowired
@@ -59,6 +64,24 @@ public class MBeanExporterParserTests {
assertTrue(properties.containsKey("bar"));
assertEquals(server, exporter.getServer());
assertSame(context.getBean("keyNamer"), TestUtils.getPropertyValue(exporter, "namingStrategy"));
MessageChannelMetrics metrics = context.getBean("foo", MessageChannelMetrics.class);
assertTrue(metrics.isCountsEnabled());
assertFalse(metrics.isStatsEnabled());
metrics = context.getBean("bar", MessageChannelMetrics.class);
assertTrue(metrics.isCountsEnabled());
assertFalse(metrics.isStatsEnabled());
metrics = context.getBean("baz", MessageChannelMetrics.class);
assertFalse(metrics.isCountsEnabled());
assertFalse(metrics.isStatsEnabled());
metrics = context.getBean("qux", MessageChannelMetrics.class);
assertFalse(metrics.isCountsEnabled());
assertFalse(metrics.isStatsEnabled());
metrics = context.getBean("fiz", MessageChannelMetrics.class);
assertTrue(metrics.isCountsEnabled());
assertTrue(metrics.isStatsEnabled());
metrics = context.getBean("buz", MessageChannelMetrics.class);
assertTrue(metrics.isCountsEnabled());
assertTrue(metrics.isStatsEnabled());
exporter.destroy();
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013 the original author or authors.
* Copyright 2013-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
@@ -28,6 +28,7 @@ import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jmx.export.naming.KeyNamingStrategy;
import org.springframework.jmx.export.naming.ObjectNamingStrategy;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -38,6 +39,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class MBeanRegistrationCustomNamingTests {
@Autowired

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -30,6 +30,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -40,6 +41,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class MBeanRegistrationTests {
@Autowired

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013 the original author or authors.
* Copyright 2013-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.
@@ -30,6 +30,7 @@ import javax.management.QueryExp;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.endpoint.SourcePollingChannelAdapter;
@@ -37,6 +38,7 @@ import org.springframework.integration.jmx.MBeanObjectConverter;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -45,8 +47,9 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @author Gary Russell
*
*/
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class MBeanTreePollingChannelAdapterParserTests {
@Autowired

View File

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2011 the original author or authors.
*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. 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.
@@ -22,21 +22,25 @@ import javax.management.ObjectName;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class MessageStoreTests {
@Autowired
private MBeanServer server;
@Test
public void testHandlerMBeanRegistration() throws Exception {
Set<ObjectName> names = server.queryNames(new ObjectName("test.MessageStore:type=SimpleMessageStore,*"), null);

View File

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. 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.
@@ -22,22 +22,26 @@ import javax.management.ObjectName;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class MethodInvokerTests {
@Autowired
@@ -56,6 +60,7 @@ public class MethodInvokerTests {
// the router and the error handler...
assertEquals(2, names.size());
underscores.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
assertEquals("foo", message.getPayload());
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -26,13 +26,15 @@ import javax.management.Notification;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.integration.jmx.NotificationListeningMessageProducer;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.integration.jmx.NotificationListeningMessageProducer;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -43,6 +45,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class NotificationListeningChannelAdapterParserTests {
@Autowired

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -58,6 +58,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class NotificationPublishingChannelAdapterParserTests {
@Autowired

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2012 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,14 +23,16 @@ import static org.junit.Assert.fail;
import org.junit.After;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.integration.jmx.JmxHeaders;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.integration.handler.advice.AbstractRequestHandlerAdvice;
import org.springframework.integration.jmx.JmxHeaders;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -38,10 +40,12 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Artem Bilan
* @author Gary Russell
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class OperationInvokingChannelAdapterParserTests {
@Autowired

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -43,6 +43,7 @@ import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -55,6 +56,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class OperationInvokingOutboundGatewayTests {
@Autowired

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -22,16 +22,20 @@ import javax.management.ObjectName;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class PollingAdapterMBeanTests {
@Autowired

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
@@ -22,9 +22,11 @@ import javax.management.ObjectName;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.channel.QueueChannelOperations;
import org.springframework.messaging.PollableChannel;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -35,6 +37,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class PriorityChannelTests {
@Autowired

View File

@@ -1,11 +1,11 @@
/*
* 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
*
*
* 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.
@@ -14,15 +14,11 @@
package org.springframework.integration.jmx.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import javax.management.MBeanOperationInfo;
import javax.management.MBeanServer;
import javax.management.ObjectName;
@@ -31,6 +27,7 @@ import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
import org.junit.runners.Parameterized.Parameters;
import org.springframework.context.support.ClassPathXmlApplicationContext;
/**
@@ -40,9 +37,9 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
@RunWith(Parameterized.class)
public class RouterMBeanTests {
private MBeanServer server;
private final MBeanServer server;
private ClassPathXmlApplicationContext context;
private final ClassPathXmlApplicationContext context;
public RouterMBeanTests(String configLocation) {
context = new ClassPathXmlApplicationContext(configLocation, getClass());
@@ -94,26 +91,6 @@ public class RouterMBeanTests {
Set<ObjectName> names = server.queryNames(new ObjectName("test.RouterMBean:type=MessageHandler,name=ptRouter,*"), null);
assertEquals(1, names.size());
}
@Test
public void testRouterMBeanHasTrackableComponent() throws Exception {
// System.err.println(server.queryNames(new ObjectName("test.RouterMBean:*"), null));
Set<ObjectName> names = server.queryNames(new ObjectName("test.RouterMBean:type=ExpressionEvaluatingRouter,*"), null);
Map<String,MBeanOperationInfo> infos = new HashMap<String, MBeanOperationInfo>();
for (MBeanOperationInfo info : server.getMBeanInfo(names.iterator().next()).getOperations()) {
infos.put(info.getName(), info);
}
// System.err.println(infos);
assertNotNull(infos.get("setShouldTrack"));
}
@Test
public void testVanillaRouterMBeanRegistration() throws Exception {
// System.err.println(server.queryNames(new ObjectName("test.RouterMBean:*"), null));
Set<ObjectName> names = server.queryNames(new ObjectName("test.RouterMBean:type=ExpressionEvaluatingRouter,*"), null);
// The router is exposed...
assertEquals(1, names.size());
}
public static interface Service {
void send(String input);

View File

@@ -1,11 +1,11 @@
/*
* Copyright 2002-2010 the original author or authors.
*
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. 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.
@@ -22,21 +22,25 @@ import javax.management.ObjectName;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class RouterMBeanVanillaTests {
@Autowired
private MBeanServer server;
@Test
public void testHandlerMBeanRegistration() throws Exception {
// System.err.println(server.queryNames(new ObjectName("test.RouterMBeanVanilla:*"), null));

View File

@@ -75,10 +75,12 @@ public class EnableMBeanExportTests {
assertSame(this.mBeanServer, this.exporter.getServer());
String[] componentNamePatterns = TestUtils.getPropertyValue(this.exporter, "componentNamePatterns", String[].class);
for (String componentNamePattern : componentNamePatterns) {
assertThat(componentNamePattern, Matchers.isOneOf("input", "in*"));
assertThat(componentNamePattern, Matchers.not(Matchers.equalTo("*")));
}
assertThat(componentNamePatterns, Matchers.arrayContaining("input", "inputX", "in*"));
String[] enabledCounts = TestUtils.getPropertyValue(this.exporter, "enabledCountsPatterns", String[].class);
assertThat(enabledCounts, Matchers.arrayContaining("foo", "bar", "baz"));
String[] enabledStatts = TestUtils.getPropertyValue(this.exporter, "enabledStatsPatterns", String[].class);
assertThat(enabledStatts, Matchers.arrayContaining("qux", "!*"));
Set<ObjectName> names = this.mBeanServer.queryNames(ObjectName.getInstance("FOO:type=MessageChannel,*"), null);
// Only one registered (out of >2 available)
assertEquals(1, names.size());
@@ -102,7 +104,11 @@ public class EnableMBeanExportTests {
@Configuration
@EnableIntegration
@EnableIntegrationMBeanExport(server = "#{mbeanServer}", defaultDomain = "${managed.domain}", managedComponents = {"input", "${managed.component}"})
@EnableIntegrationMBeanExport(server = "#{mbeanServer}",
defaultDomain = "${managed.domain}",
managedComponents = {"input", "${managed.component}"},
countsEnabled = { "foo", "${count.patterns}" },
statsEnabled = { "qux", "!*" })
public static class ContextConfiguration {
@Bean
@@ -127,8 +133,9 @@ public class EnableMBeanExportTests {
@Override
public void initialize(GenericApplicationContext applicationContext) {
applicationContext.setEnvironment(new MockEnvironment()
.withProperty("managed.component", "input,in*")
.withProperty("managed.domain", "FOO"));
.withProperty("managed.component", "inputX,in*")
.withProperty("managed.domain", "FOO")
.withProperty("count.patterns", "bar,baz"));
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -19,18 +19,22 @@ import static org.junit.Assert.assertNotNull;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Oleg Zhurakousky
* @author Gunnar Hillert
* @author Gary Russell
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class ChainWithMessageProducingHandlersTests {
@Autowired

View File

@@ -1,11 +1,11 @@
/*
* Copyright 2009-2010 the original author or authors.
*
* Copyright 2009-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
*
* 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.
@@ -17,15 +17,23 @@ import static org.junit.Assert.assertTrue;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Dave Syer
* @author Gary Russell
* @since 2.0
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class ChannelIntegrationTests {
@Autowired
@@ -39,7 +47,7 @@ public class ChannelIntegrationTests {
@Test
public void testMessageChannelStatistics() throws Exception {
requests.send(new GenericMessage<String>("foo"));
double rate = messageChannelsMonitor.getChannelSendRate("" + requests).getMean();
@@ -47,11 +55,11 @@ public class ChannelIntegrationTests {
rate = messageChannelsMonitor.getChannelSendRate("" + intermediate).getMean();
assertTrue("No statistics for intermediate channel", rate >= 0);
assertNotNull(intermediate.receive(100L));
double count = messageChannelsMonitor.getChannelReceiveCount("" + intermediate);
assertTrue("No statistics for intermediate channel", count >= 0);
}
}

View File

@@ -12,16 +12,11 @@
*/
package org.springframework.integration.monitor;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertThat;
import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.junit.Before;
import org.junit.Test;
import org.springframework.aop.framework.Advised;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.NameMatchMethodPointcutAdvisor;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
@@ -57,19 +52,6 @@ public class MessageMetricsAdviceTests {
handler = new DummyHandler();
}
@Test
public void adviceExportedHandler() throws Exception {
Advised advised = (Advised) mBeanExporter.postProcessAfterInitialization(handler, "test");
DummyInterceptor interceptor = new DummyInterceptor();
advised.addAdvice(interceptor);
assertThat(advised.getAdvisors().length, equalTo(2));
((MessageHandler) advised).handleMessage(MessageBuilder.withPayload("test").build());
assertThat(interceptor.invoked, is(true));
}
@Test
public void exportAdvisedHandler() throws Exception {

View File

@@ -0,0 +1,43 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:int-jmx="http://www.springframework.org/schema/integration/jmx"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/integration/jmx http://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd">
<context:mbean-server />
<int-jmx:mbean-export />
<int:channel id="input">
<int:queue />
</int:channel>
<int:bridge input-channel="input" output-channel="next">
<int:poller fixed-delay="500" />
</int:bridge>
<int:service-activator input-channel="next">
<bean class="org.springframework.integration.monitor.MonitorTests$TestHandler" />
</int:service-activator>
<int:inbound-channel-adapter channel="pubsub">
<bean class="org.springframework.integration.monitor.MonitorTests$TestSource" />
<int:poller fixed-delay="500" />
</int:inbound-channel-adapter>
<int:publish-subscribe-channel id="pubsub" />
<int:bridge input-channel="pubsub" output-channel="nullChannel" />
<int:bridge input-channel="pubsub" output-channel="output" />
<int:channel id="output">
<int:queue />
</int:channel>
</beans>

View File

@@ -0,0 +1,127 @@
/*
* 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 static org.hamcrest.Matchers.greaterThan;
import static org.hamcrest.Matchers.greaterThanOrEqualTo;
import static org.hamcrest.Matchers.lessThan;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.NullChannel;
import org.springframework.integration.channel.PublishSubscribeChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.messaging.Message;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Gary Russell
* @since 4.2
*
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class MonitorTests {
@Autowired
private QueueChannel input;
@Autowired
private DirectChannel next;
@Autowired
private TestHandler handler;
@Autowired
private TestSource source;
@Autowired
private PublishSubscribeChannel pubsub;
@Autowired
private QueueChannel output;
@Autowired
private NullChannel nullChannel;
@Test
public void testStats() {
Integer active = new MessagingTemplate(this.input).convertSendAndReceive("foo", Integer.class);
assertEquals(1, active.intValue());
assertEquals(0, this.handler.getActiveCount());
assertEquals(1, this.handler.getHandleCount());
assertThat(this.handler.getDuration().getMax(), greaterThan(99.0));
assertThat(this.handler.getDuration().getMax(), lessThan(10000.0));
assertEquals(1, this.input.getSendCount());
assertEquals(1, this.input.getReceiveCount());
assertEquals(1, this.next.getSendCount());
assertThat(this.next.getSendDuration().getMax(), greaterThan(99.0));
assertThat(this.next.getSendDuration().getMax(), lessThan(10000.0));
Message<?> fromInbound = this.output.receive(10000);
assertNotNull(fromInbound);
assertEquals(0, fromInbound.getPayload());
fromInbound = this.output.receive(10000);
assertNotNull(fromInbound);
assertEquals(1, fromInbound.getPayload());
assertThat(this.source.getMessageCount(), greaterThanOrEqualTo(2));
assertThat(this.nullChannel.getSendCount(), greaterThanOrEqualTo(2));
assertThat(this.pubsub.getSendCount(), greaterThanOrEqualTo(2));
}
public static class TestHandler extends AbstractReplyProducingMessageHandler {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
try {
Thread.sleep(100);
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
return getActiveCount();
}
}
public static class TestSource extends AbstractMessageSource<Integer> {
@Override
public String getComponentType() {
return "foo";
}
@Override
protected Object doReceive() {
return getMessageCount();
}
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2013 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,11 +23,13 @@ import javax.management.Notification;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -45,6 +47,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
@DirtiesContext
public class RemoteMBeanServerTests {
@Autowired
@@ -97,10 +100,12 @@ public class RemoteMBeanServerTests {
public static class Tester implements TesterMBean {
@Override
public String getFoo() {
return "foo";
}
@Override
public String fooBar(String foo) {
return "bar";
}

View File

@@ -46,9 +46,9 @@ import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.scattergather.ScatterGatherHandler;
import org.springframework.integration.jmx.config.EnableIntegrationMBeanExport;
import org.springframework.integration.router.RecipientListRouter;
import org.springframework.integration.scattergather.ScatterGatherHandler;
import org.springframework.integration.store.SimpleMessageStore;
import org.springframework.jmx.support.MBeanServerFactoryBean;
import org.springframework.messaging.Message;
@@ -63,6 +63,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Artem Bilan
* @author Gary Russell
* @since 4.1
*/
@ContextConfiguration
@@ -94,6 +95,7 @@ public class ScatterGatherHandlerIntegrationTests {
assertThat((Double) payload, lessThan(10D));
}
@SuppressWarnings("rawtypes")
@Test
public void testAuctionWithGatherChannel() {
Message<String> quoteMessage = MessageBuilder.withPayload("testQuote").build();

View File

@@ -12,7 +12,7 @@
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
<int:inbound-channel-adapter channel="toControlBus" auto-startup="false"
expression="@integrationMbeanExporter.stopActiveComponents(false, 1000)" />
expression="@integrationMbeanExporter.stopActiveComponents(1000)" />
<int:channel id="toControlBus" />

View File

@@ -54,22 +54,17 @@ public class Int2307Tests {
int count = 0;
for (ObjectInstance mbean : mbeans) {
Thread.sleep(500); //Added in order to pass test with Java 8
if (mbean.toString().startsWith("org.springframework.integration.router.RecipientListRouter[test.domain:type=RecipientListRouter,name=rlr,random=")) {
bits |= 1;
count++;
} else if (mbean.toString().startsWith("org.springframework.integration.monitor.LifecycleMessageHandlerMetrics[test.domain:type=MessageHandler,name=rlr,bean=endpoint,random=")) {
if (mbean.toString().startsWith("org.springframework.integration.monitor.LifecycleTrackableMessageHandlerMetrics[test.domain:type=MessageHandler,name=rlr,bean=endpoint,random=")) {
bits |= 2;
count++;
} else if (mbean.toString().startsWith("org.springframework.integration.router.HeaderValueRouter[test.domain:type=HeaderValueRouter,name=hvr,random=")) {
bits |= 4;
count++;
} else if (mbean.toString().startsWith("org.springframework.integration.monitor.LifecycleMessageHandlerMetrics[test.domain:type=MessageHandler,name=hvr,bean=endpoint,random=")) {
}
else if (mbean.toString().startsWith("org.springframework.integration.monitor.TrackableRouterMetrics[test.domain:type=MessageHandler,name=hvr,bean=endpoint,random=")) {
bits |= 8;
count++;
}
}
assertEquals(0xf, bits);
assertEquals(4, count);
assertEquals(0xa, bits);
assertEquals(2, count);
Class<?> clazz = Class.forName("org.springframework.integration.jmx.config.MBeanExporterHelper");
List<Object> beanPostProcessors = TestUtils.getPropertyValue(context, "beanFactory.beanPostProcessors", List.class);
@@ -85,7 +80,7 @@ public class Int2307Tests {
assertTrue(TestUtils.getPropertyValue(mBeanExporterHelper, "siBeanNames", Set.class).contains("zz"));
// make sure there are no duplicate MBean ObjectNames if 2 contexts loaded from same config
new ClassPathXmlApplicationContext("single-config.xml", this.getClass());
new ClassPathXmlApplicationContext("single-config.xml", this.getClass()).close();
}
@SuppressWarnings("unchecked")

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2014 the original author or authors.
* Copyright 2002-2015 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,6 +23,7 @@ import java.util.concurrent.locks.ReentrantLock;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.integration.mqtt.support.MqttMessageConverter;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.jmx.export.annotation.ManagedAttribute;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.export.annotation.ManagedResource;
@@ -37,6 +38,7 @@ import org.springframework.util.Assert;
*
*/
@ManagedResource
@IntegrationManagedResource
public abstract class AbstractMqttMessageDrivenChannelAdapter extends MessageProducerSupport {
private final String url;

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2014 the original author or authors
* Copyright 2014-2015 the original author or authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -31,6 +31,7 @@ import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.redis.event.RedisExceptionEvent;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
@@ -42,9 +43,11 @@ import org.springframework.util.Assert;
/**
* @author David Liu
* @author Artem Bilan
* @author Gary Russell
* @since 4.1
*/
@ManagedResource
@IntegrationManagedResource
public class RedisQueueInboundGateway extends MessagingGatewaySupport implements ApplicationEventPublisherAware {
private static final String QUEUE_NAME_SUFFIX = ".reply";

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2013 the original author or authors
* Copyright 2013-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.
@@ -31,6 +31,7 @@ import org.springframework.integration.channel.MessagePublishingErrorHandler;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.redis.event.RedisExceptionEvent;
import org.springframework.integration.support.channel.BeanFactoryChannelResolver;
import org.springframework.integration.support.management.IntegrationManagedResource;
import org.springframework.integration.util.ErrorHandlingTaskExecutor;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
@@ -44,9 +45,11 @@ import org.springframework.util.Assert;
* @author Mark Fisher
* @author Gunnar Hillert
* @author Artem Bilan
* @author Gary Russell
* @since 3.0
*/
@ManagedResource
@IntegrationManagedResource
public class RedisQueueMessageDrivenEndpoint extends MessageProducerSupport implements ApplicationEventPublisherAware {
public static final long DEFAULT_RECEIVE_TIMEOUT = 1000;