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

INT-1507: remove control bus from JMX XSD

Browse Source
This commit is contained in:
Dave Syer
2010-10-25 10:58:20 -07:00
parent 53dd793819
commit 473b8fa96d
13 changed files with 37 additions and 725 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/ControlBus.java
View File

@@ -1,101 +0,0 @@
/*
* 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.
*/
package org.springframework.integration.jmx;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.monitor.ObjectNameLocator;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
import javax.management.MBeanServer;
/**
* JMX-based Control Bus implementation. Routes control messages on an operation channel to the other control points
* (channels and handlers) via JMX. To use the control bus send a message to the operation channel with a header
* {@link #TARGET_BEAN_NAME} equal to the bean name of the channel or endpoint you want to target. Include also a header
* {@link JmxHeaders#OPERATION_NAME} to specify the operation you want to invoke and a message payload containing the
* arguments (if any).
*
* @author Mark Fisher
* @since 2.0
*/
public class ControlBus implements BeanFactoryAware, InitializingBean {
public static final String TARGET_BEAN_NAME = JmxHeaders.PREFIX + "_controlBus_targetBeanName";
private final SubscribableChannel operationChannel;
private volatile ListableBeanFactory beanFactory;
private final ObjectNameLocator exporter;
private final MBeanServer server;
/**
* Create a {@link ControlBus}.
*/
public ControlBus(ObjectNameLocator locator, MBeanServer server, SubscribableChannel operationChannel) {
this.exporter = locator;
this.server = server;
this.operationChannel = operationChannel;
}
/**
* Returns the channel to which operation-invoking Messages may be sent. Any messages sent to this channel must
* contain {@link ControlBus#TARGET_BEAN_NAME} and {@link JmxHeaders#OPERATION_NAME} header values, and the target
* bean name must match one that has been exported by this Control Bus. If the operation returns a result, the
* {@link MessageHeaders#REPLY_CHANNEL} header is also required.
*/
public SubscribableChannel getOperationChannel() {
return this.operationChannel;
}
public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
Assert.isTrue(beanFactory instanceof ListableBeanFactory, "A ListableBeanFactory is required.");
this.beanFactory = (ListableBeanFactory) beanFactory;
}
public void afterPropertiesSet() throws Exception {
OperationInvokingMessageHandler handler = new ControlBusOperationInvokingMessageHandler();
handler.setBeanFactory(this.beanFactory);
handler.setServer(this.server);
handler.afterPropertiesSet();
this.operationChannel.subscribe(handler);
}
private class ControlBusOperationInvokingMessageHandler extends OperationInvokingMessageHandler {
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
String beanName = requestMessage.getHeaders().get(TARGET_BEAN_NAME, String.class);
Assert.notNull(beanName, "The ControlBus.TARGET_BEAN_NAME is required.");
String objectName = exporter.getObjectName(beanName);
Assert.notNull(objectName, "ControlBus has not exported an MBean for '" + beanName + "'");
requestMessage = MessageBuilder.fromMessage(requestMessage).setHeader(JmxHeaders.OBJECT_NAME, objectName)
.setHeader(TARGET_BEAN_NAME, null).build();
return super.handleRequestMessage(requestMessage);
}
}
}

55
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/ControlBusFactoryBean.java
View File

@@ -1,55 +0,0 @@
/*
* 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.
*/
package org.springframework.integration.jmx.config;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.jmx.ControlBus;
import org.springframework.integration.core.SubscribableChannel;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
/**
* @author Dave Syer
* @since 2.0
*
*/
public class ControlBusFactoryBean implements FactoryBean<ControlBus> {
private final SubscribableChannel operationChannel;
private final IntegrationMBeanExporter exporter;
public ControlBusFactoryBean(IntegrationMBeanExporter exporter, SubscribableChannel operationChannel) {
this.exporter = exporter;
this.operationChannel = operationChannel;
}
public ControlBusFactoryBean(IntegrationMBeanExporter exporter) {
this(exporter, new DirectChannel());
}
public ControlBus getObject() throws Exception {
return new ControlBus(exporter, exporter.getServer(), operationChannel);
}
public Class<?> getObjectType() {
return ControlBus.class;
}
public boolean isSingleton() {
return true;
}
}

51
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/ControlBusParser.java
View File

