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 new file mode 100644 index 0000000000..74f0a4b63c --- /dev/null +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/DefaultNotificationMapper.java @@ -0,0 +1,75 @@ +/* + * 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; + +import java.util.concurrent.atomic.AtomicInteger; + +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; + +/** + * @author Mark Fisher + * @since 2.0 + */ +class DefaultNotificationMapper implements OutboundMessageMapper { + + private final ObjectName sourceObjectName; + + private volatile String defaultNotificationType = "SpringIntegrationNotification"; + + private final AtomicInteger sequence = new AtomicInteger(); + + + DefaultNotificationMapper(ObjectName sourceObjectName) { + 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()); + } + + private String resolveNotificationType(Message message) { + String type = message.getHeaders().get(JmxHeaders.NOTIFICATION_TYPE, String.class); + 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/JmxHeaders.java b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/JmxHeaders.java index 017ea36a06..b22175ac19 100644 --- a/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/JmxHeaders.java +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/JmxHeaders.java @@ -26,6 +26,8 @@ public abstract class JmxHeaders { private static final String PREFIX = MessageHeaders.PREFIX + "jmx"; - public static final String NOTIFICATION_HANDBACK = PREFIX + "_handback"; + public static final String NOTIFICATION_HANDBACK = PREFIX + "_notificationHandback"; + + public static final String NOTIFICATION_TYPE = PREFIX + "_notificationType"; } 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 new file mode 100644 index 0000000000..113c6a6171 --- /dev/null +++ b/org.springframework.integration.jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingAdapter.java @@ -0,0 +1,109 @@ +/* + * 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; + +import java.util.Map; + +import javax.management.Notification; + +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; +import org.springframework.integration.handler.AbstractMessageHandler; +import org.springframework.integration.message.OutboundMessageMapper; +import org.springframework.jmx.export.MBeanExporter; +import org.springframework.jmx.export.notification.NotificationPublisher; +import org.springframework.jmx.export.notification.NotificationPublisherAware; +import org.springframework.jmx.support.ObjectNameManager; +import org.springframework.util.Assert; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class NotificationPublishingAdapter extends AbstractMessageHandler + implements NotificationPublisherAware, BeanNameAware, BeanFactoryAware, InitializingBean { + + private volatile NotificationPublisher notificationPublisher; + + private volatile OutboundMessageMapper notificationMapper; + + private volatile String objectName; + + private volatile String beanName; + + private volatile ListableBeanFactory beanFactory; + + + public NotificationPublishingAdapter() { + this(null); + } + + public NotificationPublishingAdapter(String objectName) { + this.objectName = objectName; + } + + + public void setNotificationMapper(OutboundMessageMapper notificationMapper) { + this.notificationMapper = notificationMapper; + } + + public void setNotificationPublisher(NotificationPublisher notificationPublisher) { + this.notificationPublisher = notificationPublisher; + } + + public void setBeanName(String beanName) { + this.beanName = beanName; + } + + public void setBeanFactory(BeanFactory beanFactory) throws BeansException { + Assert.isTrue(beanFactory instanceof ListableBeanFactory, "A ListableBeanFactory is required."); + this.beanFactory = (ListableBeanFactory) beanFactory; + } + + public void afterPropertiesSet() throws Exception { + Map exporters = BeanFactoryUtils.beansOfTypeIncludingAncestors( + this.beanFactory, MBeanExporter.class); + Assert.isTrue(exporters.size() == 1, + "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); + } + exporter.registerManagedResource(this, ObjectNameManager.getInstance(this.objectName)); + if (this.logger.isInfoEnabled()) { + this.logger.info("Registered NotificationPublishingAdapter as MBean with ObjectName: " + this.objectName); + } + } + + @Override + protected void handleMessageInternal(Message message) throws Exception { + Assert.state(this.notificationPublisher != null, "NotificationPublisher must not be null."); + this.notificationPublisher.sendNotification(this.notificationMapper.fromMessage(message)); + } + +} 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 new file mode 100644 index 0000000000..e58e399ed7 --- /dev/null +++ b/org.springframework.integration.jmx/src/test/java/org/springframework/integration/jmx/NotificationPublishingAdapterTests.java @@ -0,0 +1,93 @@ +/* + * 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; + +import static org.junit.Assert.assertEquals; + +import java.util.ArrayList; +import java.util.List; + +import javax.management.Notification; +import javax.management.NotificationListener; +import javax.management.ObjectName; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import org.springframework.context.support.StaticApplicationContext; +import org.springframework.integration.message.StringMessage; +import org.springframework.jmx.export.MBeanExporter; +import org.springframework.jmx.support.ObjectNameManager; + +/** + * @author Mark Fisher + * @since 2.0 + */ +public class NotificationPublishingAdapterTests { + + private final StaticApplicationContext context = new StaticApplicationContext(); + + private final TestNotificationListener listener = new TestNotificationListener(); + + private volatile ObjectName publisherObjectName; + + + @Before + public void setup() throws Exception { + this.publisherObjectName = ObjectNameManager.getInstance( + "org.springframework.integration:type=notificationPublishingAdapter,name=testPublisher"); + context.registerSingleton("exporter", MBeanExporter.class); + context.registerSingleton("testPublisher", NotificationPublishingAdapter.class); + context.refresh(); + MBeanExporter exporter = context.getBean(MBeanExporter.class); + exporter.getServer().addNotificationListener(this.publisherObjectName, this.listener, null, null); + } + + @After + public void cleanup() { + this.listener.clearNotifications(); + } + + + @Test + public void simplePublish() { + NotificationPublishingAdapter adapter = context.getBean(NotificationPublishingAdapter.class); + assertEquals(0, this.listener.notifications.size()); + adapter.handleMessage(new StringMessage("foo")); + assertEquals(1, this.listener.notifications.size()); + Notification notification = this.listener.notifications.get(0); + assertEquals(this.publisherObjectName, notification.getSource()); + assertEquals("foo", notification.getMessage()); + assertEquals("SpringIntegrationNotification", notification.getType()); + } + + + public static class TestNotificationListener implements NotificationListener { + + private final List notifications = new ArrayList(); + + public void handleNotification(Notification notification, Object handback) { + this.notifications.add(notification); + } + + void clearNotifications() { + this.notifications.clear(); + } + } + +}