Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Build on top of Spring 4's new messaging types

Browse Source 
This commit updates Spring Integration to depend upon Spring 4, making
use of the message types that have moved from Spring Integration into
Spring's new spring-messaging module.

The default message converter no longer supports conversion of a
message that is null, throwing an IllegalArgumentException if an
attempt is made to convert null. Furthermore, GenericMessagingTemplate
does not support sending null, again throwing an
IllegalArgumentException. Previously, MessagingTemplate had no-oped an
attempt to send null.

ConcurrentAggregatorTests and AggregatorTests both had a single test
that was specifically testing the behaviour of an aggregator that
returns null for its message. These tests have been removed.

CorrelatingMessageHandlerTests have been updated to specify some
additional behaviour for its mocks so that null messages are not
returned.

These are the only functional changes that have been made. All other
changes are simply for moving to the repackaged and/or renamed types.

In the move to being part of core Spring, a number of constants and
header accessor methods have moved from MessageHeaders to
MessageHeaderAccessor. This commit continues this pattern for
the enterprise integration headers that are specific to Spring
Integration. A new class, EiMessageHeaderAccessor, has been created.
This class provides constants and methods for working with SI-specific
headers. The main code and tests have been updated to use this new
class.
This commit is contained in:
Andy Wilkinson
2013-09-10 13:35:53 +01:00
committed by Gary Russell
parent 8dca61c62a
commit 28ab8394de
1156 changed files with 3332 additions and 5834 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/AttributePollingMessageSource.java
View File

@@ -20,7 +20,7 @@ import javax.management.MBeanServerConnection;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.springframework.integration.MessagingException;
import org.springframework.messaging.MessagingException;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.jmx.support.ObjectNameManager;

2
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/DefaultNotificationMapper.java
View File

@@ -21,7 +21,7 @@ import java.util.concurrent.atomic.AtomicLong;
import javax.management.Notification;
import javax.management.ObjectName;
import org.springframework.integration.Message;
import org.springframework.messaging.Message;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.util.Assert;

2
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/MBeanTreePollingMessageSource.java
View File