@@ -1,51 +0,0 @@
/*
* 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.
*/
package org.springframework.integration.jmx.config;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* @author Mark Fisher
* @since 2.0
*/
public class ControlBusParser extends AbstractSingleBeanDefinitionParser {
@Override
protected boolean shouldGenerateIdAsFallback() {
return true;
}
@Override
protected String getBeanClassName(Element element) {
return "org.springframework.integration.jmx.config.ControlBusFactoryBean";
}
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
builder.getRawBeanDefinition().setSource(parserContext.extractSource(element));
builder.addConstructorArgReference(element.getAttribute("mbean-exporter"));
if (StringUtils.hasLength(element.getAttribute("operation-channel"))) {
builder.addConstructorArgReference(element.getAttribute("operation-channel"));
}
}
}

2
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/JmxNamespaceHandler.java
View File

@@ -23,6 +23,7 @@ import org.springframework.integration.config.xml.AbstractIntegrationNamespaceHa
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @author Dave Syer
* @since 2.0
*/
public class JmxNamespaceHandler extends AbstractIntegrationNamespaceHandler {
@@ -34,7 +35,6 @@ public class JmxNamespaceHandler extends AbstractIntegrationNamespaceHandler {
this.registerBeanDefinitionParser("notification-listening-channel-adapter", new NotificationListeningChannelAdapterParser());
this.registerBeanDefinitionParser("notification-publishing-channel-adapter", new NotificationPublishingChannelAdapterParser());
this.registerBeanDefinitionParser("mbean-export", new MBeanExporterParser());
this.registerBeanDefinitionParser("control-bus", new ControlBusParser());
}
}

39
spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.0.xsd
View File

@@ -117,45 +117,6 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="control-bus">
<xsd:annotation>
<xsd:documentation>
Control bus accepts control messages for channels and endpoints on an (optional) "operation
channel".
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:ID" use="optional" />
<xsd:attribute name="mbean-exporter" use="required">
<xsd:annotation>
<xsd:documentation>
A reference to the MBeanExporter created using &lt;mbean-exporter/&gt; in this namespace.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:expected-type type="org.springframework.integration.monitor.IntegrationMBeanExporter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="operation-channel" use="optional">
<xsd:annotation>
<xsd:documentation>
If provided a control bus will be created and subscribed to the Message Channel.
The channel can
then be used to send operation commands to
this Control Bus. It must implement SubscribableChannel.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation>
<tool:expected-type type="org.springframework.integration.core.SubscribableChannel" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
<xsd:complexType name="adapterType">
<xsd:annotation>
<xsd:documentation>

35
spring-integration-jmx/src/test/java/org/springframework/integration/IgnoredTestSuite.java
View File

