diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java index a4955df47a..e86adea2e2 100644 --- a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java @@ -16,10 +16,6 @@ package org.springframework.integration.jmx; -import java.util.Arrays; -import java.util.LinkedHashSet; -import java.util.Set; - import javax.management.InstanceNotFoundException; import javax.management.ListenerNotFoundException; import javax.management.MBeanServer; @@ -50,7 +46,7 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport private volatile MBeanServer server; - private volatile Set objectNames; + private volatile ObjectName objectName; private volatile NotificationFilter filter; @@ -66,11 +62,11 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport } /** - * Specify one or more JMX ObjectNames of notification publishers + * Specify the JMX ObjectName of the notification publisher * to which this notification listener should be subscribed. */ - public void setObjectNames(ObjectName... objectNames) { - this.objectNames = new LinkedHashSet(Arrays.asList(objectNames)); + public void setObjectName(ObjectName objectName) { + this.objectName = objectName; } /** @@ -120,10 +116,8 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport protected void doStart() { try { Assert.notNull(this.server, "MBeanServer is required."); - Assert.notEmpty(this.objectNames, "One or more ObjectNames are required."); - for (ObjectName objectName : this.objectNames) { - this.server.addNotificationListener(objectName, this, this.filter, this.handback); - } + Assert.notNull(this.objectName, "An ObjectName is required."); + this.server.addNotificationListener(this.objectName, this, this.filter, this.handback); } catch (InstanceNotFoundException e) { throw new IllegalStateException("Failed to find MBean instance.", e); @@ -135,18 +129,16 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport */ @Override protected void doStop() { - try { - Assert.notNull(this.server, "MBeanServer is required."); - Assert.notEmpty(this.objectNames, "One or more ObjectNames are required."); - for (ObjectName objectName : this.objectNames) { - this.server.removeNotificationListener(objectName, this, this.filter, this.handback); + if (this.server != null && this.objectName != null) { + try { + this.server.removeNotificationListener(this.objectName, this, this.filter, this.handback); + } + catch (InstanceNotFoundException e) { + throw new IllegalStateException("Failed to find MBean instance.", e); + } + catch (ListenerNotFoundException e) { + throw new IllegalStateException("Failed to find NotificationListener.", e); } - } - catch (InstanceNotFoundException e) { - throw new IllegalStateException("Failed to find MBean instance.", e); - } - catch (ListenerNotFoundException e) { - throw new IllegalStateException("Failed to find NotificationListener.", e); } } diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingMessageHandler.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingMessageHandler.java index 04924f2698..10194c29b7 100644 --- a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingMessageHandler.java +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingMessageHandler.java @@ -32,6 +32,7 @@ import org.springframework.integration.core.Message; import org.springframework.integration.handler.AbstractMessageHandler; import org.springframework.integration.message.OutboundMessageMapper; import org.springframework.jmx.export.MBeanExporter; +import org.springframework.jmx.export.annotation.ManagedResource; import org.springframework.jmx.export.notification.NotificationPublisher; import org.springframework.jmx.export.notification.NotificationPublisherAware; import org.springframework.jmx.support.ObjectNameManager; @@ -120,6 +121,7 @@ public class NotificationPublishingMessageHandler extends AbstractMessageHandler /** * Simple class used for the actual MBean instances to be registered. */ + @ManagedResource private static class PublisherDelegate implements NotificationPublisherAware { private volatile NotificationPublisher notificationPublisher; diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/JmxNamespaceHandler.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/JmxNamespaceHandler.java index c4be0fe8c3..557716f414 100644 --- a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/JmxNamespaceHandler.java +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/JmxNamespaceHandler.java @@ -29,6 +29,8 @@ public class JmxNamespaceHandler extends AbstractIntegrationNamespaceHandler { public void init() { this.registerBeanDefinitionParser("operation-invoking-channel-adapter", new OperationInvokingChannelAdapterParser()); this.registerBeanDefinitionParser("attribute-polling-channel-adapter", new AttributePollingChannelAdapterParser()); + this.registerBeanDefinitionParser("notification-listening-channel-adapter", new NotificationListeningChannelAdapterParser()); + this.registerBeanDefinitionParser("notification-publishing-channel-adapter", new NotificationPublishingChannelAdapterParser()); } } diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/NotificationListeningChannelAdapterParser.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/NotificationListeningChannelAdapterParser.java new file mode 100644 index 0000000000..1ad7eb7859 --- /dev/null +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/NotificationListeningChannelAdapterParser.java @@ -0,0 +1,53 @@ +/* + * 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.w3c.dom.Element; + +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.AbstractSimpleBeanDefinitionParser; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.IntegrationNamespaceUtils; +import org.springframework.util.StringUtils; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class NotificationListeningChannelAdapterParser extends AbstractSimpleBeanDefinitionParser { + + @Override + protected String getBeanClassName(Element element) { + return "org.springframework.integration.jmx.NotificationListeningMessageProducer"; + } + + @Override + protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) { + Object source = parserContext.extractSource(element); + String channel = element.getAttribute("channel"); + if (!StringUtils.hasText(channel)) { + parserContext.getReaderContext().error("The 'channel' attribute is required.", source); + } + builder.addPropertyReference("outputChannel", channel); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "mbean-server", "server"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "notification-filter", "filter"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "handback"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "send-timeout"); + IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "object-name"); + } + +} diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/NotificationPublishingChannelAdapterParser.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/NotificationPublishingChannelAdapterParser.java new file mode 100644 index 0000000000..a14bf71614 --- /dev/null +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/NotificationPublishingChannelAdapterParser.java @@ -0,0 +1,40 @@ +/* + * 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.w3c.dom.Element; + +import org.springframework.beans.factory.support.AbstractBeanDefinition; +import org.springframework.beans.factory.support.BeanDefinitionBuilder; +import org.springframework.beans.factory.xml.ParserContext; +import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class NotificationPublishingChannelAdapterParser extends AbstractOutboundChannelAdapterParser { + + @Override + protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) { + BeanDefinitionBuilder builder = BeanDefinitionBuilder.rootBeanDefinition( + "org.springframework.integration.jmx.NotificationPublishingMessageHandler"); + builder.addConstructorArgValue(element.getAttribute("object-name")); + return builder.getBeanDefinition(); + } + +} diff --git a/org.springframework.integration.jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.0.xsd b/org.springframework.integration.jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.0.xsd index 972d665ffe..17e867ef00 100644 --- a/org.springframework.integration.jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.0.xsd +++ b/org.springframework.integration.jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.0.xsd @@ -53,4 +53,34 @@ + + + + Defines an inbound Channel Adapter that listens for JMX notifications. + + + + + + + + + + + + + + + + + Defines an outbound Channel Adapter that publishes JMX notifications. + + + + + + + + + \ No newline at end of file diff --git a/org.springframework.integration.jmx/src/test/java/org/springframework/integration/jmx/NotificationListeningMessageProducerTests.java b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/jmx/NotificationListeningMessageProducerTests.java index 95c5fe1d0c..a6a01b6c7f 100644 --- a/org.springframework.integration.jmx/src/test/java/org/springframework/integration/jmx/NotificationListeningMessageProducerTests.java +++ b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/jmx/NotificationListeningMessageProducerTests.java @@ -76,7 +76,7 @@ public class NotificationListeningMessageProducerTests { QueueChannel outputChannel = new QueueChannel(); NotificationListeningMessageProducer adapter = new NotificationListeningMessageProducer(); adapter.setServer(this.server); - adapter.setObjectNames(this.objectName); + adapter.setObjectName(this.objectName); adapter.setOutputChannel(outputChannel); adapter.afterPropertiesSet(); adapter.start(); @@ -95,7 +95,7 @@ public class NotificationListeningMessageProducerTests { QueueChannel outputChannel = new QueueChannel(); NotificationListeningMessageProducer adapter = new NotificationListeningMessageProducer(); adapter.setServer(this.server); - adapter.setObjectNames(this.objectName); + adapter.setObjectName(this.objectName); adapter.setOutputChannel(outputChannel); Integer handback = new Integer(123); adapter.setHandback(handback); @@ -117,7 +117,7 @@ public class NotificationListeningMessageProducerTests { QueueChannel outputChannel = new QueueChannel(); NotificationListeningMessageProducer adapter = new NotificationListeningMessageProducer(); adapter.setServer(this.server); - adapter.setObjectNames(this.objectName); + adapter.setObjectName(this.objectName); adapter.setOutputChannel(outputChannel); adapter.setFilter(new NotificationFilter() { public boolean isNotificationEnabled(Notification notification) {