From 88e6db6355b0375bbefa896cd1e07124f5e2adcf Mon Sep 17 00:00:00 2001 From: Mark Fisher Date: Mon, 8 Mar 2010 19:54:17 +0000 Subject: [PATCH] INT-1003, INT-1004 notification publishing/listening adapters --- .../jmx/DefaultNotificationMapper.java | 46 ++++++++-------- .../jmx/NotificationListeningAdapter.java | 33 ++++++++++++ .../jmx/NotificationPublishingAdapter.java | 54 ++++++++++++------- .../NotificationPublishingAdapterTests.java | 13 +++-- 4 files changed, 98 insertions(+), 48 deletions(-) diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/DefaultNotificationMapper.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/DefaultNotificationMapper.java index d8adec3f8d..bf67c76397 100644 --- a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/DefaultNotificationMapper.java +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/DefaultNotificationMapper.java @@ -16,17 +16,21 @@ package org.springframework.integration.jmx; -import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; -import javax.management.MalformedObjectNameException; import javax.management.Notification; import javax.management.ObjectName; import org.springframework.integration.core.Message; import org.springframework.integration.message.OutboundMessageMapper; -import org.springframework.jmx.support.ObjectNameManager; +import org.springframework.util.Assert; /** + * Default Messaging Mapper implementation for the {@link NotificationPublishingAdapter}. + * If the Message has a String-typed payload, that will be passed as the 'message' of + * the Notification instance. Otherwise, the payload object will be passed as the + * 'userData' of the Notification instance. + * * @author Mark Fisher * @since 2.0 */ @@ -34,28 +38,29 @@ class DefaultNotificationMapper implements OutboundMessageMapper { private final ObjectName sourceObjectName; - private volatile String defaultNotificationType = "org.springframework.integration.jmx.event"; + private final String defaultNotificationType; - private final AtomicInteger sequence = new AtomicInteger(); + private final AtomicLong sequence = new AtomicLong(); - DefaultNotificationMapper(ObjectName sourceObjectName) { + DefaultNotificationMapper(ObjectName sourceObjectName, String defaultNotificationType) { this.sourceObjectName = sourceObjectName; - } - - DefaultNotificationMapper(String sourceObjectName) { - this(stringToObjectName(sourceObjectName)); - } - - - public void setDefaultNotificationType(String defaultNotificationType) { this.defaultNotificationType = defaultNotificationType; } + public Notification fromMessage(Message message) throws Exception { String type = this.resolveNotificationType(message); - return new Notification(type, this.sourceObjectName, this.sequence.incrementAndGet(), - message.getPayload().toString()); + Assert.hasText(type, + "No notification type header is available, and no default has been provided."); + Object payload = (message != null) ? message.getPayload() : null; + String notificationMessage = (payload instanceof String) ? (String) payload : null; + Notification notification = new Notification(type, this.sourceObjectName, + this.sequence.incrementAndGet(), System.currentTimeMillis(), notificationMessage); + if (payload != null && !(payload instanceof String)) { + notification.setUserData(payload); + } + return notification; } private String resolveNotificationType(Message message) { @@ -63,13 +68,4 @@ class DefaultNotificationMapper implements OutboundMessageMapper { return (type != null) ? type : this.defaultNotificationType; } - private static ObjectName stringToObjectName(String objectName) { - try { - return ObjectNameManager.getInstance(objectName); - } - catch (MalformedObjectNameException e) { - throw new IllegalArgumentException(e); - } - } - } diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningAdapter.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningAdapter.java index bf7a13183c..8efef3c5a9 100644 --- a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningAdapter.java +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningAdapter.java @@ -38,6 +38,9 @@ import org.springframework.integration.support.ComponentMetadata; import org.springframework.util.Assert; /** + * A JMX {@link NotificationListener} implementation that will send Messages + * containing the JMX {@link Notification} instances as their payloads. + * * @author Mark Fisher * @since 2.0 */ @@ -54,22 +57,44 @@ public class NotificationListeningAdapter extends MessageProducerSupport impleme private volatile Object handback; + /** + * Provide a reference to the MBeanServer where the notification + * publishing MBeans are registered. + */ public void setServer(MBeanServer server) { this.server = server; } + /** + * Specify one or more JMX ObjectNames of notification publishers + * to which this notification listener should be subscribed. + */ public void setObjectNames(ObjectName... objectNames) { this.objectNames = new LinkedHashSet(Arrays.asList(objectNames)); } + /** + * Specify a {@link NotificationFilter} to be passed to the server + * when registering this listener. The filter may be null. + */ public void setFilter(NotificationFilter filter) { this.filter = filter; } + /** + * Specify a handback object to provide context to the listener + * upon notification. This object may be null. + */ public void setHandback(Object handback) { this.handback = handback; } + /** + * Notification handling method implementation. Creates a Message with the + * JMX {@link Notification} as its payload, and if the handback object is + * not null, it sets that as a Message header value. The Message is then + * sent to this producer's output channel. + */ public void handleNotification(Notification notification, Object handback) { if (logger.isInfoEnabled()) { logger.info("received notification: " + notification + ", and handback: " + handback); @@ -88,10 +113,14 @@ public class NotificationListeningAdapter extends MessageProducerSupport impleme metadata.setAttribute("transport", "jmx"); } + /** + * Registers the notification listener with the specified ObjectNames. + */ @Override 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); } @@ -101,10 +130,14 @@ public class NotificationListeningAdapter extends MessageProducerSupport impleme } } + /** + * Unregisters the notification listener. + */ @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); } diff --git a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingAdapter.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingAdapter.java index c22a00a0ca..3dcffaa575 100644 --- a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingAdapter.java +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingAdapter.java @@ -18,13 +18,14 @@ package org.springframework.integration.jmx; import java.util.Map; +import javax.management.MalformedObjectNameException; import javax.management.Notification; +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.BeanNameAware; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.ListableBeanFactory; import org.springframework.integration.core.Message; @@ -40,35 +41,53 @@ import org.springframework.util.Assert; * @author Mark Fisher * @since 2.0 */ -public class NotificationPublishingAdapter extends AbstractMessageHandler - implements BeanNameAware, BeanFactoryAware, InitializingBean { +public class NotificationPublishingAdapter extends AbstractMessageHandler implements BeanFactoryAware, InitializingBean { private final PublisherDelegate delegate = new PublisherDelegate(); private volatile OutboundMessageMapper notificationMapper; - private volatile String objectName; + private final ObjectName objectName; - private volatile String beanName; + private volatile String defaultNotificationType; private volatile ListableBeanFactory beanFactory; - public NotificationPublishingAdapter() { - this(null); - } - - public NotificationPublishingAdapter(String objectName) { + public NotificationPublishingAdapter(ObjectName objectName) { + Assert.notNull(objectName, "JMX ObjectName is required"); this.objectName = objectName; } + public NotificationPublishingAdapter(String objectName) { + Assert.notNull(objectName, "JMX ObjectName is required"); + try { + this.objectName = ObjectNameManager.getInstance(objectName); + } + catch (MalformedObjectNameException e) { + throw new IllegalArgumentException(e); + } + } + + /** + * Set a mapper for creating Notifications from a Message. If not provided, + * a default implementation will be used such that String-typed payloads will be + * passed as the 'message' of the Notification and all other payload types + * will be passed as the 'userData' of the Notification. + */ public void setNotificationMapper(OutboundMessageMapper notificationMapper) { this.notificationMapper = notificationMapper; } - public void setBeanName(String beanName) { - this.beanName = beanName; + /** + * Specify a dot-delimited String representing the Notification type to + * use by default when no explicit Notification mapper + * has been configured. If not provided, then a notification type header will + * be required for each message being mapped into a Notification. + */ + public void setDefaultNotificationType(String defaultNotificationType) { + this.defaultNotificationType = defaultNotificationType; } public void setBeanFactory(BeanFactory beanFactory) throws BeansException { @@ -83,14 +102,10 @@ public class NotificationPublishingAdapter extends AbstractMessageHandler "No unique MBeanExporter is available in the current context (found " + exporters.size() + ")."); MBeanExporter exporter = exporters.values().iterator().next(); - if (this.objectName == null) { - this.objectName = "org.springframework.integration:" + - "type=notificationPublishingAdapter,name=" + this.beanName; - } if (this.notificationMapper == null) { - this.notificationMapper = new DefaultNotificationMapper(this.objectName); + this.notificationMapper = new DefaultNotificationMapper(this.objectName, this.defaultNotificationType); } - exporter.registerManagedResource(this.delegate, ObjectNameManager.getInstance(this.objectName)); + exporter.registerManagedResource(this.delegate, this.objectName); if (this.logger.isInfoEnabled()) { this.logger.info("Registered JMX notification publisher as MBean with ObjectName: " + this.objectName); } @@ -102,6 +117,9 @@ public class NotificationPublishingAdapter extends AbstractMessageHandler } + /** + * Simple class used for the actual MBean instances to be registered. + */ private static class PublisherDelegate implements NotificationPublisherAware { private volatile NotificationPublisher notificationPublisher; diff --git a/org.springframework.integration.jmx/src/test/java/org/springframework/integration/jmx/NotificationPublishingAdapterTests.java b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/jmx/NotificationPublishingAdapterTests.java index 9125dffee4..239e272e74 100644 --- a/org.springframework.integration.jmx/src/test/java/org/springframework/integration/jmx/NotificationPublishingAdapterTests.java +++ b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/jmx/NotificationPublishingAdapterTests.java @@ -29,6 +29,7 @@ import org.junit.After; import org.junit.Before; import org.junit.Test; +import org.springframework.beans.factory.support.RootBeanDefinition; import org.springframework.context.support.StaticApplicationContext; import org.springframework.integration.message.StringMessage; import org.springframework.jmx.export.MBeanExporter; @@ -49,13 +50,15 @@ public class NotificationPublishingAdapterTests { @Before public void setup() throws Exception { - this.publisherObjectName = ObjectNameManager.getInstance( - "org.springframework.integration:type=notificationPublishingAdapter,name=testPublisher"); + this.publisherObjectName = ObjectNameManager.getInstance("test:type=publisher"); context.registerSingleton("exporter", MBeanExporter.class); - context.registerSingleton("testPublisher", NotificationPublishingAdapter.class); + RootBeanDefinition publisherDefinition = new RootBeanDefinition(NotificationPublishingAdapter.class); + publisherDefinition.getConstructorArgumentValues().addGenericArgumentValue(this.publisherObjectName); + publisherDefinition.getPropertyValues().add("defaultNotificationType", "test.type"); + context.registerBeanDefinition("testPublisher", publisherDefinition); context.refresh(); MBeanExporter exporter = context.getBean(MBeanExporter.class); - exporter.getServer().addNotificationListener(this.publisherObjectName, this.listener, null, null); + exporter.getServer().addNotificationListener(publisherObjectName, this.listener, null, null); } @After @@ -74,7 +77,7 @@ public class NotificationPublishingAdapterTests { Notification notification = this.listener.notifications.get(0); assertEquals(this.publisherObjectName, notification.getSource()); assertEquals("foo", notification.getMessage()); - assertEquals("org.springframework.integration.jmx.event", notification.getType()); + assertEquals("test.type", notification.getType()); }