@@ -4,33 +4,29 @@ import org.junit.Ignore;
import org.junit.runner.RunWith;
import org.junit.runners.Suite;
import org.junit.runners.Suite.SuiteClasses;
import org.springframework.integration.control.ControlBusTests;
import org.springframework.integration.control.ControlBusXmlTests;
import org.springframework.integration.config.xml.ControlBusTests;
import org.springframework.integration.jmx.AttributePollingMessageSourceTests;
import org.springframework.integration.jmx.NotificationListeningMessageProducerTests;
import org.springframework.integration.jmx.OperationInvokingMessageHandlerTests;
import org.springframework.integration.jmx.config.NotificationListeningChannelAdapterParserTests;
import org.springframework.integration.jmx.config.OperationInvokingChannelAdapterParserTests;
import org.springframework.integration.jmx.config.OperationInvokingOutboundGatewayTests;
import org.springframework.integration.monitor.ExponentialMovingAverageTests;
import org.springframework.integration.monitor.ExponentialMovingAverageRatioTests;
import org.springframework.integration.monitor.ExponentialMovingAverageTests;
import org.springframework.integration.monitor.HandlerMonitoringIntegrationTests;
import org.springframework.integration.monitor.MessageChannelsMonitorIntegrationTests;
/*
* Copyright 2009-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.
*
* 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.
*/
/**
@@ -40,11 +36,10 @@ import org.springframework.integration.monitor.MessageChannelsMonitorIntegration
*
*/
@RunWith(Suite.class)
@SuiteClasses(value = { OperationInvokingMessageHandlerTests.class,
ExponentialMovingAverageTests.class, OperationInvokingChannelAdapterParserTests.class,
HandlerMonitoringIntegrationTests.class, NotificationListeningMessageProducerTests.class,
OperationInvokingOutboundGatewayTests.class, NotificationListeningChannelAdapterParserTests.class,
ControlBusXmlTests.class, ExponentialMovingAverageRatioTests.class,
@SuiteClasses(value = { OperationInvokingMessageHandlerTests.class, ExponentialMovingAverageTests.class,
OperationInvokingChannelAdapterParserTests.class, HandlerMonitoringIntegrationTests.class,
NotificationListeningMessageProducerTests.class, OperationInvokingOutboundGatewayTests.class,
NotificationListeningChannelAdapterParserTests.class, ExponentialMovingAverageRatioTests.class,
AttributePollingMessageSourceTests.class, ControlBusTests.class, MessageChannelsMonitorIntegrationTests.class })
@Ignore
public class IgnoredTestSuite {

101
spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusOperationChannelTests.java
View File

@@ -1,101 +0,0 @@
/*
* 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.
*/
package org.springframework.integration.control;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import org.junit.After;
import org.junit.Test;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.jmx.ControlBus;
import org.springframework.integration.jmx.JmxHeaders;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.jmx.support.MBeanServerFactoryBean;
/**
* @author Mark Fisher
* @since 2.0
*/
public class ControlBusOperationChannelTests {
private final String domain = "domain.test";
private GenericApplicationContext context;
@Test
public void replyProducingOperation() throws Exception {
context = new GenericApplicationContext();
RootBeanDefinition endpointDef = new RootBeanDefinition(EventDrivenConsumer.class);
endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new DirectChannel());
endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RootBeanDefinition(TestHandler.class));
context.registerBeanDefinition("testEndpoint", endpointDef);
registerControlBus(context, domain);
context.refresh();
ControlBus controlBus = context.getBean("controlBus", ControlBus.class);
EventDrivenConsumer endpoint = context.getBean("testEndpoint", EventDrivenConsumer.class);
Message<?> stopMessage = MessageBuilder.withPayload("test")
.setHeader(ControlBus.TARGET_BEAN_NAME, "testEndpoint")
.setHeader(JmxHeaders.OPERATION_NAME, "stop")
.build();
Message<?> startMessage = MessageBuilder.withPayload("test")
.setHeader(ControlBus.TARGET_BEAN_NAME, "testEndpoint")
.setHeader(JmxHeaders.OPERATION_NAME, "start")
.build();
assertTrue("endpoint should be running after startup", endpoint.isRunning());
controlBus.getOperationChannel().send(stopMessage);
assertFalse("endpoint should have been stopped", endpoint.isRunning());
controlBus.getOperationChannel().send(startMessage);
assertTrue("endpoint should be running after being restarted", endpoint.isRunning());
close();
}
@After
public void close() {
context.close();
}
private BeanDefinition registerControlBus(GenericApplicationContext context, String domain) {
RootBeanDefinition serverDef = new RootBeanDefinition(MBeanServerFactoryBean.class);
serverDef.getPropertyValues().add("locateExistingServerIfPossible", true);
context.registerBeanDefinition("mbeanServer", serverDef);
BeanDefinition exporterDef = new RootBeanDefinition(IntegrationMBeanExporter.class);
exporterDef.getPropertyValues().addPropertyValue("server", new RuntimeBeanReference("mbeanServer"));
exporterDef.getPropertyValues().addPropertyValue("defaultDomain", domain);
context.registerBeanDefinition("exporter", exporterDef);
BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class);
controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer"));
controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("exporter"));
controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new DirectChannel());
context.registerBeanDefinition("controlBus", controlBusDef);
return exporterDef;
}
private static class TestHandler implements MessageHandler {
public void handleMessage(Message<?> message) {
}
}
}

201
spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java
View File

