From a80717411874f4f97f4a209ab014e62c960b7832 Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 8 Mar 2010 21:46:38 +0000 Subject: [PATCH] INT-627 First commit for Control Bus. At this point, it is simply exporting channels and endpoints as MBeans. Next up: support for Messages on its own channel (e.g. invoking those MBean operations via Messages). --- .../integration/control/ControlBus.java | 192 ++++++++++++++++++ .../control/MessageChannelInfo.java | 33 +++ .../control/PollableChannelInfo.java | 33 +++ .../integration/control/QueueChannelInfo.java | 38 ++++ .../integration/control/ControlBusTests.java | 137 +++++++++++++ .../control/ControlBusXmlTests-context.xml | 32 +++ .../control/ControlBusXmlTests.java | 77 +++++++ 7 files changed, 542 insertions(+) create mode 100644 org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/ControlBus.java create mode 100644 org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/MessageChannelInfo.java create mode 100644 org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/PollableChannelInfo.java create mode 100644 org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/QueueChannelInfo.java create mode 100644 org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java create mode 100644 org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests-context.xml create mode 100644 org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests.java diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/ControlBus.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/ControlBus.java new file mode 100644 index 0000000000..45ad9075e0 --- /dev/null +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/ControlBus.java @@ -0,0 +1,192 @@ +/* + * 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 java.lang.reflect.Method; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +import javax.management.MBeanServer; +import javax.management.MalformedObjectNameException; +import javax.management.ObjectName; + +import org.springframework.beans.BeansException; +import org.springframework.beans.factory.BeanFactory; +import org.springframework.beans.factory.BeanFactoryAware; +import org.springframework.beans.factory.BeanFactoryUtils; +import org.springframework.beans.factory.BeanInitializationException; +import org.springframework.beans.factory.InitializingBean; +import org.springframework.beans.factory.ListableBeanFactory; +import org.springframework.context.Lifecycle; +import org.springframework.integration.channel.PollableChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.core.MessageChannel; +import org.springframework.integration.endpoint.AbstractEndpoint; +import org.springframework.jmx.export.MBeanExporter; +import org.springframework.jmx.export.assembler.AbstractConfigurableMBeanInfoAssembler; +import org.springframework.jmx.export.assembler.MBeanInfoAssembler; +import org.springframework.jmx.support.ObjectNameManager; +import org.springframework.util.Assert; +import org.springframework.util.ClassUtils; +import org.springframework.util.ObjectUtils; + +/** + * JMX-based Control Bus implementation. Exports all channel and endpoint + * beans from a given BeanFactory as MBeans. + * + * @author Mark Fisher + * @since 2.0 + */ +public class ControlBus implements BeanFactoryAware, InitializingBean { + + public static final String DEFAULT_DOMAIN = "org.springframework.integration"; + + + private final MBeanExporter exporter; + + private final String domain; + + private volatile ListableBeanFactory beanFactory; + + private final Set> managedTypes = new HashSet>( + Arrays.asList(new Class[] { MessageChannel.class, AbstractEndpoint.class })); + + + /** + * Create a {@link ControlBus} that will register channels and endpoints + * as MBeans with the given MBeanServer using the default domain name. + * @see #DEFAULT_DOMAIN + */ + public ControlBus(MBeanServer server) { + this(server, null); + } + + /** + * Create a {@link ControlBus} that will register channels and endpoints + * as MBeans with the given MBeanServer using the specified domain name. + */ + public ControlBus(MBeanServer server, String domain) { + Assert.notNull(server, "MBeanServer must not be null."); + this.domain = (domain != null) ? domain : DEFAULT_DOMAIN; + Assert.isTrue(!ObjectUtils.containsElement(server.getDomains(), this.domain), + "Domain [" + this.domain + "] is already in use within this MBeanServer."); + MBeanExporter exporter = new MBeanExporter(); + exporter.setServer(server); + exporter.setAutodetect(false); + exporter.setAssembler(new ControlBusMBeanInfoAssembler()); + this.exporter = exporter; + } + + + 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 { + this.exporter.afterPropertiesSet(); + for (Class type : this.managedTypes) { + Map beans = BeanFactoryUtils.beansOfTypeIncludingAncestors(this.beanFactory, type); + for (Map.Entry entry : beans.entrySet()) { + Object bean = entry.getValue(); + String beanName = entry.getKey(); + Class beanType = bean.getClass(); + try { + this.exporter.registerManagedResource(bean, this.generateObjectName(beanName, beanType)); + } + catch (MalformedObjectNameException e) { + throw new BeanInitializationException("Failed to generate JMX ObjectName.", e); + } + } + } + } + + private ObjectName generateObjectName(String beanName, Class beanType) throws MalformedObjectNameException { + StringBuilder sb = new StringBuilder(this.domain + ":name=" + beanName + ","); + if (MessageChannel.class.isAssignableFrom(beanType)) { + sb.append("type=channel"); + } + else if (AbstractEndpoint.class.isAssignableFrom(beanType)) { + sb.append("type=endpoint"); + } + else { + sb.append("type=" + ClassUtils.getShortNameAsProperty(beanType)); + } + return ObjectNameManager.getInstance(sb.toString()); + } + + + /** + * An {@link MBeanInfoAssembler} implementation for channels and endpoints. + */ + private static class ControlBusMBeanInfoAssembler extends AbstractConfigurableMBeanInfoAssembler { + + @Override + protected boolean includeOperation(Method method, String beanKey) { + Class declaringClass = method.getDeclaringClass(); + return this.shouldInclude(method, declaringClass); + } + + @Override + protected boolean includeReadAttribute(Method method, String beanKey) { + Class declaringClass = method.getDeclaringClass(); + return this.shouldInclude(method, declaringClass); + } + + @Override + protected boolean includeWriteAttribute(Method method, String beanKey) { + Class declaringClass = method.getDeclaringClass(); + return this.shouldInclude(method, declaringClass); + } + + private boolean shouldInclude(Method method, Class declaringClass) { + if (MessageChannel.class.isAssignableFrom(declaringClass) || + AbstractEndpoint.class.isAssignableFrom(declaringClass)) { + Class managementInterface = this.getManagementInterface(declaringClass); + if (managementInterface != null) { + for (Method interfaceMethod : managementInterface.getMethods()) { + if (interfaceMethod.getName().equals(method.getName()) && + Arrays.equals(interfaceMethod.getParameterTypes(), method.getParameterTypes())) { + return true; + } + } + } + } + return false; + } + + private Class getManagementInterface(Class type) { + if (AbstractEndpoint.class.isAssignableFrom(type)) { + return Lifecycle.class; + } + if (QueueChannel.class.isAssignableFrom(type)) { + return QueueChannelInfo.class; + } + if (PollableChannel.class.isAssignableFrom(type)) { + return PollableChannelInfo.class; + } + if (MessageChannel.class.isAssignableFrom(type)) { + return MessageChannelInfo.class; + } + return null; + } + } + +} diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/MessageChannelInfo.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/MessageChannelInfo.java new file mode 100644 index 0000000000..d2bc931d5b --- /dev/null +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/MessageChannelInfo.java @@ -0,0 +1,33 @@ +/* + * 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; + +/** + * This interface defines the MBean attributes and operations + * to be exposed for any MessageChannel instance. + * + * @author Mark Fisher + * @since 2.0 + */ +interface MessageChannelInfo { + + /** + * Returns the name of the channel. + */ + String getName(); + +} diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/PollableChannelInfo.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/PollableChannelInfo.java new file mode 100644 index 0000000000..fb6c6abddc --- /dev/null +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/PollableChannelInfo.java @@ -0,0 +1,33 @@ +/* + * 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; + +/** + * This interface defines the MBean attributes and operations + * to be exposed for any PollableChannel instance. + * + * @author Mark Fisher + * @since 2.0 + */ +interface PollableChannelInfo extends MessageChannelInfo { + + /** + * Clears all Messages currently held by the channel. + */ + void clear(); + +} diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/QueueChannelInfo.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/QueueChannelInfo.java new file mode 100644 index 0000000000..ea70117afb --- /dev/null +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/QueueChannelInfo.java @@ -0,0 +1,38 @@ +/* + * 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; + +/** + * This interface defines the MBean attributes and operations + * to be exposed for any QueueChannel instance. + * + * @author Mark Fisher + * @since 2.0 + */ +interface QueueChannelInfo extends PollableChannelInfo { + + /** + * Returns the number of Messages contained within the queue at the time of invocation. + */ + int getQueueSize(); + + /** + * Returns the remaining capacity of the queue at the time of invocation. + */ + int getRemainingCapacity(); + +} diff --git a/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java new file mode 100644 index 0000000000..1956395a47 --- /dev/null +++ b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusTests.java @@ -0,0 +1,137 @@ +/* + * 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 javax.management.MBeanServer; +import javax.management.MBeanServerFactory; +import javax.management.ObjectInstance; + +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.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.endpoint.PollingConsumer; +import org.springframework.integration.handler.BridgeHandler; +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 + */ +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)); + BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.test1"); + context.registerBeanDefinition("controlBus", controlBusDef); + context.refresh(); + MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class); + ObjectInstance instance = mbeanServer.getObjectInstance( + ObjectNameManager.getInstance("domain.test1:type=channel,name=directChannel")); + assertEquals(DirectChannel.class.getName(), instance.getClassName()); + } + + @Test + public void queueChannelRegistered() throws Exception { + context.registerBeanDefinition("queueChannel", new RootBeanDefinition(QueueChannel.class)); + BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.test2"); + context.registerBeanDefinition("controlBus", controlBusDef); + context.refresh(); + MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class); + ObjectInstance instance = mbeanServer.getObjectInstance( + ObjectNameManager.getInstance("domain.test2:type=channel,name=queueChannel")); + assertEquals(QueueChannel.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 BridgeHandler()); + context.registerBeanDefinition("eventDrivenConsumer", endpointDef); + BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.test3"); + context.registerBeanDefinition("controlBus", controlBusDef); + context.refresh(); + MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class); + ObjectInstance instance = mbeanServer.getObjectInstance( + ObjectNameManager.getInstance("domain.test3:type=endpoint,name=eventDrivenConsumer")); + assertEquals(EventDrivenConsumer.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 BridgeHandler()); + endpointDef.getPropertyValues().add("trigger", new PeriodicTrigger(10000)); + context.registerBeanDefinition("pollingConsumer", endpointDef); + context.registerBeanDefinition("taskScheduler", new RootBeanDefinition(ThreadPoolTaskScheduler.class)); + BeanDefinition controlBusDef = new RootBeanDefinition(ControlBus.class); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue(new RuntimeBeanReference("mbeanServer")); + controlBusDef.getConstructorArgumentValues().addGenericArgumentValue("domain.test4"); + context.registerBeanDefinition("controlBus", controlBusDef); + context.refresh(); + MBeanServer mbeanServer = context.getBean("mbeanServer", MBeanServer.class); + ObjectInstance instance = mbeanServer.getObjectInstance( + ObjectNameManager.getInstance("domain.test4:type=endpoint,name=pollingConsumer")); + assertEquals(PollingConsumer.class.getName(), instance.getClassName()); + } + +} diff --git a/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests-context.xml b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests-context.xml new file mode 100644 index 0000000000..cbdd352ee1 --- /dev/null +++ b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests-context.xml @@ -0,0 +1,32 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests.java b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests.java new file mode 100644 index 0000000000..b026f2de35 --- /dev/null +++ b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests.java @@ -0,0 +1,77 @@ +/* + * 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 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.channel.DirectChannel; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.endpoint.EventDrivenConsumer; +import org.springframework.integration.endpoint.PollingConsumer; +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 = "org.springframework.integration"; + + + @Autowired + private MBeanServer mbeanServer; + + + @Test + public void directChannelRegistered() throws Exception { + ObjectInstance instance = mbeanServer.getObjectInstance( + ObjectNameManager.getInstance(DOMAIN + ":type=channel,name=testDirectChannel")); + assertEquals(DirectChannel.class.getName(), instance.getClassName()); + } + + @Test + public void queueChannelRegistered() throws Exception { + ObjectInstance instance = mbeanServer.getObjectInstance( + ObjectNameManager.getInstance(DOMAIN + ":type=channel,name=testQueueChannel")); + assertEquals(QueueChannel.class.getName(), instance.getClassName()); + } + + @Test + public void eventDrivenConsumerRegistered() throws Exception { + ObjectInstance instance = mbeanServer.getObjectInstance( + ObjectNameManager.getInstance(DOMAIN + ":type=endpoint,name=testEventDrivenBridge")); + assertEquals(EventDrivenConsumer.class.getName(), instance.getClassName()); + } + + @Test + public void pollingConsumerRegistered() throws Exception { + ObjectInstance instance = mbeanServer.getObjectInstance( + ObjectNameManager.getInstance(DOMAIN + ":type=endpoint,name=testPollingBridge")); + assertEquals(PollingConsumer.class.getName(), instance.getClassName()); + } + +}