INT-1003, INT-1004 notification publishing/listening adapters

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<Notification> {
 
 	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(),
+		Assert.hasText(type,
-				message.getPayload().toString());
+				"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<Notification> {
 		return (type != null) ? type : this.defaultNotificationType;
 	}
 
-	private static ObjectName stringToObjectName(String objectName) {
-		try {
-			return ObjectNameManager.getInstance(objectName);
-		}
-		catch (MalformedObjectNameException e) {
-			throw new IllegalArgumentException(e);
-		}
-	}
-
 }

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

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
+public class NotificationPublishingAdapter extends AbstractMessageHandler implements BeanFactoryAware, InitializingBean {
-		implements BeanNameAware, BeanFactoryAware, InitializingBean {
 
 	private final PublisherDelegate delegate = new PublisherDelegate();
 
 	private volatile OutboundMessageMapper<Notification> notificationMapper;
 
-	private volatile String objectName;
+	private final ObjectName objectName;
 
-	private volatile String beanName;
+	private volatile String defaultNotificationType;
 
 	private volatile ListableBeanFactory beanFactory;
 
 
-	public NotificationPublishingAdapter() {
+	public NotificationPublishingAdapter(ObjectName objectName) {
-		this(null);
+		Assert.notNull(objectName, "JMX ObjectName is required");
-	}
-
-	public NotificationPublishingAdapter(String objectName) {
 		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<Notification> 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 <emphasis>no</emphasis> 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;

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(
+		this.publisherObjectName = ObjectNameManager.getInstance("test:type=publisher");
-				"org.springframework.integration:type=notificationPublishingAdapter,name=testPublisher");
 		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());
 	}