@@ -1,201 +0,0 @@
/*
* 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.
*/
package org.springframework.integration.control;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import java.util.Collections;
import java.util.concurrent.atomic.AtomicInteger;
import javax.management.MBeanServer;
import javax.management.MBeanServerFactory;
import javax.management.ObjectInstance;
import javax.management.ObjectName;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.config.RuntimeBeanReference;
import org.springframework.beans.factory.support.RootBeanDefinition;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.jmx.ControlBus;
import org.springframework.integration.monitor.IntegrationMBeanExporter;
import org.springframework.integration.monitor.LifecycleMessageHandlerMetrics;
import org.springframework.integration.monitor.QueueChannelMetrics;
import org.springframework.integration.monitor.DirectChannelMetrics;
import org.springframework.integration.scheduling.PollerMetadata;
import org.springframework.jmx.support.MBeanServerFactoryBean;
import org.springframework.jmx.support.ObjectNameManager;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.PeriodicTrigger;
/**
* @author Mark Fisher
* @since 2.0
*/
public class ControlBusTests {
private volatile GenericApplicationContext context;
@Before
public void createContext() {
this.context = new GenericApplicationContext();
RootBeanDefinition serverDef = new RootBeanDefinition(MBeanServerFactoryBean.class);
serverDef.getPropertyValues().add("locateExistingServerIfPossible", true);
context.registerBeanDefinition("mbeanServer", serverDef);
}
@After
public void closeContext() {
MBeanServer mbeanServer = this.context.getBean("mbeanServer", MBeanServer.class);
try {
MBeanServerFactory.releaseMBeanServer(mbeanServer);
}
catch (Exception e) {
// ignore
}
this.context.close();
}
@Test
public void directChannelRegistered() throws Exception {
context.registerBeanDefinition("directChannel", new RootBeanDefinition(DirectChannel.class));
registerControlBus(context, "domain.test1");
context.refresh();
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager
.getInstance("domain.test1:type=MessageChannel,name=directChannel"));
assertEquals(DirectChannelMetrics.class.getName(), instance.getClassName());
}
@Test
public void anonymousDirectChannelRegistered() throws Exception {
context.registerBeanDefinition("org.springframework.integration.generated#0", new RootBeanDefinition(
DirectChannel.class));
registerControlBus(context, "domain.test1b");
context.refresh();
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer
.getObjectInstance(ObjectNameManager
.getInstance("domain.test1b:type=MessageChannel,name=org.springframework.integration.generated#0,source=anonymous"));
assertEquals(DirectChannelMetrics.class.getName(), instance.getClassName());
}
@Test
public void staticObjectNamePropertiesRegistered() throws Exception {
context.registerBeanDefinition("directChannel", new RootBeanDefinition(DirectChannel.class));
BeanDefinition exporterDef = registerControlBus(context, "domain.test1a");
exporterDef.getPropertyValues().add("objectNameStaticProperties", Collections.singletonMap("foo", "bar"));
context.refresh();
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager
.getInstance("domain.test1a:type=MessageChannel,name=directChannel,foo=bar"));
assertEquals(DirectChannelMetrics.class.getName(), instance.getClassName());
}
@Test
public void queueChannelRegistered() throws Exception {
context.registerBeanDefinition("queueChannel", new RootBeanDefinition(QueueChannel.class));
registerControlBus(context, "domain.test2");
context.refresh();
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager
.getInstance("domain.test2:type=MessageChannel,name=queueChannel"));
assertEquals(QueueChannelMetrics.class.getName(), instance.getClassName());
}
@Test
public void eventDrivenConsumerRegistered() throws Exception {
context.registerBeanDefinition("testChannel", new RootBeanDefinition(DirectChannel.class));
RootBeanDefinition endpointDef = new RootBeanDefinition(EventDrivenConsumer.class);
endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("testChannel"));
endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RootBeanDefinition(BridgeHandler.class));
context.registerBeanDefinition("eventDrivenConsumer", endpointDef);
registerControlBus(context, "domain.test3");
context.refresh();
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager
.getInstance("domain.test3:type=MessageHandler,name=eventDrivenConsumer,bean=endpoint"));
assertEquals(LifecycleMessageHandlerMetrics.class.getName(), instance.getClassName());
}
@Test
public void pollingConsumerRegistered() throws Exception {
context.registerBeanDefinition("testChannel", new RootBeanDefinition(QueueChannel.class));
RootBeanDefinition endpointDef = new RootBeanDefinition(PollingConsumer.class);
endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("testChannel"));
endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RootBeanDefinition(BridgeHandler.class));
RootBeanDefinition pollerMetaDefinition = new RootBeanDefinition(PollerMetadata.class);
pollerMetaDefinition.getPropertyValues().add("trigger", new PeriodicTrigger(10000));
endpointDef.getPropertyValues().add("pollerMetadata", pollerMetaDefinition);
context.registerBeanDefinition("pollingConsumer", endpointDef);
context.registerBeanDefinition("taskScheduler", new RootBeanDefinition(ThreadPoolTaskScheduler.class));
registerControlBus(context, "domain.test4");
context.refresh();
MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class);
ObjectInstance instance = mbeanServer.getObjectInstance(ObjectNameManager
.getInstance("domain.test4:type=MessageHandler,name=pollingConsumer,bean=endpoint"));
assertEquals(LifecycleMessageHandlerMetrics.class.getName(), instance.getClassName());
}
@Test
public void channelMonitoring() throws Exception {
RootBeanDefinition channelDef = new RootBeanDefinition(DirectChannel.class);
context.registerBeanDefinition("testChannel", channelDef);
RootBeanDefinition endpointDef = new RootBeanDefinition(EventDrivenConsumer.class);
endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("testChannel"));
endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new MessageHandler() {
private final AtomicInteger count = new AtomicInteger();
public void handleMessage(Message<?> message) {
int current = count.incrementAndGet();
if (current % 10 == 0) {
throw new RuntimeException("intentional test failure");
}
}
});
context.registerBeanDefinition("testEndpoint", endpointDef);
registerControlBus(context, "domain.channel.monitor");
context.refresh();
MBeanServer server = context.getBean("mbeanServer", MBeanServer.class);
ObjectName objectName = ObjectNameManager.getInstance("domain.channel.monitor:type=MessageChannel,name=testChannel");
assertNotNull(server.getObjectInstance(objectName));
}
private BeanDefinition registerControlBus(GenericApplicationContext context, String domain) {
BeanDefinition exporterDef = new RootBeanDefinition(IntegrationMBeanExporter.class);
exporterDef.getPropertyValues().addPropertyValue("server", new RuntimeBeanReference("mbeanServer"));
exporterDef.getPropertyValues().addPropertyValue("defaultDomain", domain);
context.registerBeanDefinition("exporter", exporterDef);
BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class);
controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer"));
controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("exporter"));
controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new DirectChannel());
context.registerBeanDefinition("controlBus", controlBusDef);
return exporterDef;
}
}

