Merge pull request #607 from garyrussell/INT-2487

* INT-2487:
  INT-2487 Object Name Patterns for Notifications

spring-integration-jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.java

@@ -17,6 +17,11 @@
 package org.springframework.integration.jmx;
 
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
 
 import javax.management.InstanceNotFoundException;
 import javax.management.ListenerNotFoundException;
@@ -24,6 +29,7 @@ import javax.management.MBeanServerConnection;
 import javax.management.Notification;
 import javax.management.NotificationFilter;
 import javax.management.NotificationListener;
+import javax.management.ObjectInstance;
 import javax.management.ObjectName;
 
 import org.apache.commons.logging.Log;
@@ -32,6 +38,7 @@ import org.springframework.integration.Message;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
 
 /**
  * A JMX {@link NotificationListener} implementation that will send Messages
@@ -47,7 +54,7 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
 
 	private volatile MBeanServerConnection server;
 
-	private volatile ObjectName objectName;
+	private volatile ObjectName[] objectNames;
 
 	private volatile NotificationFilter filter;
 
@@ -63,11 +70,13 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
 	}
 
 	/**
-	 * Specify the JMX ObjectName of the notification publisher
+	 * Specify the JMX ObjectNames (or patterns)
- 	 * to which this notification listener should be subscribed.
+	 * of the notification publisher
+	 * to which this notification listener should be subscribed.
 	 */
