INT-627 added operation channel for invoking JMX operations on the beans exported by the Control Bus

org.springframework.integration.jmx/src/main/java/org/springframework/integration/control/ControlBus.java

@@ -18,6 +18,7 @@ package org.springframework.integration.control;
 
 import java.lang.reflect.Method;
 import java.util.Arrays;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
@@ -34,10 +35,16 @@ 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.DirectChannel;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.channel.QueueChannel;
+import org.springframework.integration.core.Message;
 import org.springframework.integration.core.MessageChannel;
+import org.springframework.integration.core.MessageHeaders;
 import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.jmx.JmxHeaders;
+import org.springframework.integration.jmx.OperationInvokingMessageHandler;
+import org.springframework.integration.message.MessageBuilder;
 import org.springframework.jmx.export.MBeanExporter;
 import org.springframework.jmx.export.assembler.AbstractConfigurableMBeanInfoAssembler;
 import org.springframework.jmx.export.assembler.MBeanInfoAssembler;
@@ -57,11 +64,17 @@ public class ControlBus implements BeanFactoryAware, InitializingBean {
 
 	public static final String DEFAULT_DOMAIN = "org.springframework.integration";
 
+	public static final String TARGET_BEAN_NAME = JmxHeaders.PREFIX + "_controlBus_targetBeanName";
+
 
 	private final MBeanExporter exporter;
 
 	private final String domain;
 
+	private final Map<String, ObjectName> exportedBeanObjectNameMap = new HashMap<String, ObjectName>();
+
+	private final DirectChannel operationChannel = new DirectChannel();
+
 	private volatile ListableBeanFactory beanFactory;
 
 	private final Set<Class<?>> managedTypes = new HashSet<Class<?>>(
@@ -94,6 +107,17 @@ public class ControlBus implements BeanFactoryAware, InitializingBean {
 	}
 
 
+	/**
+	 * 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 MessageChannel getOperationChannel() {
+		return this.operationChannel;
+	}
+
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		Assert.isTrue(beanFactory instanceof ListableBeanFactory,
 				"A ListableBeanFactory is required.");
@@ -109,13 +133,20 @@ public class ControlBus implements BeanFactoryAware, InitializingBean {
 				String beanName = entry.getKey();
 				Class<?> beanType = bean.getClass();
 				try {
-					this.exporter.registerManagedResource(bean, this.generateObjectName(beanName, beanType));
+					ObjectName objectName = this.generateObjectName(beanName, beanType);
+					this.exporter.registerManagedResource(bean, objectName);
+					this.exportedBeanObjectNameMap.put(beanName, objectName);
 				}
 				catch (MalformedObjectNameException e) {
 					throw new BeanInitializationException("Failed to generate JMX ObjectName.", e);
 				}
 			}
 		}
+		OperationInvokingMessageHandler handler = new ControlBusOperationInvokingMessageHandler();
+		handler.setBeanFactory(this.beanFactory);
+		handler.setServer(this.exporter.getServer());
+		handler.afterPropertiesSet();
+		this.operationChannel.subscribe(handler);
 	}
 
 	private ObjectName generateObjectName(String beanName, Class<?> beanType) throws MalformedObjectNameException {
@@ -189,4 +220,22 @@ public class ControlBus implements BeanFactoryAware, InitializingBean {
 		}
 	}
 
+
+	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.");
+			ObjectName objectName = exportedBeanObjectNameMap.get(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);
+		}
+	}
+
 }

org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusOperationChannelTests.java

@@ -0,0 +1,83 @@
+/*
+ * 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 javax.management.MBeanServer;
+
+import org.junit.Test;
+
+import org.springframework.beans.factory.support.RootBeanDefinition;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.integration.channel.DirectChannel;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.endpoint.EventDrivenConsumer;
+import org.springframework.integration.jmx.JmxHeaders;
+import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.MessageHandler;
+import org.springframework.jmx.support.JmxUtils;
+
+/**
+ * @author Mark Fisher
+ * @since 2.0
+ */
+public class ControlBusOperationChannelTests {
+
+	private final MBeanServer server = JmxUtils.locateMBeanServer();
+
+	private final String domain = "domain.test";
+
+
+	@Test
+	public void replyProducingOperation() throws Exception {
+		GenericApplicationContext context = new GenericApplicationContext();
+		RootBeanDefinition endpointDef = new RootBeanDefinition(EventDrivenConsumer.class);
+		endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new DirectChannel());
+		endpointDef.getConstructorArgumentValues().addGenericArgumentValue(new TestHandler());
+		context.registerBeanDefinition("testEndpoint", endpointDef);
+		RootBeanDefinition busDef = new RootBeanDefinition(ControlBus.class);
+		busDef.getConstructorArgumentValues().addGenericArgumentValue(server);
+		busDef.getConstructorArgumentValues().addGenericArgumentValue(domain);
+		context.registerBeanDefinition("controlBus", busDef);
+		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());
+		context.close();
+	}
+
+
+	private static class TestHandler implements MessageHandler {
+		public void handleMessage(Message<?> message) {
+		}
+	}
+
+}