39
spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests-context.xml
View File

@@ -1,39 +0,0 @@
<?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:int="http://www.springframework.org/schema/integration" xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd">
<int:channel id="controlChannel" />
<int:channel id="testDirectChannel" />
<int:channel id="testQueueChannel">
<int:queue capacity="99" />
</int:channel>
<int:bridge id="testEventDrivenBridge" input-channel="testDirectChannel" />
<int:bridge input-channel="testDirectChannel" />
<int:bridge id="testPollingBridge" input-channel="testQueueChannel" output-channel="nullChannel">
<int:poller max-messages-per-poll="1" fixed-rate="10000" />
</int:bridge>
<bean id="controlBus" class="org.springframework.integration.jmx.ControlBus">
<constructor-arg ref="mbeanExporter" />
<constructor-arg ref="mbeanServer" />
<constructor-arg ref="controlChannel" />
</bean>
<bean id="mbeanExporter" class="org.springframework.integration.monitor.IntegrationMBeanExporter">
<property name="server" ref="mbeanServer" />
<property name="defaultDomain" value="control.bus.xml.test" />
</bean>
<bean id="mbeanServer" class="org.springframework.jmx.support.MBeanServerFactoryBean">
<property name="locateExistingServerIfPossible" value="true" />
</bean>
</beans>

86
spring-integration-jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests.java
View File

