initial commit for NotificationPublishingAdapter
This commit is contained in:
Mark Fisher
@@ -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<Notification> {
|
||||
|
||||
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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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";
|
||||
|
||||
}
|
||||
|
||||
@@ -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<Notification> 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<Notification> 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<String, MBeanExporter> 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));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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<Notification> notifications = new ArrayList<Notification>();
|
||||
|
||||
public void handleNotification(Notification notification, Object handback) {
|
||||
this.notifications.add(notification);
|
||||
}
|
||||
|
||||
void clearNotifications() {
|
||||
this.notifications.clear();
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user