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 index 45ad9075e0..6bdfe16ecc 100644 --- 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 @@ -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 exportedBeanObjectNameMap = new HashMap(); + + private final DirectChannel operationChannel = new DirectChannel(); + private volatile ListableBeanFactory beanFactory; private final Set> managedTypes = new HashSet>( @@ -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); + } + } + } diff --git a/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusOperationChannelTests.java b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/control/ControlBusOperationChannelTests.java new file mode 100644 index 0000000000..13e21d58ad --- /dev/null +++ b/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) { + } + } + +}