@@ -1,86 +0,0 @@
/*
* 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.
*/
package org.springframework.integration.control;
import static org.junit.Assert.assertEquals;
import java.util.Set;
import javax.management.MBeanServer;
import javax.management.ObjectInstance;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.integration.monitor.LifecycleMessageHandlerMetrics;
import org.springframework.integration.monitor.QueueChannelMetrics;
import org.springframework.integration.monitor.DirectChannelMetrics;
import org.springframework.jmx.support.ObjectNameManager;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
/**
* @author Mark Fisher
*/
@ContextConfiguration
@RunWith(SpringJUnit4ClassRunner.class)
public class ControlBusXmlTests {
private static final String DOMAIN = "control.bus.xml.test";
@Autowired
private MBeanServer mbeanServer;
@Test
public void directChannelRegistered() throws Exception {
ObjectInstance instance = mbeanServer.getObjectInstance(
ObjectNameManager.getInstance(DOMAIN + ":type=MessageChannel,name=testDirectChannel"));
assertEquals(DirectChannelMetrics.class.getName(), instance.getClassName());
}
@Test
public void queueChannelRegistered() throws Exception {
ObjectInstance instance = mbeanServer.getObjectInstance(
ObjectNameManager.getInstance(DOMAIN + ":type=MessageChannel,name=testQueueChannel"));
assertEquals(QueueChannelMetrics.class.getName(), instance.getClassName());
}
@Test
public void eventDrivenConsumerRegistered() throws Exception {
ObjectInstance instance = mbeanServer.getObjectInstance(
ObjectNameManager.getInstance(DOMAIN + ":type=MessageHandler,name=testEventDrivenBridge,bean=endpoint"));
assertEquals(LifecycleMessageHandlerMetrics.class.getName(), instance.getClassName());
}
@Test
public void anonymousConsumerRegistered() throws Exception {
Set<ObjectInstance> instances = mbeanServer.queryMBeans(
ObjectNameManager.getInstance(DOMAIN + ":type=MessageHandler,bean=anonymous,*"), null);
assertEquals(1, instances.size());
assertEquals(LifecycleMessageHandlerMetrics.class.getName(), instances.iterator().next().getClassName());
}
@Test
public void pollingConsumerRegistered() throws Exception {
ObjectInstance instance = mbeanServer.getObjectInstance(
ObjectNameManager.getInstance(DOMAIN + ":type=MessageHandler,name=testPollingBridge,bean=endpoint"));
assertEquals(LifecycleMessageHandlerMetrics.class.getName(), instance.getClassName());
}
}

4
spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/ControlBusParserTests-context.xml
View File

@@ -17,7 +17,9 @@
<si:channel id="testChannel"/>
<jmx:control-bus mbean-exporter="mbeanExporter" operation-channel="testChannel"/>
<si:channel id="controlChannel"/>
<si:control-bus input-channel="controlChannel"/>
<jmx:mbean-export id="mbeanExporter" server="mbs" default-domain="tests.ControlBusParser"/>

40
spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/ControlBusParserTests.java
View File

@@ -1,31 +1,26 @@
/*
* 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.
*
* 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.jmx.config;
import static org.junit.Assert.assertEquals;
import javax.management.MBeanServer;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.jmx.ControlBus;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.core.MessagingTemplate;
import org.springframework.jmx.export.MBeanExporter;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@@ -42,12 +37,13 @@ public class ControlBusParserTests {
private ApplicationContext context;
@Test
public void test() throws InterruptedException {
ControlBus controlBus = this.context.getBean(ControlBus.class);
assertEquals(controlBus.getOperationChannel(), this.context.getBean("testChannel"));
MBeanServer server = this.context.getBean("mbs", MBeanServer.class);
MBeanExporter exporter = (MBeanExporter) new DirectFieldAccessor(controlBus).getPropertyValue("exporter");
assertEquals(server, exporter.getServer());
public void testControlMessageToChannelMetrics() throws InterruptedException {
MessageChannel control = this.context.getBean("controlChannel", MessageChannel.class);
MessagingTemplate messagingTemplate = new MessagingTemplate();
Object value = messagingTemplate.convertSendAndReceive(control,
"@mbeanExporter.getChannelSendRate('testChannel').count");
assertEquals(new Integer(0), value);
MBeanExporter exporter = this.context.getBean(MBeanExporter.class);
exporter.destroy();
}

8
spring-integration-jmx/src/test/java/org/springframework/integration/monitor/ChainWithMessageProducingHandlersTests-context.xml
View File

@@ -6,14 +6,6 @@
xmlns:int="http://www.springframework.org/schema/integration"
xmlns:p="http://www.springframework.org/schema/p">
<int:channel id="controlChannel" />
<bean id="controlBus" class="org.springframework.integration.jmx.ControlBus">
<constructor-arg ref="mbeanExporter" />
<constructor-arg ref="mbeanServer" />
<constructor-arg ref="controlChannel" />
</bean>
<bean id="mbeanExporter"
class="org.springframework.integration.monitor.IntegrationMBeanExporter"
p:server-ref="mbeanServer" p:defaultDomain="forum" />