@@ -25,9 +25,9 @@ import javax.management.ObjectInstance;
import javax.management.ObjectName;
import javax.management.QueryExp;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.endpoint.AbstractMessageSource;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
/**

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

@@ -34,7 +34,7 @@ import javax.management.ObjectName;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.Message;
import org.springframework.messaging.Message;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;

2
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/NotificationPublishingMessageHandler.java
View File

@@ -26,7 +26,7 @@ import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.BeanFactoryUtils;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.integration.Message;
import org.springframework.messaging.Message;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.integration.mapping.OutboundMessageMapper;
import org.springframework.integration.monitor.IntegrationMBeanExporter;

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

@@ -32,12 +32,11 @@ import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.integration.Message;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.jmx.support.ObjectNameManager;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.util.Assert;
import org.springframework.util.ObjectUtils;
@@ -63,172 +62,172 @@ import org.springframework.util.ObjectUtils;
*/
public class OperationInvokingMessageHandler extends AbstractReplyProducingMessageHandler implements InitializingBean {
private volatile MBeanServerConnection server;
private volatile MBeanServerConnection server;
private volatile ObjectName objectName;
private volatile ObjectName objectName;
private volatile String operationName;
private volatile String operationName;
/**
* Provide a reference to the MBeanServer within which the MBean
* target for operation invocation has been registered.
*/
public void setServer(MBeanServerConnection server) {
this.server = server;
}
/**
* Provide a reference to the MBeanServer within which the MBean
* target for operation invocation has been registered.
*/
public void setServer(MBeanServerConnection server) {
this.server = server;
}
/**
* Specify a default ObjectName to use when no such header is
* available on the Message being handled.
*/
public void setObjectName(String objectName) {
try {
if (objectName != null) {
this.objectName = ObjectNameManager.getInstance(objectName);
}
}
catch (MalformedObjectNameException e) {
throw new IllegalArgumentException(e);
}
}
/**
* Specify a default ObjectName to use when no such header is
* available on the Message being handled.
*/
public void setObjectName(String objectName) {
try {
if (objectName != null) {
this.objectName = ObjectNameManager.getInstance(objectName);
}
}
catch (MalformedObjectNameException e) {
throw new IllegalArgumentException(e);
}
}
/**
* 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;
}
/**
* 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;
}
@Override
public final void onInit() {
Assert.notNull(this.server, "MBeanServer is required.");
super.onInit();
}
@Override
public final void onInit() {
Assert.notNull(this.server, "MBeanServer is required.");
super.onInit();
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
ObjectName objectName = this.resolveObjectName(requestMessage);
String operationName = this.resolveOperationName(requestMessage);
Map<String, Object> paramsFromMessage = this.resolveParameters(requestMessage);
try {
MBeanInfo mbeanInfo = this.server.getMBeanInfo(objectName);
MBeanOperationInfo[] opInfoArray = mbeanInfo.getOperations();
boolean hasNoArgOption = false;
for (MBeanOperationInfo opInfo : opInfoArray) {
if (operationName.equals(opInfo.getName())) {
MBeanParameterInfo[] paramInfoArray = opInfo.getSignature();
if (paramInfoArray.length == 0) {
hasNoArgOption = true;
}
if (paramInfoArray.length == paramsFromMessage.size()) {
int index = 0;
Object values[] = new Object[paramInfoArray.length];
String signature[] = new String[paramInfoArray.length];
for (MBeanParameterInfo paramInfo : paramInfoArray) {
Object value = paramsFromMessage.get(paramInfo.getName());
if (value == null) {
/*
* With Spring 3.2.3 and greater, the parameter names are
* registered instead of the JVM's default p1, p2 etc.
* Fall back to that naming style if not found.
*/
value = paramsFromMessage.get("p" + (index + 1));
}
if (value != null && value.getClass().getName().equals(paramInfo.getType())) {
values[index] = value;
signature[index] = paramInfo.getType();
index++;
}
}
if (index == paramInfoArray.length) {
return this.server.invoke(objectName, operationName, values, signature);
}
}
}
}
if (hasNoArgOption) {
return this.server.invoke(objectName, operationName, null, null);
}
throw new MessagingException(requestMessage, "failed to find JMX operation '"
+ operationName + "' on MBean [" + objectName + "] of type [" + mbeanInfo.getClassName()
+ "] with " + paramsFromMessage.size() + " parameters: " + paramsFromMessage.keySet());
}
catch (JMException e) {
throw new MessageHandlingException(requestMessage, "failed to invoke JMX operation '" +
operationName + "' on MBean [" + objectName + "]" + " with " +
paramsFromMessage.size() + " parameters: " + paramsFromMessage.keySet(), e);
}
catch (IOException e) {
throw new MessageHandlingException(requestMessage, "IOException on MBeanServerConnection", e);
}
}
@Override
protected Object handleRequestMessage(Message<?> requestMessage) {
ObjectName objectName = this.resolveObjectName(requestMessage);
String operationName = this.resolveOperationName(requestMessage);
Map<String, Object> paramsFromMessage = this.resolveParameters(requestMessage);
try {
MBeanInfo mbeanInfo = this.server.getMBeanInfo(objectName);
MBeanOperationInfo[] opInfoArray = mbeanInfo.getOperations();
boolean hasNoArgOption = false;
for (MBeanOperationInfo opInfo : opInfoArray) {
if (operationName.equals(opInfo.getName())) {
MBeanParameterInfo[] paramInfoArray = opInfo.getSignature();
if (paramInfoArray.length == 0) {
hasNoArgOption = true;
}
if (paramInfoArray.length == paramsFromMessage.size()) {
int index = 0;
Object values[] = new Object[paramInfoArray.length];
String signature[] = new String[paramInfoArray.length];
for (MBeanParameterInfo paramInfo : paramInfoArray) {
Object value = paramsFromMessage.get(paramInfo.getName());
if (value == null) {
/*
* With Spring 3.2.3 and greater, the parameter names are
* registered instead of the JVM's default p1, p2 etc.
* Fall back to that naming style if not found.
*/
value = paramsFromMessage.get("p" + (index + 1));
}
if (value != null && value.getClass().getName().equals(paramInfo.getType())) {
values[index] = value;
signature[index] = paramInfo.getType();
index++;
}
}
if (index == paramInfoArray.length) {
return this.server.invoke(objectName, operationName, values, signature);
}
}
}
}
if (hasNoArgOption) {
return this.server.invoke(objectName, operationName, null, null);
}
throw new MessagingException(requestMessage, "failed to find JMX operation '"
+ operationName + "' on MBean [" + objectName + "] of type [" + mbeanInfo.getClassName()
+ "] with " + paramsFromMessage.size() + " parameters: " + paramsFromMessage.keySet());
}
catch (JMException e) {
throw new MessageHandlingException(requestMessage, "failed to invoke JMX operation '" +
operationName + "' on MBean [" + objectName + "]" + " with " +
paramsFromMessage.size() + " parameters: " + paramsFromMessage.keySet(), e);
}
catch (IOException e) {
throw new MessageHandlingException(requestMessage, "IOException on MBeanServerConnection", e);
}
}
/**
* First checks if defaultObjectName is set, otherwise falls back on {@link JmxHeaders#OBJECT_NAME} header.
*/
private ObjectName resolveObjectName(Message<?> message) {
ObjectName objectName = this.objectName;
if (objectName == null){
Object objectNameHeader = message.getHeaders().get(JmxHeaders.OBJECT_NAME);
if (objectNameHeader instanceof ObjectName) {
objectName = (ObjectName) objectNameHeader;
}
else if (objectNameHeader instanceof String) {
try {
objectName = ObjectNameManager.getInstance(objectNameHeader);
}
catch (MalformedObjectNameException e) {
throw new IllegalArgumentException(e);
}
}
}
Assert.notNull(objectName, "Failed to resolve ObjectName.");
return objectName;
}
/**
* First checks if defaultObjectName is set, otherwise falls back on {@link JmxHeaders#OBJECT_NAME} header.
*/
private ObjectName resolveObjectName(Message<?> message) {
ObjectName objectName = this.objectName;
if (objectName == null){
Object objectNameHeader = message.getHeaders().get(JmxHeaders.OBJECT_NAME);
if (objectNameHeader instanceof ObjectName) {
objectName = (ObjectName) objectNameHeader;
}
else if (objectNameHeader instanceof String) {
try {
objectName = ObjectNameManager.getInstance(objectNameHeader);
}
catch (MalformedObjectNameException e) {
throw new IllegalArgumentException(e);
}
}
}
Assert.notNull(objectName, "Failed to resolve ObjectName.");
return objectName;
}
/**
* First checks if defaultOperationName is set, otherwise falls back on {@link JmxHeaders#OPERATION_NAME} header.
*/
private String resolveOperationName(Message<?> message) {
String operationName = this.operationName;
if (operationName == null){
operationName = message.getHeaders().get(JmxHeaders.OPERATION_NAME, String.class);
}
Assert.notNull(operationName, "Failed to resolve operation name.");
return operationName;
}
/**
* First checks if defaultOperationName is set, otherwise falls back on {@link JmxHeaders#OPERATION_NAME} header.
*/
private String resolveOperationName(Message<?> message) {
String operationName = this.operationName;
if (operationName == null){
operationName = message.getHeaders().get(JmxHeaders.OPERATION_NAME, String.class);
}
Assert.notNull(operationName, "Failed to resolve operation name.");
return operationName;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
private Map<String, Object> resolveParameters(Message<?> message) {
Map<String, Object> map = null;
if (message.getPayload() instanceof Map) {
map = (Map<String, Object>) message.getPayload();
}
else if (message.getPayload() instanceof List) {
map = this.createParameterMapFromList((List) message.getPayload());
}
else if (message.getPayload() != null && message.getPayload().getClass().isArray()) {
map = this.createParameterMapFromList(
Arrays.asList(ObjectUtils.toObjectArray(message.getPayload())));
}
else if (message.getPayload() != null) {
map = this.createParameterMapFromList(Collections.singletonList(message.getPayload()));
}
else {
map = Collections.EMPTY_MAP;
}
return map;
}
@SuppressWarnings({ "unchecked", "rawtypes" })
private Map<String, Object> resolveParameters(Message<?> message) {
Map<String, Object> map = null;
if (message.getPayload() instanceof Map) {
map = (Map<String, Object>) message.getPayload();
}
else if (message.getPayload() instanceof List) {
map = this.createParameterMapFromList((List) message.getPayload());
}
else if (message.getPayload() != null && message.getPayload().getClass().isArray()) {
map = this.createParameterMapFromList(
Arrays.asList(ObjectUtils.toObjectArray(message.getPayload())));
}
else if (message.getPayload() != null) {
map = this.createParameterMapFromList(Collections.singletonList(message.getPayload()));
}
else {
map = Collections.EMPTY_MAP;
}
return map;
}
@SuppressWarnings("rawtypes")
private Map<String, Object> createParameterMapFromList(List parameters) {
Map<String, Object> map = new HashMap<String, Object>();
for (int i = 0; i < parameters.size(); i++) {
map.put("p" + (i + 1), parameters.get(i));
}
return map;
}
@SuppressWarnings("rawtypes")
private Map<String, Object> createParameterMapFromList(List parameters) {
Map<String, Object> map = new HashMap<String, Object>();
for (int i = 0; i < parameters.size(); i++) {
map.put("p" + (i + 1), parameters.get(i));
}
return map;
}
}
}

