Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1013, INT-1014 Added namespace support for notification-publishing-channel-adapter and notification-listening-channel-adapter.

Browse Source
This commit is contained in:
Mark Fisher
2010-03-11 15:22:58 +00:00
parent 4a4a867f63
commit 11bf69173d
7 changed files with 145 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java
View File

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

2
org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingMessageHandler.java
View File

@@ -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;

2
org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/JmxNamespaceHandler.java
View File

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

53
org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/NotificationListeningChannelAdapterParser.java Normal file
View File

@@ -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");
}
}

40
org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/config/NotificationPublishingChannelAdapterParser.java Normal file
View File

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

30
org.springframework.integration.jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.0.xsd
View File

@@ -53,4 +53,34 @@
</xsd:complexType>
</xsd:element>
<xsd:element name="notification-listening-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines an inbound Channel Adapter that listens for JMX notifications.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string" use="optional"/>
<xsd:attribute name="channel" type="xsd:string" use="required"/>
<xsd:attribute name="mbean-server" type="xsd:string" default="mbeanServer"/>
<xsd:attribute name="object-name" type="xsd:string" use="required"/>
<xsd:attribute name="notification-filter" type="xsd:string" use="optional"/>
<xsd:attribute name="handback" type="xsd:string" use="optional"/>
<xsd:attribute name="send-timeout" type="xsd:string" use="optional"/>
</xsd:complexType>
</xsd:element>
<xsd:element name="notification-publishing-channel-adapter">
<xsd:annotation>
<xsd:documentation>
Defines an outbound Channel Adapter that publishes JMX notifications.
</xsd:documentation>
</xsd:annotation>
<xsd:complexType>
<xsd:attribute name="id" type="xsd:string" use="optional"/>
<xsd:attribute name="channel" type="xsd:string" use="required"/>
<xsd:attribute name="object-name" type="xsd:string" use="required"/>
</xsd:complexType>
</xsd:element>
</xsd:schema>

6
org.springframework.integration.jmx/src/test/java/org/springframework/integration/jmx/NotificationListeningMessageProducerTests.java
View File

@@ -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) {