Merge pull request #554 from garyrussell/INT-2665

* INT-2665:
  INT-2665 JMX Endpoints And Remote MBeanServer

spring-integration-jmx/src/main/java/org/springframework/integration/jmx/AttributePollingMessageSource.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.
@@ -16,7 +16,7 @@
 
 package org.springframework.integration.jmx;
 
-import javax.management.MBeanServer;
+import javax.management.MBeanServerConnection;
 import javax.management.MalformedObjectNameException;
 import javax.management.ObjectName;
 
@@ -29,8 +29,9 @@ import org.springframework.util.Assert;
 /**
  * A {@link MessageSource} implementation that retrieves the current
  * value of a JMX attribute each time {@link #receive()} is invoked.
- * 
+ *
  * @author Mark Fisher
+ * @author Gary Russell
  * @since 2.0
  */
 public class AttributePollingMessageSource extends AbstractMessageSource<Object> {
@@ -39,18 +40,18 @@ public class AttributePollingMessageSource extends AbstractMessageSource<Object>
 
 	private volatile String attributeName;
 
-	private volatile MBeanServer server;
+	private volatile MBeanServerConnection server;
 
 
 	/**
 	 * Provide the MBeanServer where the JMX MBean has been registered.
 	 */
-	public void setServer(MBeanServer server) {
+	public void setServer(MBeanServerConnection server) {
 		this.server = server;
 	}
 
 	/**
-	 * Specify the String value of the JMX MBean's {@link ObjectName}.  
+	 * Specify the String value of the JMX MBean's {@link ObjectName}.
 	 */
 	public void setObjectName(String objectName) {
 		try {

spring-integration-jmx/src/main/java/org/springframework/integration/jmx/NotificationListeningMessageProducer.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.
@@ -16,9 +16,11 @@
 
 package org.springframework.integration.jmx;
 
+import java.io.IOException;
+
 import javax.management.InstanceNotFoundException;
 import javax.management.ListenerNotFoundException;
-import javax.management.MBeanServer;
+import javax.management.MBeanServerConnection;
 import javax.management.Notification;
 import javax.management.NotificationFilter;
 import javax.management.NotificationListener;
@@ -26,7 +28,6 @@ import javax.management.ObjectName;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-
 import org.springframework.integration.Message;
 import org.springframework.integration.endpoint.MessageProducerSupport;
 import org.springframework.integration.support.MessageBuilder;
@@ -35,15 +36,16 @@ 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
+ * @author Gary Russell
  * @since 2.0
  */
 public class NotificationListeningMessageProducer extends MessageProducerSupport implements NotificationListener {
 
 	private final Log logger = LogFactory.getLog(this.getClass());
 
-	private volatile MBeanServer server;
+	private volatile MBeanServerConnection server;
 
 	private volatile ObjectName objectName;
 
@@ -54,9 +56,9 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
 
 	/**
 	 * Provide a reference to the MBeanServer where the notification
-	 * publishing MBeans are registered. 
+	 * publishing MBeans are registered.
 	 */
-	public void setServer(MBeanServer server) {
+	public void setServer(MBeanServerConnection server) {
 		this.server = server;
 	}
 
@@ -118,7 +120,10 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
 			this.server.addNotificationListener(this.objectName, this, this.filter, this.handback);
 		}
 		catch (InstanceNotFoundException e) {
-			throw new IllegalStateException("Failed to find MBean instance.", e); 
+			throw new IllegalStateException("Failed to find MBean instance.", e);
+		}
+		catch (IOException e) {
+			throw new IllegalStateException("IOException on MBeanServerConnection.", e);
 		}
 	}
 
@@ -137,6 +142,9 @@ public class NotificationListeningMessageProducer extends MessageProducerSupport
 			catch (ListenerNotFoundException e) {
 				throw new IllegalStateException("Failed to find NotificationListener.", e);
 			}
+			catch (IOException e) {
+				throw new IllegalStateException("IOException on MBeanServerConnection.", e);
+			}
 		}
 	}
 

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

@@ -16,6 +16,7 @@
 
 package org.springframework.integration.jmx;
 
+import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -26,7 +27,7 @@ import javax.management.JMException;
 import javax.management.MBeanInfo;
 import javax.management.MBeanOperationInfo;
 import javax.management.MBeanParameterInfo;
-import javax.management.MBeanServer;
+import javax.management.MBeanServerConnection;
 import javax.management.MalformedObjectNameException;
 import javax.management.ObjectName;
 
@@ -50,7 +51,7 @@ import org.springframework.util.ObjectUtils;
  * will fallback to the defaults, if any have been configured on this instance via
  * {@link #setObjectName(String)} and {@link #setOperationName(String)},
  * respectively.
- * 
+ *
  * <p>The operation parameter(s), if any, must be available within the payload of the
  * Message being handled. If the target operation expects multiple parameters, they
  * can be provided in either a List or Map typed payload.
@@ -61,7 +62,7 @@ import org.springframework.util.ObjectUtils;
  */
 public class OperationInvokingMessageHandler extends AbstractReplyProducingMessageHandler implements InitializingBean {
 
-	private volatile MBeanServer server;
+	private volatile MBeanServerConnection server;
 
 	private volatile ObjectName objectName;
 
@@ -72,7 +73,7 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
 	 * Provide a reference to the MBeanServer within which the MBean
 	 * target for operation invocation has been registered.
 	 */
-	public void setServer(MBeanServer server) {
+	public void setServer(MBeanServerConnection server) {
 		this.server = server;
 	}
 
@@ -94,9 +95,9 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
 	/**
 	 * Specify an operation name to be invoked when no such
 	 * header is available on the Message being handled.
-	 */	
+	 */
 	public void setOperationName(String operationName) {
-		this.operationName = operationName; 
+		this.operationName = operationName;
 	}
 
 	@Override
@@ -111,7 +112,7 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
 		Map<String, Object> paramsFromMessage = this.resolveParameters(requestMessage);
 		try {
 			MBeanInfo mbeanInfo = this.server.getMBeanInfo(objectName);
-			MBeanOperationInfo[] opInfoArray = mbeanInfo.getOperations();	
+			MBeanOperationInfo[] opInfoArray = mbeanInfo.getOperations();
 			boolean hasNoArgOption = false;
 			for (MBeanOperationInfo opInfo : opInfoArray) {
 				if (operationName.equals(opInfo.getName())) {
@@ -146,9 +147,12 @@ public class OperationInvokingMessageHandler extends AbstractReplyProducingMessa
 		}
 		catch (JMException e) {
 			throw new MessageHandlingException(requestMessage, "failed to invoke JMX operation '" +
-					operationName + "' on MBean [" + objectName + "]" + " with " + 
+					operationName + "' on MBean [" + objectName + "]" + " with " +
 					paramsFromMessage.size() + " parameters: " + paramsFromMessage.keySet(), e);
 		}
+		catch (IOException e) {
+			throw new MessageHandlingException(requestMessage, "IOException on MBeanServerConnection", e);
+		}
 	}
 
 	/**

spring-integration-jmx/src/test/java/org/springframework/integration/monitor/RemoteMBeanServerTests-context.xml

@@ -0,0 +1,80 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:int-jmx="http://www.springframework.org/schema/integration/jmx"
+	xmlns:int="http://www.springframework.org/schema/integration"
+	xmlns:util="http://www.springframework.org/schema/util"
+	xsi:schemaLocation="http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
+		http://www.springframework.org/schema/integration/jmx http://www.springframework.org/schema/integration/jmx/spring-integration-jmx.xsd
+		http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
+		http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
+		http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
+
+	<context:mbean-server />
+	<context:mbean-export default-domain="foo" />
+
+	<bean id="test" class="org.springframework.integration.monitor.RemoteMBeanServerTests$Tester" />
+
+	<int-jmx:attribute-polling-channel-adapter
+			server="clientConnector"
+			attribute-name="Foo"
+			object-name="foo:name=test,type=RemoteMBeanServerTests.Tester"
+			channel="attrChannel">
+		<int:poller fixed-rate="2000" />
+	</int-jmx:attribute-polling-channel-adapter>
+
+	<int:channel id="attrChannel">
+		<int:queue />
+	</int:channel>
+
+	<int:channel id="opChannel" />
+
+	<int-jmx:operation-invoking-outbound-gateway
+		server="clientConnector"
+		operation-name="fooBar"
+		object-name="foo:name=test,type=RemoteMBeanServerTests.Tester"
+		request-channel="opChannel"
+		reply-channel="opOutChannel" />
+
+	<int:channel id="opOutChannel">
+		<int:queue />
+	</int:channel>
+
+	<int:channel id="publishChannel" />
+
+	<int-jmx:notification-publishing-channel-adapter
+		channel="publishChannel"
+		object-name="foo:name=pub"
+		default-notification-type="foo" />
+
+	<int-jmx:notification-listening-channel-adapter
+		server="clientConnector"
+		channel="publishInChannel"
+		object-name="foo:name=pub"/>
+
+	<int:channel id="publishInChannel">
+		<int:queue />
+	</int:channel>
+
+	<!--
+		To test against the MBeanServerConnection instead of the (default) MBeanServer, use
+			 -Dspring.profiles.active=remote
+			 -Dcom.sun.management.jmxremote.port=11099<p/>
+			 -Dcom.sun.management.jmxremote.authenticate=false<p/>
+			 -Dcom.sun.management.jmxremote.ssl=false<p/>
+
+		The CI build runs with the default profile.
+	 -->
+
+	<beans profile="remote">
+		<bean id="clientConnector" class="org.springframework.jmx.support.MBeanServerConnectionFactoryBean">
+			<property name="serviceUrl" value="service:jmx:rmi://localhost/jndi/rmi://localhost:11099/jmxrmi"/>
+		</bean>
+	</beans>
+
+	<beans profile="default">
+		<context:mbean-server id="clientConnector"/>
+	</beans>
+
+</beans>

spring-integration-jmx/src/test/java/org/springframework/integration/monitor/RemoteMBeanServerTests.java

@@ -0,0 +1,110 @@
+/*
+ * 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.
+ * 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.monitor;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import javax.management.Notification;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.Message;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.core.PollableChannel;
+import org.springframework.integration.message.GenericMessage;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+/**
+ * To run, add JVM Args:
+ * <p/>
+ * -Dspring.profiles.active=remote
+ * -Dcom.sun.management.jmxremote.port=11099<p/>
+ * -Dcom.sun.management.jmxremote.authenticate=false<p/>
+ * -Dcom.sun.management.jmxremote.ssl=false<p/>
+ *
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class RemoteMBeanServerTests {
+
+	@Autowired
+	private PollableChannel attrChannel;
+
+	@Autowired
+	private MessageChannel publishChannel;
+
+	@Autowired
+	private PollableChannel publishInChannel;
+
+	@Autowired
+	private MessageChannel opChannel;
+
+	@Autowired
+	private PollableChannel opOutChannel;
+
+	@Test
+	public void testAttrPoll() throws Exception {
+		Message<?> message = attrChannel.receive(100000);
+		assertNotNull(message);
+		assertEquals("foo", message.getPayload());
+	}
+
+	@Test
+	public void testPublish() {
+		publishChannel.send(new GenericMessage<String>("bar"));
+		Message<?> message = publishInChannel.receive(100000);
+		assertNotNull(message);
+		assertTrue(message.getPayload() instanceof Notification);
+		Notification notification = (Notification) message.getPayload();
+		assertEquals("bar", notification.getMessage());
+		assertEquals("foo", notification.getType());
+	}
+
+	@Test
+	public void testOperation() {
+		opChannel.send(new GenericMessage<String>("foo"));
+		Message<?> message = opOutChannel.receive(100000);
+		assertNotNull(message);
+		assertEquals("bar", message.getPayload());
+	}
+
+	public static interface TesterMBean {
+
+		String getFoo();
+
+		String fooBar(String foo);
+	}
+
+	public static class Tester implements TesterMBean {
+
+		public String getFoo() {
+			return "foo";
+		}
+
+		public String fooBar(String foo) {
+			return "bar";
+		}
+
+
+	}
+}