-	public void setObjectName(ObjectName objectName) {
+	public void setObjectName(ObjectName... objectNames) {
-		this.objectName = objectName;
+		Assert.isTrue(!ObjectUtils.isEmpty(objectNames), "'objectNames' must contain at least one ObjectName");
+		this.objectNames = objectNames;
 	}
 
 	/**
@@ -114,10 +123,18 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
 	 */
 	@Override
 	protected void doStart() {
+		logger.debug("Registering to receive notifications");
 		try {
 			Assert.notNull(this.server, "MBeanServer is required.");
-			Assert.notNull(this.objectName, "An ObjectName is required.");
+			Assert.notNull(this.objectNames, "An ObjectName is required.");
-			this.server.addNotificationListener(this.objectName, this, this.filter, this.handback);
+			Collection<ObjectName> objectNames = this.retrieveMBeanNames();
+			if (objectNames.size() < 1) {
+				logger.error("No MBeans found matching ObjectName pattern(s): " +
+							Arrays.asList(this.objectNames));
+			}
+			for (ObjectName objectName : objectNames) {
+				this.server.addNotificationListener(objectName, this, this.filter, this.handback);
+			}
 		}
 		catch (InstanceNotFoundException e) {
 			throw new IllegalStateException("Failed to find MBean instance.", e);
@@ -132,20 +149,46 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
 	 */
 	@Override
 	protected void doStop() {
-		if (this.server != null && this.objectName != null) {
+		logger.debug("Unregistering notifications");
-			try {
+		if (this.server != null && this.objectNames != null) {
-				this.server.removeNotificationListener(this.objectName, this, this.filter, this.handback);
+			Collection<ObjectName> objectNames = this.retrieveMBeanNames();
-			}
+			for (ObjectName objectName : objectNames) {
-			catch (InstanceNotFoundException e) {
+				try {
-				throw new IllegalStateException("Failed to find MBean instance.", e);
+					this.server.removeNotificationListener(objectName, this, this.filter, this.handback);
-			}
+				}
-			catch (ListenerNotFoundException e) {
+				catch (InstanceNotFoundException e) {
-				throw new IllegalStateException("Failed to find NotificationListener.", e);
+					logger.error("Failed to find MBean instance.", e);
-			}
+				}
-			catch (IOException e) {
+				catch (ListenerNotFoundException e) {
-				throw new IllegalStateException("IOException on MBeanServerConnection.", e);
+					logger.error("Failed to find NotificationListener.", e);
+				}
+				catch (IOException e) {
+					logger.error("IOException on MBeanServerConnection.", e);
+				}
 			}
 		}
 	}
 
+	protected Collection<ObjectName> retrieveMBeanNames() {
+		List<ObjectName> objectNames = new ArrayList<ObjectName>();
+		for (ObjectName pattern : this.objectNames) {
+			Set<ObjectInstance> mBeanInfos;
+			try {
+				mBeanInfos = this.server.queryMBeans(pattern, null);
+			}
+			catch (IOException e) {
+				throw new IllegalStateException("IOException on MBeanServerConnection.", e);
+			}
+			if (mBeanInfos.size() == 0 && logger.isDebugEnabled()) {
+				logger.debug("No MBeans found matching pattern:" + pattern);
+			}
+			for (ObjectInstance instance : mBeanInfos) {
+				if (logger.isDebugEnabled()) {
+					logger.debug("Found MBean:" + instance.getObjectName().toString());
+				}
+				objectNames.add(instance.getObjectName());
+			}
+		}
+		return objectNames;
+	}
 }

spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/NotificationListeningChannelAdapterParser.java

@@ -30,7 +30,7 @@ import org.w3c.dom.Element;
  * @since 2.0
  */
 public class NotificationListeningChannelAdapterParser extends AbstractChannelAdapterParser {
-	
+
 	@Override
 	protected boolean shouldGenerateIdAsFallback() {
 		return true;

spring-integration-jmx/src/main/resources/org/springframework/integration/jmx/config/spring-integration-jmx-2.2.xsd

@@ -72,9 +72,12 @@
 
 	<xsd:element name="notification-listening-channel-adapter">
 		<xsd:annotation>
-			<xsd:documentation>
+			<xsd:documentation><![CDATA[
-				Defines an inbound Channel Adapter that listens for JMX notifications.
+				Defines an inbound Channel Adapter that listens for JMX notifications. The 'object-name'
-			</xsd:documentation>
+				attribute on this endpoint can contain an ObjectName pattern and the MBeanServer will
+				be queried for MBeans with ObjectNames matching the pattern. In addition, it can contain
+				a SpEL expression that references a <util:list/> of ObjectNames or ObjectName patterns.
+			]]></xsd:documentation>
 		</xsd:annotation>
 		<xsd:complexType>
 			<xsd:complexContent>

spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/NotificationListeningChannelAdapterParserTests-context.xml

@@ -4,14 +4,12 @@
 	xmlns:context="http://www.springframework.org/schema/context"
 	xmlns:si="http://www.springframework.org/schema/integration"
 	xmlns:jmx="http://www.springframework.org/schema/integration/jmx"
-	xsi:schemaLocation="http://www.springframework.org/schema/beans
+	xmlns:util="http://www.springframework.org/schema/util"
-			http://www.springframework.org/schema/beans/spring-beans.xsd
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
-			http://www.springframework.org/schema/context
+		http://www.springframework.org/schema/integration/jmx http://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd
-			http://www.springframework.org/schema/context/spring-context.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
-			http://www.springframework.org/schema/integration
+		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
-			http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
-			http://www.springframework.org/schema/integration/jmx
-			http://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd">
 
 	<context:mbean-export/>
 	<context:mbean-server/>
@@ -26,9 +24,32 @@
 
 	<bean id="testPublisher" class="org.springframework.integration.jmx.config.TestPublisher"/>
 
+	<bean id="testPublisher2" class="org.springframework.integration.jmx.config.TestPublisher"/>
+
 	<jmx:notification-listening-channel-adapter id="autoChannel"
 			object-name="org.springframework.integration.jmx.config:type=TestPublisher,name=testPublisher"/>
 
 	<si:bridge input-channel="autoChannel" output-channel="nullChannel" />
 
+	<si:channel id="patternChannel">
+		<si:queue/>
+	</si:channel>
+
+	<jmx:notification-listening-channel-adapter id="patternAdapter"
+			channel="patternChannel"
+			object-name="org.springframework.integration.jmx.config:type=*,name=testPublisher"/>
+
+	<si:channel id="multiChannel">
+		<si:queue/>
+	</si:channel>
+
+	<jmx:notification-listening-channel-adapter id="multiAdapter"
+			channel="multiChannel"
+			object-name="#{patterns}"/>
+
+	<util:list id="patterns">
+		<value>org.springframework.integration.jmx.config:type=*,name=testPublisher</value>
+		<value>org.springframework.integration.jmx.config:type=*,name=testPub*</value>
+	</util:list>
+
 </beans>

spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/NotificationListeningChannelAdapterParserTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2010 the original author or authors.
+ * Copyright 2002-2012 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.
@@ -20,6 +20,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
 
 import javax.management.Notification;
 
@@ -37,6 +38,7 @@ import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 
 /**
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.0
  */
 @ContextConfiguration
@@ -46,9 +48,18 @@ public class NotificationListeningChannelAdapterParserTests {
 	@Autowired
 	private PollableChannel channel;
 
+	@Autowired
+	private PollableChannel patternChannel;
+
+	@Autowired
+	private PollableChannel multiChannel;
+
 	@Autowired
 	private TestPublisher testPublisher;
 
+	@Autowired
+	private TestPublisher testPublisher2;
+
 	@Autowired
 	private MessageChannel autoChannel;
 
@@ -59,10 +70,26 @@ public class NotificationListeningChannelAdapterParserTests {
 	public void receiveNotification() throws Exception {
 		assertNull(channel.receive(0));
 		testPublisher.send("ABC");
+		verifyReceipt(channel, "testPublisher");
+		verifyReceipt(patternChannel, "testPublisher");
+		// multiChannel should see 2 copies
+		verifyReceipt(multiChannel, "testPublisher");
+		verifyReceipt(multiChannel, "testPublisher");
+
+		testPublisher2.send("ABC");
+		assertNull(channel.receive(0));
+		assertNull(patternChannel.receive(0));
+		// multiChannel should see only 1 copy
+		verifyReceipt(multiChannel, "testPublisher2");
+		assertNull(multiChannel.receive(0));
+	}
+
+	private void verifyReceipt(PollableChannel channel, String beanName) {
 		Message<?> message = channel.receive(1000);
 		assertNotNull(message);
 		assertEquals(Notification.class, message.getPayload().getClass());
 		assertEquals("ABC", ((Notification) message.getPayload()).getMessage());
+		assertTrue(((String) ((Notification) message.getPayload()).getSource()).endsWith(beanName));
 	}
 
 	@Test

spring-integration-jmx/src/test/java/org/springframework/integration/jmx/config/TestPublisher.java

@@ -18,6 +18,7 @@ package org.springframework.integration.jmx.config;
 
 import javax.management.Notification;
 
+import org.springframework.beans.factory.BeanNameAware;
 import org.springframework.jmx.export.annotation.ManagedResource;
 import org.springframework.jmx.export.notification.NotificationPublisher;
 import org.springframework.jmx.export.notification.NotificationPublisherAware;
@@ -27,17 +28,22 @@ import org.springframework.jmx.export.notification.NotificationPublisherAware;
  * @since 2.0
  */
 @ManagedResource
-public class TestPublisher implements NotificationPublisherAware {
+public class TestPublisher implements NotificationPublisherAware, BeanNameAware {
 
 	private volatile NotificationPublisher notificationPublisher;
 
+	private String beanName;
+
 	public void setNotificationPublisher(NotificationPublisher notificationPublisher) {
 		this.notificationPublisher = notificationPublisher;
 	}
 
+	public void setBeanName(String name) {
+		this.beanName = name;
+	}
 	public void send(String s) {
 		Notification notification = new Notification("test.type",
-				"org.springframework.integration.jmx.config:type=TestPublisher,name=testPublisher", 1, s);
+				"org.springframework.integration.jmx.config:type=TestPublisher,name=" + this.beanName, 1, s);
 		this.notificationPublisher.sendNotification(notification);
 	}
 

src/reference/docbook/jmx.xml

@@ -56,6 +56,21 @@
 			event-driven and registered with the <interfacename>MBeanServer</interfacename>
 			directly. It does not require any poller configuration.
 		</para>
+		<note>
+			For this component only, the <emphasis>object-name</emphasis> attribute can contain an
+			ObjectName pattern (e.g. "org.foo:type=Bar,name=*") and the adapter will receive notifications
+			from all MBeans with ObjectNames that match the pattern. In addition, the <emphasis>object-name</emphasis>
+			attribute can contain a SpEL reference to a &lt;util:list/&gt; of ObjectName patterns:
+			<programlisting><![CDATA[<jmx:notification-listening-channel-adapter id="manyNotificationsAdapter"
+		channel="manyNotificationsChannel"
+		object-name="#{patterns}"/>
+
+<util:list id="patterns">
+	<value>org.foo:type=Foo,name=*</value>
+	<value>org.foo:type=Bar,name=*</value>
+</util:list>]]></programlisting>
+			The names of the located MBean(s) will be logged when DEBUG level logging is enabled.
+		</note>
 	</section>
 
 	<section id="jmx-notification-publishing-channel-adapter">