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).

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<Class<?>> managedTypes = new HashSet<Class<?>>(
+			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<String, ?> beans = BeanFactoryUtils.beansOfTypeIncludingAncestors(this.beanFactory, type);
+			for (Map.Entry<String, ?> 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;
+		}
+	}
+
+}

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();
+
+}

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();
+
+}

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();
+
+}

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());
+	}
+
+}

org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusXmlTests-context.xml

@@ -0,0 +1,32 @@
+<?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="testDirectChannel"/>
+
+	<int:channel id="testQueueChannel">
+		<int:queue capacity="99"/>
+	</int:channel>
+
+	<int:bridge id="testEventDrivenBridge" input-channel="testDirectChannel"/>
+
+	<int:bridge id="testPollingBridge" input-channel="testQueueChannel" output-channel="nullChannel">
+		<int:poller max-messages-per-poll="1">
+			<int:interval-trigger interval="10000"/>
+		</int:poller>
+	</int:bridge>
+
+	<bean id="controlBus" class="org.springframework.integration.control.ControlBus">
+		<constructor-arg ref="mbeanServer"/>
+	</bean>
+
+	<bean id="mbeanServer" class="org.springframework.jmx.support.MBeanServerFactoryBean">
+		<property name="locateExistingServerIfPossible" value="true"/>
+	</bean>
+
+</beans>

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());
+	}
+
+}