2
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/config/package-info.java
View File

@@ -1,4 +1,4 @@
/**
* Provides classes for configuration - parsers, namespace handlers.
*/
package org.springframework.integration.jmx.config;
package org.springframework.integration.jmx.config;

2
spring-integration-jmx/src/main/java/org/springframework/integration/jmx/package-info.java
View File

@@ -1,4 +1,4 @@
/**
* Base package for JMX support.
*/
package org.springframework.integration.jmx;
package org.springframework.integration.jmx;

4
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/DirectChannelMetrics.java
View File

@@ -19,8 +19,8 @@ import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.util.StopWatch;

8
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/IntegrationMBeanExporter.java
View File

@@ -56,13 +56,13 @@ import org.springframework.context.ApplicationContextAware;
import org.springframework.context.Lifecycle;
import org.springframework.context.SmartLifecycle;
import org.springframework.core.task.support.ExecutorServiceAdapter;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessagingException;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.core.MessageHandler;
import org.springframework.messaging.MessageHandler;
import org.springframework.integration.core.MessageSource;
import org.springframework.integration.core.PollableChannel;
import org.springframework.messaging.PollableChannel;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.support.context.NamedComponent;
import org.springframework.jmx.export.MBeanExporter;

2
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/PollableChannelMetrics.java
View File

@@ -19,7 +19,7 @@ package org.springframework.integration.monitor;
import java.util.concurrent.atomic.AtomicInteger;
import org.aopalliance.intercept.MethodInvocation;
import org.springframework.integration.MessageChannel;
import org.springframework.messaging.MessageChannel;
import org.springframework.jmx.export.annotation.ManagedMetric;
import org.springframework.jmx.export.annotation.ManagedOperation;
import org.springframework.jmx.support.MetricType;

4
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/SimpleMessageHandlerMetrics.java
View File

@@ -22,8 +22,8 @@ import org.aopalliance.intercept.MethodInterceptor;
import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.Message;
import org.springframework.integration.core.MessageHandler;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHandler;
import org.springframework.jmx.export.annotation.ManagedResource;
import org.springframework.util.StopWatch;

2
spring-integration-jmx/src/main/java/org/springframework/integration/monitor/package-info.java
View File

@@ -1,4 +1,4 @@
/**
* Provides classes related to Spring Integration managed resources.
*/
package org.springframework.integration.monitor;
package org.springframework.integration.monitor;