Merge pull request #189 from olegz/INT-2083

HeaderMapper refactoring
This commit is contained in:
Mark Fisher
2011-11-22 17:07:41 -05:00
50 changed files with 2330 additions and 500 deletions

View File

@@ -491,6 +491,7 @@ project('spring-integration-ws') {
compile("javax.activation:activation:$javaxActivationVersion") { optional = true }
testCompile project(":spring-integration-test")
testCompile "stax:stax-api:1.0.1"
testCompile "xstream:xstream:1.2.2"
}
// suppress saaj path warnings

View File

@@ -20,20 +20,21 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import org.w3c.dom.Element;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* Base class for inbound adapter parsers for the AMQP namespace.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.1
*/
abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefinitionParser {
@@ -99,7 +100,9 @@ abstract class AbstractAmqpInboundAdapterParser extends AbstractSingleBeanDefini
builder.addConstructorArgValue(listenerContainerBeanDef);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "message-converter");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "header-mapper");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "error-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "auto-startup");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "phase");

View File

@@ -20,6 +20,7 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
@@ -29,6 +30,7 @@ import org.w3c.dom.Element;
* Parser for the AMQP 'outbound-channel-adapter' element.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.1
*/
public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
@@ -45,6 +47,9 @@ public class AmqpOutboundChannelAdapterParser extends AbstractOutboundChannelAda
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "exchange-name-expression");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);
return builder.getBeanDefinition();
}

View File

@@ -18,6 +18,7 @@ import org.w3c.dom.Element;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.amqp.support.DefaultAmqpHeaderMapper;
import org.springframework.integration.config.xml.AbstractConsumerEndpointParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.util.StringUtils;
@@ -26,6 +27,7 @@ import org.springframework.util.StringUtils;
* Parser for the AMQP 'outbound-channel-adapter' element.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.1
*/
public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
@@ -49,6 +51,9 @@ public class AmqpOutboundGatewayParser extends AbstractConsumerEndpointParser {
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "routing-key-expression");
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel", "outputChannel");
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultAmqpHeaderMapper.class, null);
return builder;
}

View File

@@ -69,7 +69,7 @@ public class AmqpInboundChannelAdapter extends MessageProducerSupport {
this.messageListenerContainer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
Object payload = messageConverter.fromMessage(message);
Map<String, ?> headers = headerMapper.toHeaders(message.getMessageProperties());
Map<String, ?> headers = headerMapper.toHeadersFromRequest(message.getMessageProperties());
sendMessage(MessageBuilder.withPayload(payload).copyHeaders(headers).build());
}
});

View File

@@ -81,7 +81,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
this.messageListenerContainer.setMessageListener(new MessageListener() {
public void onMessage(Message message) {
Object payload = amqpMessageConverter.fromMessage(message);
Map<String, ?> headers = headerMapper.toHeaders(message.getMessageProperties());
Map<String, ?> headers = headerMapper.toHeadersFromRequest(message.getMessageProperties());
org.springframework.integration.Message<?> request =
MessageBuilder.withPayload(payload).copyHeaders(headers).build();
final org.springframework.integration.Message<?> reply = sendAndReceiveMessage(request);
@@ -97,7 +97,7 @@ public class AmqpInboundGateway extends MessagingGatewaySupport {
String contentEncoding = messageProperties.getContentEncoding();
long contentLength = messageProperties.getContentLength();
String contentType = messageProperties.getContentType();
headerMapper.fromHeaders(reply.getHeaders(), messageProperties);
headerMapper.fromHeadersToReply(reply.getHeaders(), messageProperties);
// clear the replyTo from the original message since we are using it now
messageProperties.setReplyTo(null);
// reset the content-* properties as determined by the MessageConverter

View File

@@ -37,6 +37,7 @@ import org.springframework.util.Assert;
* Adapter that converts and sends Messages to an AMQP Exchange.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.1
*/
public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
@@ -62,7 +63,6 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
private volatile AmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
@Override
protected void onInit() {
super.onInit();
@@ -84,6 +84,10 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
Assert.notNull(amqpTemplate, "AmqpTemplate must not be null");
this.amqpTemplate = amqpTemplate;
}
public void setHeaderMapper(AmqpHeaderMapper headerMapper) {
this.headerMapper = headerMapper;
}
public void setExchangeName(String exchangeName) {
this.exchangeName = exchangeName;
@@ -134,7 +138,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
new MessagePostProcessor() {
public org.springframework.amqp.core.Message postProcessMessage(
org.springframework.amqp.core.Message message) throws AmqpException {
headerMapper.fromHeaders(requestMessage.getHeaders(), message.getMessageProperties());
headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), message.getMessageProperties());
return message;
}
});
@@ -145,7 +149,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
Assert.isTrue(amqpTemplate instanceof RabbitTemplate, "RabbitTemplate implementation is required for send and receive");
MessageConverter converter = ((RabbitTemplate) this.amqpTemplate).getMessageConverter();
MessageProperties amqpMessageProperties = new MessageProperties();
this.headerMapper.fromHeaders(requestMessage.getHeaders(), amqpMessageProperties);
this.headerMapper.fromHeadersToRequest(requestMessage.getHeaders(), amqpMessageProperties);
org.springframework.amqp.core.Message amqpMessage = converter.toMessage(requestMessage.getPayload(), amqpMessageProperties);
org.springframework.amqp.core.Message amqpReplyMessage = this.amqpTemplate.sendAndReceive(exchangeName, routingKey, amqpMessage);
if (amqpReplyMessage == null) {
@@ -155,7 +159,7 @@ public class AmqpOutboundEndpoint extends AbstractReplyProducingMessageHandler {
MessageBuilder<?> builder = (replyObject instanceof Message)
? MessageBuilder.fromMessage((Message<?>) replyObject)
: MessageBuilder.withPayload(replyObject);
Map<String, ?> headers = this.headerMapper.toHeaders(amqpReplyMessage.getMessageProperties());
Map<String, ?> headers = this.headerMapper.toHeadersFromReply(amqpReplyMessage.getMessageProperties());
builder.copyHeadersIfAbsent(headers);
return builder.build();
}

View File

@@ -18,12 +18,15 @@ package org.springframework.integration.amqp.support;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.mapping.RequestReplyHeaderMapper;
/**
* A convenience interface that extends {@link HeaderMapper}
* but parameterized with {@link MessageProperties}.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.1
*/
public interface AmqpHeaderMapper extends HeaderMapper<MessageProperties> {
public interface AmqpHeaderMapper extends RequestReplyHeaderMapper<MessageProperties> {
}

View File

@@ -16,20 +16,17 @@
package org.springframework.integration.amqp.support;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.integration.mapping.AbstractHeaderMapper;
import org.springframework.util.StringUtils;
/**
@@ -47,166 +44,36 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
* @since 2.1
*/
public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper {
public class DefaultAmqpHeaderMapper extends AbstractHeaderMapper<MessageProperties> implements AmqpHeaderMapper {
private static final List<String> STANDARD_HEADER_NAMES = new ArrayList<String>();
private static final String[] TRANSIENT_HEADER_NAMES = new String[] {
MessageHeaders.ID,
MessageHeaders.ERROR_CHANNEL,
MessageHeaders.REPLY_CHANNEL,
MessageHeaders.TIMESTAMP
};
private final Log logger = LogFactory.getLog(this.getClass());
private volatile String inboundPrefix = "";
private volatile String outboundPrefix = "";
/**
* Specify a prefix to be appended to the integration message header name for
* any user-defined AMQP header that is being mapped into the MessageHeaders.
* The Default is an empty string (no prefix).
* <p/>
* This does not affect the standard AMQP properties, such as contentType, etc.
* The header names used for mapping such properties are all defined in the
* {@link AmqpHeaders} class as constants.
*/
public void setInboundPrefix(String inboundPrefix) {
this.inboundPrefix = (inboundPrefix != null) ? inboundPrefix : "";
static {
STANDARD_HEADER_NAMES.add(AmqpHeaders.APP_ID);
STANDARD_HEADER_NAMES.add(AmqpHeaders.CLUSTER_ID);
STANDARD_HEADER_NAMES.add(AmqpHeaders.CONTENT_ENCODING);
STANDARD_HEADER_NAMES.add(AmqpHeaders.CONTENT_LENGTH);
STANDARD_HEADER_NAMES.add(AmqpHeaders.CONTENT_TYPE);
STANDARD_HEADER_NAMES.add(AmqpHeaders.CORRELATION_ID);
STANDARD_HEADER_NAMES.add(AmqpHeaders.DELIVERY_MODE);
STANDARD_HEADER_NAMES.add(AmqpHeaders.DELIVERY_TAG);
STANDARD_HEADER_NAMES.add(AmqpHeaders.EXPIRATION);
STANDARD_HEADER_NAMES.add(AmqpHeaders.MESSAGE_COUNT);
STANDARD_HEADER_NAMES.add(AmqpHeaders.MESSAGE_ID);
STANDARD_HEADER_NAMES.add(AmqpHeaders.RECEIVED_EXCHANGE);
STANDARD_HEADER_NAMES.add(AmqpHeaders.RECEIVED_ROUTING_KEY);
STANDARD_HEADER_NAMES.add(AmqpHeaders.REDELIVERED);
STANDARD_HEADER_NAMES.add(AmqpHeaders.REPLY_TO);
STANDARD_HEADER_NAMES.add(AmqpHeaders.TIMESTAMP);
STANDARD_HEADER_NAMES.add(AmqpHeaders.TYPE);
STANDARD_HEADER_NAMES.add(AmqpHeaders.USER_ID);
}
/**
* Specify a prefix to be appended to the AMQP header name for any
* integration message header that is being mapped into the AMQP Message.
* The Default is an empty string (no prefix).
* <p/>
* This does not affect the standard AMQP properties, such as contentType, etc.
* The header names used for mapping such properties are all defined in
* the {@link AmqpHeaders} class as constants.
* Extract "standard" headers from an AMQP MessageProperties instance.
*/
public void setOutboundPrefix(String outboundPrefix) {
this.outboundPrefix = (outboundPrefix != null) ? outboundPrefix : "";
}
/**
* Maps headers from a Spring Integration MessageHeaders instance to the MessageProperties
* of an AMQP Message.
*/
public void fromHeaders(MessageHeaders headers, MessageProperties amqpMessageProperties) {
try {
String appId = getHeaderIfAvailable(headers, AmqpHeaders.APP_ID, String.class);
if (StringUtils.hasText(appId)) {
amqpMessageProperties.setAppId(appId);
}
String clusterId = getHeaderIfAvailable(headers, AmqpHeaders.CLUSTER_ID, String.class);
if (StringUtils.hasText(clusterId)) {
amqpMessageProperties.setClusterId(clusterId);
}
String contentEncoding = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_ENCODING, String.class);
if (StringUtils.hasText(contentEncoding)) {
amqpMessageProperties.setContentEncoding(contentEncoding);
}
Long contentLength = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_LENGTH, Long.class);
if (contentLength != null) {
amqpMessageProperties.setContentLength(contentLength);
}
String contentType = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_TYPE, String.class);
if (StringUtils.hasText(contentType)) {
amqpMessageProperties.setContentType(contentType);
}
Object correlationId = headers.get(AmqpHeaders.CORRELATION_ID);
if (correlationId instanceof byte[]) {
amqpMessageProperties.setCorrelationId((byte[]) correlationId);
}
MessageDeliveryMode deliveryMode = getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.class);
if (deliveryMode != null) {
amqpMessageProperties.setDeliveryMode(deliveryMode);
}
Long deliveryTag = getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_TAG, Long.class);
if (deliveryTag != null) {
amqpMessageProperties.setDeliveryTag(deliveryTag);
}
String expiration = getHeaderIfAvailable(headers, AmqpHeaders.EXPIRATION, String.class);
if (StringUtils.hasText(expiration)) {
amqpMessageProperties.setExpiration(expiration);
}
Integer messageCount = getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_COUNT, Integer.class);
if (messageCount != null) {
amqpMessageProperties.setMessageCount(messageCount);
}
String messageId = getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_ID, String.class);
if (StringUtils.hasText(messageId)) {
amqpMessageProperties.setMessageId(messageId);
}
Integer priority = headers.getPriority();
if (priority != null) {
amqpMessageProperties.setPriority(priority);
}
String receivedExchange = getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_EXCHANGE, String.class);
if (StringUtils.hasText(receivedExchange)) {
amqpMessageProperties.setReceivedExchange(receivedExchange);
}
String receivedRoutingKey = getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_ROUTING_KEY, String.class);
if (StringUtils.hasText(receivedRoutingKey)) {
amqpMessageProperties.setReceivedRoutingKey(receivedRoutingKey);
}
Boolean redelivered = getHeaderIfAvailable(headers, AmqpHeaders.REDELIVERED, Boolean.class);
if (redelivered != null) {
amqpMessageProperties.setRedelivered(redelivered);
}
String replyTo = getHeaderIfAvailable(headers, AmqpHeaders.REPLY_TO, String.class);
if (replyTo != null) {
amqpMessageProperties.setReplyTo(replyTo);
}
Date timestamp = getHeaderIfAvailable(headers, AmqpHeaders.TIMESTAMP, Date.class);
if (timestamp != null) {
amqpMessageProperties.setTimestamp(timestamp);
}
String type = getHeaderIfAvailable(headers, AmqpHeaders.TYPE, String.class);
if (type != null) {
amqpMessageProperties.setType(type);
}
String userId = getHeaderIfAvailable(headers, AmqpHeaders.USER_ID, String.class);
if (StringUtils.hasText(userId)) {
amqpMessageProperties.setUserId(userId);
}
// now map to the user-defined headers, if any, within the AMQP MessageProperties
Set<String> headerNames = headers.keySet();
for (String headerName : headerNames) {
if (this.shouldMapOutboundHeader(headerName)) {
Object value = headers.get(headerName);
if (value != null) {
try {
String key = this.fromHeaderName(headerName);
// do not overwrite an existing header with the same key
// TODO: do we need to expose a boolean 'overwrite' flag?
if (!amqpMessageProperties.getHeaders().containsKey(key)) {
amqpMessageProperties.setHeader(key, value);
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("failed to map Message header '" + headerName + "' to AMQP header", e);
}
}
}
}
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("error occurred while mapping from MessageHeaders to AMQP properties", e);
}
}
}
/**
* Maps headers from an AMQP MessageProperties instance to the MessageHeaders of a
* Spring Integration Message.
*/
public Map<String, Object> toHeaders(MessageProperties amqpMessageProperties) {
@Override
protected Map<String, Object> extractStandardHeaders(MessageProperties amqpMessageProperties) {
Map<String, Object> headers = new HashMap<String, Object>();
try {
String appId = amqpMessageProperties.getAppId();
@@ -285,23 +152,6 @@ public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper {
if (StringUtils.hasText(userId)) {
headers.put(AmqpHeaders.USER_ID, userId);
}
Map<String, Object> amqpHeaders = amqpMessageProperties.getHeaders();
if (!CollectionUtils.isEmpty(amqpHeaders)) {
for (Map.Entry<String, Object> entry : amqpHeaders.entrySet()) {
try {
String headerName = this.toHeaderName(entry.getKey());
if (!ObjectUtils.containsElement(TRANSIENT_HEADER_NAMES, headerName)) {
headers.put(headerName, entry.getValue());
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("error occurred while mapping AMQP header '"
+ entry.getKey() + "' to Message header", e);
}
}
}
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
@@ -311,44 +161,123 @@ public class DefaultAmqpHeaderMapper implements AmqpHeaderMapper {
return headers;
}
private boolean shouldMapOutboundHeader(String headerName) {
return StringUtils.hasText(headerName)
&& !headerName.startsWith(AmqpHeaders.PREFIX)
&& !ObjectUtils.containsElement(TRANSIENT_HEADER_NAMES, headerName);
}
private <T> T getHeaderIfAvailable(MessageHeaders headers, String name, Class<T> type) {
try {
return headers.get(name, type);
}
catch (IllegalArgumentException e) {
if (logger.isWarnEnabled()) {
logger.warn("skipping header '" + name + "' since it is not of expected type [" + type + "]", e);
}
return null;
}
/**
* Extract user-defined headers from an AMQP MessageProperties instance.
*/
@Override
protected Map<String, Object> extractUserDefinedHeaders(MessageProperties amqpMessageProperties) {
return amqpMessageProperties.getHeaders();
}
/**
* Adds the outbound prefix if necessary.
* Maps headers from a Spring Integration MessageHeaders instance to the MessageProperties
* of an AMQP Message.
*/
private String fromHeaderName(String headerName) {
String propertyName = headerName;
if (StringUtils.hasText(this.outboundPrefix) && !propertyName.startsWith(this.outboundPrefix)) {
propertyName = this.outboundPrefix + headerName;
@Override
protected void populateStandardHeaders(Map<String, Object> headers, MessageProperties amqpMessageProperties) {
String appId = getHeaderIfAvailable(headers, AmqpHeaders.APP_ID, String.class);
if (StringUtils.hasText(appId)) {
amqpMessageProperties.setAppId(appId);
}
String clusterId = getHeaderIfAvailable(headers, AmqpHeaders.CLUSTER_ID, String.class);
if (StringUtils.hasText(clusterId)) {
amqpMessageProperties.setClusterId(clusterId);
}
String contentEncoding = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_ENCODING, String.class);
if (StringUtils.hasText(contentEncoding)) {
amqpMessageProperties.setContentEncoding(contentEncoding);
}
Long contentLength = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_LENGTH, Long.class);
if (contentLength != null) {
amqpMessageProperties.setContentLength(contentLength);
}
String contentType = getHeaderIfAvailable(headers, AmqpHeaders.CONTENT_TYPE, String.class);
if (StringUtils.hasText(contentType)) {
amqpMessageProperties.setContentType(contentType);
}
Object correlationId = headers.get(AmqpHeaders.CORRELATION_ID);
if (correlationId instanceof byte[]) {
amqpMessageProperties.setCorrelationId((byte[]) correlationId);
}
MessageDeliveryMode deliveryMode = getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.class);
if (deliveryMode != null) {
amqpMessageProperties.setDeliveryMode(deliveryMode);
}
Long deliveryTag = getHeaderIfAvailable(headers, AmqpHeaders.DELIVERY_TAG, Long.class);
if (deliveryTag != null) {
amqpMessageProperties.setDeliveryTag(deliveryTag);
}
String expiration = getHeaderIfAvailable(headers, AmqpHeaders.EXPIRATION, String.class);
if (StringUtils.hasText(expiration)) {
amqpMessageProperties.setExpiration(expiration);
}
Integer messageCount = getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_COUNT, Integer.class);
if (messageCount != null) {
amqpMessageProperties.setMessageCount(messageCount);
}
String messageId = getHeaderIfAvailable(headers, AmqpHeaders.MESSAGE_ID, String.class);
if (StringUtils.hasText(messageId)) {
amqpMessageProperties.setMessageId(messageId);
}
Integer priority = getHeaderIfAvailable(headers, MessageHeaders.PRIORITY, Integer.class);
if (priority != null) {
amqpMessageProperties.setPriority(priority);
}
String receivedExchange = getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_EXCHANGE, String.class);
if (StringUtils.hasText(receivedExchange)) {
amqpMessageProperties.setReceivedExchange(receivedExchange);
}
String receivedRoutingKey = getHeaderIfAvailable(headers, AmqpHeaders.RECEIVED_ROUTING_KEY, String.class);
if (StringUtils.hasText(receivedRoutingKey)) {
amqpMessageProperties.setReceivedRoutingKey(receivedRoutingKey);
}
Boolean redelivered = getHeaderIfAvailable(headers, AmqpHeaders.REDELIVERED, Boolean.class);
if (redelivered != null) {
amqpMessageProperties.setRedelivered(redelivered);
}
String replyTo = getHeaderIfAvailable(headers, AmqpHeaders.REPLY_TO, String.class);
if (replyTo != null) {
amqpMessageProperties.setReplyTo(replyTo);
}
Date timestamp = getHeaderIfAvailable(headers, AmqpHeaders.TIMESTAMP, Date.class);
if (timestamp != null) {
amqpMessageProperties.setTimestamp(timestamp);
}
String type = getHeaderIfAvailable(headers, AmqpHeaders.TYPE, String.class);
if (type != null) {
amqpMessageProperties.setType(type);
}
String userId = getHeaderIfAvailable(headers, AmqpHeaders.USER_ID, String.class);
if (StringUtils.hasText(userId)) {
amqpMessageProperties.setUserId(userId);
}
return propertyName;
}
/**
* Adds the inbound prefix if necessary.
*/
private String toHeaderName(String propertyName) {
String headerName = propertyName;
if (StringUtils.hasText(this.inboundPrefix) && !headerName.startsWith(this.inboundPrefix)) {
headerName = this.inboundPrefix + propertyName;
@Override
protected void populateUserDefinedHeader(String headerName, Object headerValue, MessageProperties amqpMessageProperties) {
// do not overwrite an existing header with the same key
// TODO: do we need to expose a boolean 'overwrite' flag?
if (!amqpMessageProperties.getHeaders().containsKey(headerName)) {
amqpMessageProperties.setHeader(headerName, headerValue);
}
return headerName;
}
@Override
protected List<String> getStandardRequestHeaderNames() {
return STANDARD_HEADER_NAMES;
}
@Override
protected List<String> getStandardReplyHeaderNames() {
return STANDARD_HEADER_NAMES;
}
@Override
protected String getStandardHeaderPrefix() {
return AmqpHeaders.PREFIX;
}
}

View File

@@ -85,6 +85,24 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
@@ -201,6 +219,35 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-reply-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of MessageHeaders to be mapped into the AMQP Message Properties of the AMQP reply message.
All standard Headers (e.g., contentType) will be mapped to AMQP Message Properties while user-defined headers will be mapped to 'headers' property
which itself is a Map.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
@@ -238,6 +285,17 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-reply-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of MessageHeaders to be mapped into the AMQP Message Properties of the AMQP reply message.
All standard Headers (e.g., contentType) will be mapped to AMQP Message Properties while user-defined headers will be mapped to 'headers' property
which itself is a Map.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>
@@ -380,6 +438,24 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper" type="xsd:string">
<xsd:annotation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of AMQP Headers to be mapped from the AMQP request into the MessageHeaders.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="message-converter" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
@@ -392,18 +468,6 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper" type="xsd:string">
<xsd:annotation>
<xsd:documentation>
HeaderMapper to use when receiving AMQP Messages.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.amqp.support.AmqpHeaderMapper"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="listener-container" type="xsd:string">
<xsd:annotation>
<xsd:documentation>

View File

@@ -13,6 +13,25 @@
<amqp:inbound-channel-adapter id="autoStartFalse" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"/>
<amqp:inbound-channel-adapter id="withHeaderMapperStandardAndCustomHeaders" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"
mapped-request-headers="foo*, STANDARD_REQUEST_HEADERS"/>
<amqp:inbound-channel-adapter id="withHeaderMapperOnlyCustomHeaders" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"
mapped-request-headers="foo*"/>
<amqp:inbound-channel-adapter id="withHeaderMapperNothingToMap" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"
mapped-request-headers=""/>
<amqp:inbound-channel-adapter id="withHeaderMapperDefaultMapping" channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"/>
<int:channel id="requestChannel">
<int:queue/>
</int:channel>
<bean id="rabbitConnectionFactory" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>

View File

@@ -16,19 +16,26 @@
package org.springframework.integration.amqp.config;
import static org.junit.Assert.assertEquals;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.inbound.AmqpInboundChannelAdapter;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
/**
* @author Mark Fisher
* @since 2.1
@@ -56,4 +63,113 @@ public class AmqpInboundChannelAdapterParserTests {
assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(adapter, "autoStartup"));
assertEquals(123, TestUtils.getPropertyValue(adapter, "phase"));
}
@Test
public void withHeaderMapperStandardAndCustomHeaders() {
AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperStandardAndCustomHeaders", AmqpInboundChannelAdapter.class);
AbstractMessageListenerContainer mlc =
TestUtils.getPropertyValue(adapter, "messageListenerContainer", AbstractMessageListenerContainer.class);
MessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", MessageListener.class);
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setClusterId("test.clusterId");
amqpProperties.setContentEncoding("test.contentEncoding");
amqpProperties.setContentLength(99L);
amqpProperties.setContentType("test.contentType");
amqpProperties.setHeader("foo", "foo");
amqpProperties.setHeader("bar", "bar");
Message amqpMessage = new Message("hello".getBytes(), amqpProperties);
listener.onMessage(amqpMessage);
QueueChannel requestChannel = context.getBean("requestChannel", QueueChannel.class);
org.springframework.integration.Message<?> siMessage = requestChannel.receive(0);
assertEquals("foo", siMessage.getHeaders().get("foo"));
assertNull(siMessage.getHeaders().get("bar"));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.CONTENT_ENCODING));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.CLUSTER_ID));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.APP_ID));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
}
@Test
public void withHeaderMapperOnlyCustomHeaders() {
AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperOnlyCustomHeaders", AmqpInboundChannelAdapter.class);
AbstractMessageListenerContainer mlc =
TestUtils.getPropertyValue(adapter, "messageListenerContainer", AbstractMessageListenerContainer.class);
MessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", MessageListener.class);
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setClusterId("test.clusterId");
amqpProperties.setContentEncoding("test.contentEncoding");
amqpProperties.setContentLength(99L);
amqpProperties.setContentType("test.contentType");
amqpProperties.setHeader("foo", "foo");
amqpProperties.setHeader("bar", "bar");
Message amqpMessage = new Message("hello".getBytes(), amqpProperties);
listener.onMessage(amqpMessage);
QueueChannel requestChannel = context.getBean("requestChannel", QueueChannel.class);
org.springframework.integration.Message<?> siMessage = requestChannel.receive(0);
assertEquals("foo", siMessage.getHeaders().get("foo"));
assertNull(siMessage.getHeaders().get("bar"));
assertNull(siMessage.getHeaders().get(AmqpHeaders.CONTENT_ENCODING));
assertNull(siMessage.getHeaders().get(AmqpHeaders.CLUSTER_ID));
assertNull(siMessage.getHeaders().get(AmqpHeaders.APP_ID));
assertNull(siMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
}
@Test
public void withHeaderMapperNothingToMap() {
AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperNothingToMap", AmqpInboundChannelAdapter.class);
AbstractMessageListenerContainer mlc =
TestUtils.getPropertyValue(adapter, "messageListenerContainer", AbstractMessageListenerContainer.class);
MessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", MessageListener.class);
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setClusterId("test.clusterId");
amqpProperties.setContentEncoding("test.contentEncoding");
amqpProperties.setContentLength(99L);
amqpProperties.setContentType("test.contentType");
amqpProperties.setHeader("foo", "foo");
amqpProperties.setHeader("bar", "bar");
Message amqpMessage = new Message("hello".getBytes(), amqpProperties);
listener.onMessage(amqpMessage);
QueueChannel requestChannel = context.getBean("requestChannel", QueueChannel.class);
org.springframework.integration.Message<?> siMessage = requestChannel.receive(0);
assertNull(siMessage.getHeaders().get("foo"));
assertNull(siMessage.getHeaders().get("bar"));
assertNull(siMessage.getHeaders().get(AmqpHeaders.CONTENT_ENCODING));
assertNull(siMessage.getHeaders().get(AmqpHeaders.CLUSTER_ID));
assertNull(siMessage.getHeaders().get(AmqpHeaders.APP_ID));
assertNull(siMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
}
@Test
public void withHeaderMapperDefaultMapping() {
AmqpInboundChannelAdapter adapter = context.getBean("withHeaderMapperDefaultMapping", AmqpInboundChannelAdapter.class);
AbstractMessageListenerContainer mlc =
TestUtils.getPropertyValue(adapter, "messageListenerContainer", AbstractMessageListenerContainer.class);
MessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", MessageListener.class);
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setClusterId("test.clusterId");
amqpProperties.setContentEncoding("test.contentEncoding");
amqpProperties.setContentLength(99L);
amqpProperties.setContentType("test.contentType");
amqpProperties.setHeader("foo", "foo");
amqpProperties.setHeader("bar", "bar");
Message amqpMessage = new Message("hello".getBytes(), amqpProperties);
listener.onMessage(amqpMessage);
QueueChannel requestChannel = context.getBean("requestChannel", QueueChannel.class);
org.springframework.integration.Message<?> siMessage = requestChannel.receive(0);
assertNull(siMessage.getHeaders().get("bar"));
assertNull(siMessage.getHeaders().get("foo"));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.CONTENT_ENCODING));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.CLUSTER_ID));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.APP_ID));
assertNotNull(siMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
}
}

View File

@@ -23,5 +23,12 @@
<si-amqp:inbound-gateway id="autoStartFalseGateway" request-channel="requests" queue-names="test"
connection-factory="rabbitConnectionFactory" message-converter="testConverter"
auto-startup="false" phase="123"/>
<si-amqp:inbound-gateway id="withHeaderMapper" request-channel="requestChannel" queue-names="inboundchanneladapter.test.2"
auto-startup="false" phase="123"
mapped-request-headers="foo*, STANDARD_REQUEST_HEADERS"
mapped-reply-headers="bar*"/>
<int:channel id="requestChannel"/>
</beans>

View File

@@ -16,19 +16,36 @@
package org.springframework.integration.amqp.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import java.lang.reflect.Field;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessagingException;
import org.springframework.integration.amqp.inbound.AmqpInboundGateway;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ReflectionUtils;
import static junit.framework.Assert.assertEquals;
import static org.junit.Assert.assertSame;
/**
* @author Mark Fisher
@@ -59,6 +76,55 @@ public class AmqpInboundGatewayParserTests {
assertEquals(Boolean.FALSE, TestUtils.getPropertyValue(gateway, "autoStartup"));
assertEquals(123, TestUtils.getPropertyValue(gateway, "phase"));
}
@SuppressWarnings("rawtypes")
@Test
public void verifyUsageWithHeaderMapper() throws Exception{
DirectChannel requestChannel = context.getBean("requestChannel", DirectChannel.class);
requestChannel.subscribe(new MessageHandler() {
public void handleMessage(org.springframework.integration.Message<?> siMessage)
throws MessagingException {
org.springframework.integration.Message<?> replyMessage = MessageBuilder.fromMessage(siMessage).setHeader("bar", "bar").build();
MessageChannel replyChannel = (MessageChannel) siMessage.getHeaders().getReplyChannel();
replyChannel.send(replyMessage);
}
});
final AmqpInboundGateway gateway = context.getBean("withHeaderMapper", AmqpInboundGateway.class);
Field amqpTemplateField = ReflectionUtils.findField(AmqpInboundGateway.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(gateway, "amqpTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
Message amqpReplyMessage = (Message) args[2];
MessageProperties properties = amqpReplyMessage.getMessageProperties();
assertEquals("bar", properties.getHeaders().get("bar"));
return null;
}})
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(Message.class));
ReflectionUtils.setField(amqpTemplateField, gateway, amqpTemplate);
AbstractMessageListenerContainer mlc =
TestUtils.getPropertyValue(gateway, "messageListenerContainer", AbstractMessageListenerContainer.class);
MessageListener listener = TestUtils.getPropertyValue(mlc, "messageListener", MessageListener.class);
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setClusterId("test.clusterId");
amqpProperties.setContentEncoding("test.contentEncoding");
amqpProperties.setContentLength(99L);
amqpProperties.setReplyTo("oleg");
amqpProperties.setContentType("test.contentType");
amqpProperties.setHeader("foo", "foo");
amqpProperties.setHeader("bar", "bar");
Message amqpMessage = new Message("hello".getBytes(), amqpProperties);
listener.onMessage(amqpMessage);
Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(Message.class));
}
private static class TestConverter extends SimpleMessageConverter {}

View File

@@ -16,5 +16,11 @@
<bean id="connectionFactory" class="org.mockito.Mockito" factory-method="mock">
<constructor-arg value="org.springframework.amqp.rabbit.connection.ConnectionFactory"/>
</bean>
<amqp:outbound-channel-adapter id="withHeaderMapperCustomHeaders" channel="requestChannel"
exchange-name="outboundchanneladapter.test.1"
mapped-request-headers="foo*"/>
<int:channel id="requestChannel"/>
</beans>

View File

@@ -16,23 +16,36 @@
package org.springframework.integration.amqp.config;
import static org.junit.Assert.assertEquals;
import java.lang.reflect.Field;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageHandler;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ReflectionUtils;
import static org.junit.Assert.assertNull;
import static junit.framework.Assert.assertEquals;
/**
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.1
*/
@ContextConfiguration
@@ -51,5 +64,37 @@ public class AmqpOutboundChannelAdapterParserTests {
MessageHandler handler = TestUtils.getPropertyValue(adapter, "handler", MessageHandler.class);
assertEquals(AmqpOutboundEndpoint.class, handler.getClass());
}
@SuppressWarnings("rawtypes")
@Test
public void withHeaderMapperCustomHeaders() {
Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomHeaders");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpReplyMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpReplyMessage.getMessageProperties();
assertEquals("foo", properties.getHeaders().get("foo"));
assertEquals("foobar", properties.getHeaders().get("foobar"));
assertNull(properties.getHeaders().get("bar"));
return null;
}})
.when(amqpTemplate).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("requestChannel", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").setHeader("bar", "bar").setHeader("foobar", "foobar").build();
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).send(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
}
}

View File

@@ -23,6 +23,35 @@
</bean>
<int:channel id="toRabbit"/>
<int:channel id="fromRabbit"/>
<int:channel id="fromRabbit">
<int:queue/>
</int:channel>
<amqp:outbound-gateway id="withHeaderMapperCustomRequestResponse" request-channel="toRabbit"
reply-channel="fromRabbit"
exchange-name="si.test.exchange"
routing-key="si.test.binding"
amqp-template="amqpTemplate"
order="5"
mapped-request-headers="foo*"
mapped-reply-headers="bar*"/>
<amqp:outbound-gateway id="withHeaderMapperCustomAndStandardResponse" request-channel="toRabbit"
reply-channel="fromRabbit"
exchange-name="si.test.exchange"
routing-key="si.test.binding"
amqp-template="amqpTemplate"
order="5"
mapped-request-headers="foo*"
mapped-reply-headers="bar*, STANDARD_REPLY_HEADERS"/>
<amqp:outbound-gateway id="withHeaderMapperNothingToMap" request-channel="toRabbit"
reply-channel="fromRabbit"
exchange-name="si.test.exchange"
routing-key="si.test.binding"
amqp-template="amqpTemplate"
order="5"
mapped-request-headers=""
mapped-reply-headers=""/>
</beans>

View File

@@ -15,15 +15,30 @@
*/
package org.springframework.integration.amqp.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.lang.reflect.Field;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.amqp.AmqpHeaders;
import org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.util.ReflectionUtils;
import static junit.framework.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
/**
* @author Oleg Zhurakousky
@@ -34,9 +49,151 @@ public class AmqpOutboundGatewayParserTests {
@Test
public void testGatewayConfig(){
ApplicationContext context = new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-context.xml", this.getClass());
AmqpOutboundEndpoint gateway = context.getBean(AmqpOutboundEndpoint.class);
Object edc = context.getBean("rabbitGateway");
AmqpOutboundEndpoint gateway = TestUtils.getPropertyValue(edc, "handler", AmqpOutboundEndpoint.class);
assertEquals(5, gateway.getOrder());
assertTrue(context.containsBean("rabbitGateway"));
assertEquals(context.getBean("fromRabbit"), TestUtils.getPropertyValue(gateway, "outputChannel"));
}
@SuppressWarnings("rawtypes")
@Test
public void withHeaderMapperCustomRequestResponse() {
ApplicationContext context = new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomRequestResponse");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
assertEquals("foo", properties.getHeaders().get("foo"));
// mock reply AMQP message
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
org.springframework.amqp.core.Message amqpReplyMessage = new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
return amqpReplyMessage;
}})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("toRabbit", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build();
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
// verify reply
QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class);
Message<?> replyMessage = queueChannel.receive(0);
assertEquals("bar", replyMessage.getHeaders().get("bar"));
assertEquals("foo", replyMessage.getHeaders().get("foo")); // copied from request Message
assertNull(replyMessage.getHeaders().get("foobar"));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID));
}
@SuppressWarnings("rawtypes")
@Test
public void withHeaderMapperCustomAndStandardResponse() {
ApplicationContext context = new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object eventDrivernConsumer = context.getBean("withHeaderMapperCustomAndStandardResponse");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
assertEquals("foo", properties.getHeaders().get("foo"));
// mock reply AMQP message
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
org.springframework.amqp.core.Message amqpReplyMessage = new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
return amqpReplyMessage;
}})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("toRabbit", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build();
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
// verify reply
QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class);
Message<?> replyMessage = queueChannel.receive(0);
assertEquals("bar", replyMessage.getHeaders().get("bar"));
assertEquals("foo", replyMessage.getHeaders().get("foo")); // copied from request Message
assertNull(replyMessage.getHeaders().get("foobar"));
assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE));
assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
assertNotNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID));
}
@SuppressWarnings("rawtypes")
@Test
public void withHeaderMapperNothingToMap() {
ApplicationContext context = new ClassPathXmlApplicationContext("AmqpOutboundGatewayParserTests-context.xml", this.getClass());
Object eventDrivernConsumer = context.getBean("withHeaderMapperNothingToMap");
AmqpOutboundEndpoint endpoint = TestUtils.getPropertyValue(eventDrivernConsumer, "handler", AmqpOutboundEndpoint.class);
Field amqpTemplateField = ReflectionUtils.findField(AmqpOutboundEndpoint.class, "amqpTemplate");
amqpTemplateField.setAccessible(true);
RabbitTemplate amqpTemplate = TestUtils.getPropertyValue(endpoint, "amqpTemplate", RabbitTemplate.class);
amqpTemplate = Mockito.spy(amqpTemplate);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.springframework.amqp.core.Message amqpRequestMessage = (org.springframework.amqp.core.Message) args[2];
MessageProperties properties = amqpRequestMessage.getMessageProperties();
assertNull(properties.getHeaders().get("foo"));
// mock reply AMQP message
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setHeader("foobar", "foobar");
amqpProperties.setHeader("bar", "bar");
org.springframework.amqp.core.Message amqpReplyMessage = new org.springframework.amqp.core.Message("hello".getBytes(), amqpProperties);
return amqpReplyMessage;
}})
.when(amqpTemplate).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
ReflectionUtils.setField(amqpTemplateField, endpoint, amqpTemplate);
MessageChannel requestChannel = context.getBean("toRabbit", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader("foo", "foo").build();
requestChannel.send(message);
Mockito.verify(amqpTemplate, Mockito.times(1)).sendAndReceive(Mockito.any(String.class), Mockito.any(String.class), Mockito.any(org.springframework.amqp.core.Message.class));
// verify reply
QueueChannel queueChannel = context.getBean("fromRabbit", QueueChannel.class);
Message<?> replyMessage = queueChannel.receive(0);
assertNull(replyMessage.getHeaders().get("bar"));
assertEquals("foo", replyMessage.getHeaders().get("foo")); // copied from request Message
assertNull(replyMessage.getHeaders().get("foobar"));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.DELIVERY_MODE));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.CONTENT_TYPE));
assertNull(replyMessage.getHeaders().get(AmqpHeaders.APP_ID));
}
}

View File

@@ -18,15 +18,17 @@ package org.springframework.integration.amqp.support;
import static org.junit.Assert.assertEquals;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import org.junit.Test;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.JsonMessageConverter;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.amqp.AmqpHeaders;
/**
* @author Mark Fisher
@@ -34,6 +36,96 @@ import org.springframework.integration.MessageHeaders;
*/
public class DefaultAmqpHeaderMapperTests {
@Test
public void fromHeaders() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
headerMap.put(AmqpHeaders.APP_ID, "test.appId");
headerMap.put(AmqpHeaders.CLUSTER_ID, "test.clusterId");
headerMap.put(AmqpHeaders.CONTENT_ENCODING, "test.contentEncoding");
headerMap.put(AmqpHeaders.CONTENT_LENGTH, 99L);
headerMap.put(AmqpHeaders.CONTENT_TYPE, "test.contentType");
byte[] testCorrelationId = new byte[] {1,2,3};
headerMap.put(AmqpHeaders.CORRELATION_ID, testCorrelationId);
headerMap.put(AmqpHeaders.DELIVERY_MODE, MessageDeliveryMode.NON_PERSISTENT);
headerMap.put(AmqpHeaders.DELIVERY_TAG, 1234L);
headerMap.put(AmqpHeaders.EXPIRATION, "test.expiration");
headerMap.put(AmqpHeaders.MESSAGE_COUNT, 42);
headerMap.put(AmqpHeaders.MESSAGE_ID, "test.messageId");
headerMap.put(AmqpHeaders.RECEIVED_EXCHANGE, "test.receivedExchange");
headerMap.put(AmqpHeaders.RECEIVED_ROUTING_KEY, "test.receivedRoutingKey");
headerMap.put(AmqpHeaders.REPLY_TO, "test.replyTo");
Date testTimestamp = new Date();
headerMap.put(AmqpHeaders.TIMESTAMP, testTimestamp);
headerMap.put(AmqpHeaders.TYPE, "test.type");
headerMap.put(AmqpHeaders.USER_ID, "test.userId");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
MessageProperties amqpProperties = new MessageProperties();
headerMapper.fromHeadersToRequest(integrationHeaders, amqpProperties);
assertEquals("test.appId", amqpProperties.getAppId());
assertEquals("test.clusterId", amqpProperties.getClusterId());
assertEquals("test.contentEncoding", amqpProperties.getContentEncoding());
assertEquals(99L, amqpProperties.getContentLength());
assertEquals("test.contentType", amqpProperties.getContentType());
assertEquals(testCorrelationId, amqpProperties.getCorrelationId());
assertEquals(MessageDeliveryMode.NON_PERSISTENT, amqpProperties.getDeliveryMode());
assertEquals(1234L, amqpProperties.getDeliveryTag());
assertEquals("test.expiration", amqpProperties.getExpiration());
assertEquals(new Integer(42), amqpProperties.getMessageCount());
assertEquals("test.messageId", amqpProperties.getMessageId());
assertEquals("test.receivedExchange", amqpProperties.getReceivedExchange());
assertEquals("test.receivedRoutingKey", amqpProperties.getReceivedRoutingKey());
assertEquals("test.replyTo", amqpProperties.getReplyTo());
assertEquals(testTimestamp, amqpProperties.getTimestamp());
assertEquals("test.type", amqpProperties.getType());
assertEquals("test.userId", amqpProperties.getUserId());
}
@Test
public void toHeaders() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
MessageProperties amqpProperties = new MessageProperties();
amqpProperties.setAppId("test.appId");
amqpProperties.setClusterId("test.clusterId");
amqpProperties.setContentEncoding("test.contentEncoding");
amqpProperties.setContentLength(99L);
amqpProperties.setContentType("test.contentType");
byte[] testCorrelationId = new byte[] {1,2,3};
amqpProperties.setCorrelationId(testCorrelationId);
amqpProperties.setDeliveryMode(MessageDeliveryMode.NON_PERSISTENT);
amqpProperties.setDeliveryTag(1234L);
amqpProperties.setExpiration("test.expiration");
amqpProperties.setMessageCount(42);
amqpProperties.setMessageId("test.messageId");
amqpProperties.setPriority(22);
amqpProperties.setReceivedExchange("test.receivedExchange");
amqpProperties.setReceivedRoutingKey("test.receivedRoutingKey");
amqpProperties.setRedelivered(true);
amqpProperties.setReplyTo("test.replyTo");
Date testTimestamp = new Date();
amqpProperties.setTimestamp(testTimestamp);
amqpProperties.setType("test.type");
amqpProperties.setUserId("test.userId");
Map<String, Object> headerMap = headerMapper.toHeadersFromReply(amqpProperties);
assertEquals("test.appId", headerMap.get(AmqpHeaders.APP_ID));
assertEquals("test.clusterId", headerMap.get(AmqpHeaders.CLUSTER_ID));
assertEquals("test.contentEncoding", headerMap.get(AmqpHeaders.CONTENT_ENCODING));
assertEquals(99L, headerMap.get(AmqpHeaders.CONTENT_LENGTH));
assertEquals("test.contentType", headerMap.get(AmqpHeaders.CONTENT_TYPE));
assertEquals(testCorrelationId, headerMap.get(AmqpHeaders.CORRELATION_ID));
assertEquals(MessageDeliveryMode.NON_PERSISTENT, headerMap.get(AmqpHeaders.DELIVERY_MODE));
assertEquals(1234L, headerMap.get(AmqpHeaders.DELIVERY_TAG));
assertEquals("test.expiration", headerMap.get(AmqpHeaders.EXPIRATION));
assertEquals(new Integer(42), headerMap.get(AmqpHeaders.MESSAGE_COUNT));
assertEquals("test.messageId", headerMap.get(AmqpHeaders.MESSAGE_ID));
assertEquals("test.receivedExchange", headerMap.get(AmqpHeaders.RECEIVED_EXCHANGE));
assertEquals("test.receivedRoutingKey", headerMap.get(AmqpHeaders.RECEIVED_ROUTING_KEY));
assertEquals("test.replyTo", headerMap.get(AmqpHeaders.REPLY_TO));
assertEquals(testTimestamp, headerMap.get(AmqpHeaders.TIMESTAMP));
assertEquals("test.type", headerMap.get(AmqpHeaders.TYPE));
assertEquals("test.userId", headerMap.get(AmqpHeaders.USER_ID));
}
@Test
public void replyChannelNotMappedToAmqpProperties() {
DefaultAmqpHeaderMapper headerMapper = new DefaultAmqpHeaderMapper();
@@ -41,7 +133,7 @@ public class DefaultAmqpHeaderMapperTests {
headerMap.put(MessageHeaders.REPLY_CHANNEL, "foo");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
MessageProperties amqpProperties = new MessageProperties();
headerMapper.fromHeaders(integrationHeaders, amqpProperties);
headerMapper.fromHeadersToRequest(integrationHeaders, amqpProperties);
assertEquals(null, amqpProperties.getHeaders().get(MessageHeaders.REPLY_CHANNEL));
}
@@ -52,7 +144,7 @@ public class DefaultAmqpHeaderMapperTests {
headerMap.put(MessageHeaders.ERROR_CHANNEL, "foo");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
MessageProperties amqpProperties = new MessageProperties();
headerMapper.fromHeaders(integrationHeaders, amqpProperties);
headerMapper.fromHeadersToRequest(integrationHeaders, amqpProperties);
assertEquals(null, amqpProperties.getHeaders().get(MessageHeaders.ERROR_CHANNEL));
}
@@ -65,7 +157,7 @@ public class DefaultAmqpHeaderMapperTests {
Map<String, Object> headerMap = new HashMap<String, Object>();
headerMap.put("__TypeId__", "java.lang.Integer");
MessageHeaders integrationHeaders = new MessageHeaders(headerMap);
headerMapper.fromHeaders(integrationHeaders, amqpProperties);
headerMapper.fromHeadersToRequest(integrationHeaders, amqpProperties);
assertEquals("java.lang.String", amqpProperties.getHeaders().get("__TypeId__"));
Object result = converter.fromMessage(new Message("123".getBytes(), amqpProperties));
assertEquals(String.class, result.getClass());

View File

@@ -26,7 +26,6 @@ import org.springframework.core.Conventions;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
import org.w3c.dom.Element;
/**
@@ -220,4 +219,37 @@ public abstract class IntegrationNamespaceUtils {
return innerComponentDefinition;
}
/**
* Utility method to configure HeaderMapper for Inbound and Outbound channel adapters/gateway
*/
public static void configureHeaderMapper(Element element, BeanDefinitionBuilder rootBuilder, ParserContext parserContext, Class<?> headerMapperClass, String replyHeaderValue){
String defaultMappedReplyHeadersAttributeName = "mapped-reply-headers";
if (!StringUtils.hasText(replyHeaderValue)){
replyHeaderValue = defaultMappedReplyHeadersAttributeName;
}
boolean hasHeaderMapper = element.hasAttribute("header-mapper");
boolean hasMappedRequestHeaders = element.hasAttribute("mapped-request-headers");
boolean hasMappedReplyHeaders = element.hasAttribute(replyHeaderValue);
if (hasHeaderMapper && (hasMappedRequestHeaders || hasMappedReplyHeaders)){
parserContext.getReaderContext().error("The 'header-mapper' attribute is mutually exclusive with" +
" 'mapped-request-headers' or 'mapped-reply-headers'. " +
"You can only use one or the others", element);
}
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(rootBuilder, element, "header-mapper");
if (hasMappedRequestHeaders || hasMappedReplyHeaders){
BeanDefinitionBuilder headerMapperBuilder = BeanDefinitionBuilder.genericBeanDefinition(headerMapperClass);
if (hasMappedRequestHeaders) {
headerMapperBuilder.addPropertyValue("requestHeaderNames", element.getAttribute("mapped-request-headers"));
}
if (hasMappedReplyHeaders) {
headerMapperBuilder.addPropertyValue("replyHeaderNames", element.getAttribute(replyHeaderValue));
}
rootBuilder.addPropertyValue("headerMapper", headerMapperBuilder.getBeanDefinition());
}
}
}

View File

@@ -0,0 +1,302 @@
/*
* Copyright 2002-2011 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.mapping;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.integration.MessageHeaders;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.PatternMatchUtils;
import org.springframework.util.StringUtils;
/**
* Abstract base class for HeaderMapper implementations.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.1
*/
public abstract class AbstractHeaderMapper<T> implements RequestReplyHeaderMapper<T> {
public static final String STANDARD_REQUEST_HEADER_NAME_PATTERN = "STANDARD_REQUEST_HEADERS";
public static final String STANDARD_REPLY_HEADER_NAME_PATTERN = "STANDARD_REPLY_HEADERS";
private static final String[] TRANSIENT_HEADER_NAMES = new String[] {
MessageHeaders.ID,
MessageHeaders.ERROR_CHANNEL,
MessageHeaders.REPLY_CHANNEL,
MessageHeaders.TIMESTAMP
};
protected final Log logger = LogFactory.getLog(this.getClass());
private final String standardHeaderPrefix;
private volatile String userDefinedHeaderPrefix = "";
private volatile List<String> requestHeaderNames = new ArrayList<String>();
private volatile List<String> replyHeaderNames = new ArrayList<String>();
protected AbstractHeaderMapper() {
this.standardHeaderPrefix = this.getStandardHeaderPrefix();
this.requestHeaderNames.addAll(this.getStandardRequestHeaderNames());
this.replyHeaderNames.addAll(this.getStandardReplyHeaderNames());
}
/**
* Provide the header names that should be mapped from a request (for inbound/outbound adapters)
* TO a Spring Integration Message's headers.
* The values can also contain simple wildcard patterns (e.g. "foo*" or "*foo") to be matched.
* <p>
* This will match the header name directly or, for non-standard headers, it will match
* the header name prefixed with the value specified by {@link #setInboundPrefix(String)}.
*/
public void setRequestHeaderNames(String[] requestHeaderNames) {
Assert.notNull(requestHeaderNames, "'requestHeaderNames' must not be null");
this.requestHeaderNames = Arrays.asList(requestHeaderNames);
}
/**
* Provide the header names that should be mapped to a response (for inbound/outbound adapters)
* FROM a Spring Integration Message's headers.
* The values can also contain simple wildcard patterns (e.g. "foo*" or "*foo") to be matched.
* <p>
* Any non-standard headers will be prefixed with the value specified by {@link #setOutboundPrefix(String)}.
*/
public void setReplyHeaderNames(String[] replyHeaderNames) {
Assert.notNull(replyHeaderNames, "'replyHeaderNames' must not be null");
this.replyHeaderNames = Arrays.asList(replyHeaderNames);
}
/**
* Specify a prefix to be prepended to the header name for any integration
* message header that is being mapped to or from a user-defined value.
* <p/>
* This does not affect the standard properties for the particular protocol, such as
* contentType for AMQP, etc. The header names used for mapping such properties are
* defined in a corresponding Headers class as constants (e.g. AmqpHeaders).
*/
public void setUserDefinedHeaderPrefix(String userDefinedHeaderPrefix) {
this.userDefinedHeaderPrefix = (userDefinedHeaderPrefix != null) ? userDefinedHeaderPrefix : "";
}
/**
* Maps headers from a Spring Integration MessageHeaders instance to the target instance
* matching on the set of REQUEST headers (if different).
*/
public void fromHeadersToRequest(MessageHeaders headers, T target) {
this.fromHeaders(headers, target, this.requestHeaderNames);
}
/**
* Maps headers from a Spring Integration MessageHeaders instance to the target instance
* matching on the set of REPLY headers (if different).
*/
public void fromHeadersToReply(MessageHeaders headers, T target) {
this.fromHeaders(headers, target, this.replyHeaderNames);
}
/**
* Maps headers/properties of the target object to Map of MessageHeaders
* matching on the set of REQUEST headers
*/
public Map<String, Object> toHeadersFromRequest(T source) {
return this.toHeaders(source, this.requestHeaderNames);
}
/**
* Maps headers/properties of the target object to Map of MessageHeaders
* matching on the set of REPLY headers
*/
public Map<String, Object> toHeadersFromReply(T source) {
return this.toHeaders(source, this.replyHeaderNames);
}
private void fromHeaders(MessageHeaders headers, T target, List<String> headerPatterns){
try {
Map<String, Object> subset = new HashMap<String, Object>();
for (String headerName : headers.keySet()) {
if (this.shouldMapHeader(headerName, headerPatterns)){
subset.put(headerName, headers.get(headerName));
}
}
this.populateStandardHeaders(subset, target);
this.populateUserDefinedHeaders(subset, target);
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("error occurred while mapping from MessageHeaders", e);
}
}
}
private void populateUserDefinedHeaders(Map<String, Object> headers, T target) {
for (String headerName : headers.keySet()) {
Object value = headers.get(headerName);
if (value != null) {
try {
String key = this.addPrefixIfNecessary(this.userDefinedHeaderPrefix, headerName);
this.populateUserDefinedHeader(key, value, target);
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("failed to map from Message header '" + headerName + "' to target", e);
}
}
}
}
}
/**
* Maps headers from a source instance to the MessageHeaders of a
* Spring Integration Message.
*/
private Map<String, Object> toHeaders(T source, List<String> headerPatterns) {
Map<String, Object> headers = new HashMap<String, Object>();
Map<String, Object> standardHeaders = this.extractStandardHeaders(source);
this.copyHeaders(this.standardHeaderPrefix, standardHeaders, headers, headerPatterns);
Map<String, Object> userDefinedHeaders = this.extractUserDefinedHeaders(source);
this.copyHeaders(this.userDefinedHeaderPrefix, userDefinedHeaders, headers, headerPatterns);
return headers;
}
private <V> void copyHeaders(String prefix, Map<String, Object> source, Map<String, Object> target, List<String> headerPatterns) {
if (!CollectionUtils.isEmpty(source)) {
for (Map.Entry<String, Object> entry : source.entrySet()) {
try {
String headerName = this.addPrefixIfNecessary(prefix, entry.getKey());
if (this.shouldMapHeader(headerName, headerPatterns)){
target.put(headerName, entry.getValue());
}
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("error occurred while mapping header '"
+ entry.getKey() + "' to Message header", e);
}
}
}
}
}
private boolean shouldMapHeader(String headerName, List<String> patterns) {
if (!StringUtils.hasText(headerName)
|| ObjectUtils.containsElement(TRANSIENT_HEADER_NAMES, headerName)) {
return false;
}
if (patterns != null && patterns.size() > 0) {
for (String pattern : patterns) {
if (PatternMatchUtils.simpleMatch(pattern.toLowerCase(), headerName.toLowerCase())) {
if (logger.isDebugEnabled()) {
logger.debug(MessageFormat.format("headerName=[{0}] WILL be mapped, matched pattern={1}", headerName, pattern));
}
return true;
}
else if (STANDARD_REQUEST_HEADER_NAME_PATTERN.equals(pattern)
&& this.containsElementIgnoreCase(this.getStandardRequestHeaderNames(), headerName)) {
if (logger.isDebugEnabled()) {
logger.debug(MessageFormat.format("headerName=[{0}] WILL be mapped, matched pattern={1}", headerName, pattern));
}
return true;
}
else if (STANDARD_REPLY_HEADER_NAME_PATTERN.equals(pattern)
&& this.containsElementIgnoreCase(this.getStandardReplyHeaderNames(), headerName)) {
if (logger.isDebugEnabled()) {
logger.debug(MessageFormat.format("headerName=[{0}] WILL be mapped, matched pattern={1}", headerName, pattern));
}
return true;
}
}
}
if (logger.isDebugEnabled()) {
logger.debug(MessageFormat.format("headerName=[{0}] WILL NOT be mapped", headerName));
}
return false;
}
@SuppressWarnings("unchecked")
protected <V> V getHeaderIfAvailable(Map<String, Object> headers, String name, Class<V> type) {
Object value = headers.get(name);
if (value == null) {
return null;
}
if (!type.isAssignableFrom(value.getClass())) {
if (logger.isWarnEnabled()) {
logger.warn("skipping header '" + name + "' since it is not of expected type [" + type + "]");
}
}
return (V) value;
}
private boolean containsElementIgnoreCase(List<String> headerNames, String name) {
for (String headerName : headerNames) {
if (headerName.equalsIgnoreCase(name)){
return true;
}
}
return false;
}
/**
* Adds the prefix to the header name
*/
private String addPrefixIfNecessary(String prefix, String propertyName) {
String headerName = propertyName;
if (StringUtils.hasText(prefix) && !headerName.startsWith(prefix)) {
headerName = prefix + propertyName;
}
return headerName;
}
/**
* Returns the list of standard REQUEST headers. Implementation provided by a subclass
*/
protected List<String> getStandardReplyHeaderNames(){
return Collections.emptyList();
}
/**
* Returns the PREFIX used by standard headers (if any)
*/
protected List<String> getStandardRequestHeaderNames(){
return Collections.emptyList();
}
/**
* Returns the list of standard REPLY headers. Implementation provided by a subclass
*/
protected abstract String getStandardHeaderPrefix();
protected abstract Map<String, Object> extractStandardHeaders(T source);
protected abstract Map<String, Object> extractUserDefinedHeaders(T source);
protected abstract void populateStandardHeaders(Map<String, Object> headers, T target);
protected abstract void populateUserDefinedHeader(String headerName, Object headerValue, T target);
}

View File

@@ -0,0 +1,41 @@
/*
* Copyright 2002-2011 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.mapping;
import java.util.Map;
import org.springframework.integration.MessageHeaders;
/**
* Request/Reply strategy interface for mapping {@link MessageHeaders} to and from other
* types of objects. This would typically be used by adapters where the "other type"
* has a concept of headers or properties (HTTP, JMS, AMQP, etc).
*
* @author Oleg Zhurakousky
* @since 2.1
*
*/
public interface RequestReplyHeaderMapper<T> {
void fromHeadersToRequest(MessageHeaders headers, T target);
void fromHeadersToReply(MessageHeaders headers, T target);
Map<String, Object> toHeadersFromRequest(T source);
Map<String, Object> toHeadersFromReply(T source);
}

View File

@@ -88,12 +88,6 @@
</repository>
</repositories>
<dependencies>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>2.1.0.BUILD-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>com.sun.xml.messaging.saaj</groupId>
<artifactId>saaj-impl</artifactId>
@@ -151,6 +145,12 @@
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>xstream</groupId>
<artifactId>xstream</artifactId>
<version>1.2.2</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-web</artifactId>
@@ -213,6 +213,12 @@
<version>1.8.4</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
<version>2.1.0.BUILD-SNAPSHOT</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit-dep</artifactId>

View File

@@ -21,14 +21,12 @@ import org.springframework.expression.ExpressionException;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.integration.gateway.MessagingGatewaySupport;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.ws.WebServiceMessage;
import org.springframework.ws.context.MessageContext;
import org.springframework.ws.server.endpoint.MessageEndpoint;
import org.springframework.ws.soap.SoapHeader;
import org.springframework.ws.soap.SoapMessage;
/**
@@ -37,13 +35,13 @@ import org.springframework.ws.soap.SoapMessage;
*/
abstract public class AbstractWebServiceInboundGateway extends MessagingGatewaySupport implements MessageEndpoint {
protected volatile HeaderMapper<SoapHeader> headerMapper = new DefaultSoapHeaderMapper();
protected volatile SoapHeaderMapper headerMapper = new DefaultSoapHeaderMapper();
public String getComponentType() {
return "ws:outbound-gateway";
return "ws:inbound-gateway";
}
public void setHeaderMapper(HeaderMapper<SoapHeader> headerMapper) {
public void setHeaderMapper(SoapHeaderMapper headerMapper) {
Assert.notNull(headerMapper, "headerMapper must not be null");
this.headerMapper = headerMapper;
}
@@ -73,7 +71,7 @@ abstract public class AbstractWebServiceInboundGateway extends MessagingGatewayS
}
if (request instanceof SoapMessage) {
SoapMessage soapMessage = (SoapMessage) request;
Map<String, ?> headers = this.headerMapper.toHeaders(soapMessage.getSoapHeader());
Map<String, ?> headers = this.headerMapper.toHeadersFromRequest(soapMessage);
if (!CollectionUtils.isEmpty(headers)) {
builder.copyHeaders(headers);
}
@@ -82,8 +80,8 @@ abstract public class AbstractWebServiceInboundGateway extends MessagingGatewayS
protected void toSoapHeaders(WebServiceMessage response, Message<?> replyMessage){
if (response instanceof SoapMessage) {
this.headerMapper.fromHeaders(
replyMessage.getHeaders(), ((SoapMessage) response).getSoapHeader());
this.headerMapper.fromHeadersToReply(
replyMessage.getHeaders(), (SoapMessage) response);
}
}

View File

@@ -23,6 +23,8 @@ import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;
import javax.xml.transform.TransformerException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.context.expression.BeanFactoryResolver;
import org.springframework.context.expression.MapAccessor;
@@ -33,8 +35,8 @@ import org.springframework.expression.spel.support.StandardTypeConverter;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageDeliveryException;
import org.springframework.integration.MessagingException;
import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.web.util.UriTemplate;
@@ -43,12 +45,13 @@ import org.springframework.ws.WebServiceMessage;
import org.springframework.ws.WebServiceMessageFactory;
import org.springframework.ws.client.core.FaultMessageResolver;
import org.springframework.ws.client.core.WebServiceMessageCallback;
import org.springframework.ws.client.core.WebServiceMessageExtractor;
import org.springframework.ws.client.core.WebServiceTemplate;
import org.springframework.ws.client.support.destination.DestinationProvider;
import org.springframework.ws.client.support.interceptor.ClientInterceptor;
import org.springframework.ws.soap.SoapMessage;
import org.springframework.ws.soap.client.core.SoapActionCallback;
import org.springframework.ws.transport.WebServiceMessageSender;
import org.springframework.xml.transform.TransformerObjectSupport;
/**
* Base class for outbound Web Service-invoking Messaging Gateways.
@@ -71,7 +74,8 @@ public abstract class AbstractWebServiceOutboundGateway extends AbstractReplyPro
private volatile WebServiceMessageCallback requestCallback;
private volatile boolean ignoreEmptyResponses = true;
protected volatile SoapHeaderMapper headerMapper = new DefaultSoapHeaderMapper();
public AbstractWebServiceOutboundGateway(String uri, WebServiceMessageFactory messageFactory) {
Assert.hasText(uri, "URI must not be empty");
@@ -92,6 +96,9 @@ public abstract class AbstractWebServiceOutboundGateway extends AbstractReplyPro
this.uriTemplate = null;
}
public void setHeaderMapper(SoapHeaderMapper headerMapper) {
this.headerMapper = headerMapper;
}
/**
* Set the Map of URI variable expressions to evaluate against the outbound message
@@ -160,13 +167,13 @@ public abstract class AbstractWebServiceOutboundGateway extends AbstractReplyPro
}
@Override
public final Object handleRequestMessage(Message<?> message) {
URI uri = prepareUri(message);
public final Object handleRequestMessage(Message<?> requestMessage) {
URI uri = prepareUri(requestMessage);
if (uri == null) {
throw new MessageDeliveryException(message, "Failed to determine URI for " +
throw new MessageDeliveryException(requestMessage, "Failed to determine URI for " +
"Web Service request in outbound gateway: " + this.getComponentName());
}
Object responsePayload = this.doHandle(uri.toString(), message.getPayload(), this.getRequestCallback(message));
Object responsePayload = this.doHandle(uri.toString(), requestMessage, this.requestCallback);
if (responsePayload != null) {
boolean shouldIgnore = (this.ignoreEmptyResponses
&& responsePayload instanceof String && !StringUtils.hasText((String) responsePayload));
@@ -177,7 +184,7 @@ public abstract class AbstractWebServiceOutboundGateway extends AbstractReplyPro
return null;
}
protected abstract Object doHandle(String uri, Object requestPayload, WebServiceMessageCallback requestCallback);
protected abstract Object doHandle(String uri, Message<?> requestMessage, WebServiceMessageCallback requestCallback);
private URI prepareUri(Message<?> requestMessage) {
@@ -191,39 +198,51 @@ public abstract class AbstractWebServiceOutboundGateway extends AbstractReplyPro
}
return this.uriTemplate.expand(uriVariables);
}
private WebServiceMessageCallback getRequestCallback(Message<?> requestMessage) {
String soapAction = requestMessage.getHeaders().get(WebServiceHeaders.SOAP_ACTION, String.class);
return (soapAction != null) ?
new TypeCheckingSoapActionCallback(soapAction, this.requestCallback) : this.requestCallback;
}
private static class TypeCheckingSoapActionCallback extends SoapActionCallback {
private final WebServiceMessageCallback callbackDelegate;
TypeCheckingSoapActionCallback(String soapAction, WebServiceMessageCallback callbackDelegate) {
super(soapAction);
this.callbackDelegate = callbackDelegate;
protected abstract class RequestMessageCallback extends TransformerObjectSupport implements WebServiceMessageCallback {
private final WebServiceMessageCallback requestCallback;
private final Message<?> requestMessage;
public RequestMessageCallback(WebServiceMessageCallback requestCallback, Message<?> requestMessage){
this.requestCallback = requestCallback;
this.requestMessage = requestMessage;
}
@Override
public void doWithMessage(WebServiceMessage message) throws IOException {
if (message instanceof SoapMessage) {
super.doWithMessage(message);
public void doWithMessage(WebServiceMessage message) throws IOException, TransformerException {
Object payload = this.requestMessage.getPayload();
if (message instanceof SoapMessage){
this.doWithMessageInternal(message, payload);
headerMapper.fromHeadersToRequest(this.requestMessage.getHeaders(), (SoapMessage)message);
if (requestCallback != null) {
requestCallback.doWithMessage(message);
}
}
if (this.callbackDelegate != null) {
try {
this.callbackDelegate.doWithMessage(message);
}
catch (Exception e) {
throw new MessagingException("error occurred in WebServiceMessageCallback", e);
}
}
}
}
public abstract void doWithMessageInternal(WebServiceMessage message, Object payload) throws IOException, TransformerException;
}
protected abstract class ResponseMessageExtractor extends TransformerObjectSupport implements WebServiceMessageExtractor<Object> {
public Object extractData(WebServiceMessage message)
throws IOException, TransformerException {
Object resultObject = this.doExtractData(message);
if (message instanceof SoapMessage){
Map<String, Object> mappedMessageHeaders = headerMapper.toHeadersFromReply((SoapMessage) message);
Message<?> siMessage = MessageBuilder.withPayload(resultObject).copyHeaders(mappedMessageHeaders).build();
return siMessage;
}
else {
return message.getPayloadSource();
}
}
public abstract Object doExtractData(WebServiceMessage message) throws IOException, TransformerException;
}
/**
* HTTP-specific subclass of UriTemplate, overriding the encode method.

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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,19 +16,21 @@
package org.springframework.integration.ws;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import javax.xml.namespace.QName;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.mapping.AbstractHeaderMapper;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.PatternMatchUtils;
import org.springframework.util.StringUtils;
import org.springframework.ws.soap.SoapHeader;
import org.springframework.ws.soap.SoapHeaderElement;
import org.springframework.ws.soap.SoapMessage;
import org.springframework.xml.namespace.QNameUtils;
/**
@@ -41,85 +43,77 @@ import org.springframework.xml.namespace.QNameUtils;
* one should implement the HeaderMapper interface directly.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.0
*/
public class DefaultSoapHeaderMapper implements HeaderMapper<SoapHeader> {
public class DefaultSoapHeaderMapper extends AbstractHeaderMapper<SoapMessage> implements SoapHeaderMapper {
private static final List<String> STANDARD_HEADER_NAMES = new ArrayList<String>();
private volatile String[] outboundHeaderNames = new String[0];
private volatile String[] inboundHeaderNames = new String[] { "*" };
public void setOutboundHeaderNames(String[] outboundHeaderNames) {
this.outboundHeaderNames = (outboundHeaderNames != null) ? outboundHeaderNames : new String[0];
static {
STANDARD_HEADER_NAMES.add(WebServiceHeaders.SOAP_ACTION);
}
@Override
protected Map<String, Object> extractStandardHeaders(SoapMessage source) {
return Collections.emptyMap();
}
public void setInboundHeaderNames(String[] inboundHeaderNames) {
this.inboundHeaderNames = (inboundHeaderNames != null) ? inboundHeaderNames : new String[0];
}
public void fromHeaders(MessageHeaders headers, SoapHeader target) {
if (target != null && !CollectionUtils.isEmpty(headers)) {
for (String headerName : headers.keySet()) {
if (this.shouldMapOutboundHeader(headerName)) {
Object value = headers.get(headerName);
if (value instanceof String) {
QName qname = QNameUtils.parseQNameString(headerName);
target.addAttribute(qname, (String) value);
}
}
}
}
}
public Map<String, Object> toHeaders(SoapHeader source) {
@Override
protected Map<String, Object> extractUserDefinedHeaders(SoapMessage source) {
SoapHeader soapHeader = source.getSoapHeader();
Map<String, Object> headers = new HashMap<String, Object>();
if (source != null) {
Iterator<?> attributeIter = source.getAllAttributes();
Iterator<?> attributeIter = soapHeader.getAllAttributes();
while (attributeIter.hasNext()) {
Object name = attributeIter.next();
if (name instanceof QName) {
String qnameString = QNameUtils.toQualifiedName((QName) name);
if (this.shouldMapInboundHeader(qnameString)) {
String value = source.getAttributeValue((QName) name);
if (value != null) {
headers.put(qnameString, value);
}
String value = soapHeader.getAttributeValue((QName) name);
if (value != null) {
headers.put(qnameString, value);
}
}
}
Iterator<?> elementIter = source.examineAllHeaderElements();
Iterator<?> elementIter = soapHeader.examineAllHeaderElements();
while (elementIter.hasNext()) {
Object element = elementIter.next();
if (element instanceof SoapHeaderElement) {
QName qname = ((SoapHeaderElement) element).getName();
String qnameString = QNameUtils.toQualifiedName(qname);
if (this.shouldMapInboundHeader(qnameString)) {
headers.put(qnameString, element);
}
headers.put(qnameString, element);
}
}
}
return headers;
}
private boolean shouldMapInboundHeader(String headerName) {
return matchesAny(this.inboundHeaderNames, headerName);
@Override
protected void populateStandardHeaders(Map<String, Object> headers, SoapMessage target) {
String soapAction = getHeaderIfAvailable(headers, WebServiceHeaders.SOAP_ACTION, String.class);
if (!StringUtils.hasText(soapAction)) {
soapAction = "\"\"";
}
target.setSoapAction(soapAction);
}
private boolean shouldMapOutboundHeader(String headerName) {
return matchesAny(this.outboundHeaderNames, headerName);
}
private static boolean matchesAny(String[] patterns, String candidate) {
if (!ObjectUtils.isEmpty(patterns) && QNameUtils.validateQName(candidate)) {
for (String pattern : patterns) {
if (PatternMatchUtils.simpleMatch(pattern, candidate)) {
return true;
}
}
@Override
protected void populateUserDefinedHeader(String headerName, Object headerValue, SoapMessage target) {
SoapHeader soapHeader = target.getSoapHeader();
if (headerValue instanceof String) {
QName qname = QNameUtils.parseQNameString(headerName);
soapHeader.addAttribute(qname, (String) headerValue);
}
return false;
}
@Override
protected List<String> getStandardRequestHeaderNames() {
return STANDARD_HEADER_NAMES;
}
@Override
protected String getStandardHeaderPrefix() {
return WebServiceHeaders.PREFIX;
}
}

View File

@@ -16,22 +16,31 @@
package org.springframework.integration.ws;
import java.io.IOException;
import org.springframework.integration.Message;
import org.springframework.oxm.Marshaller;
import org.springframework.oxm.Unmarshaller;
import org.springframework.util.Assert;
import org.springframework.ws.WebServiceMessage;
import org.springframework.ws.WebServiceMessageFactory;
import org.springframework.ws.client.core.WebServiceMessageCallback;
import org.springframework.ws.client.support.destination.DestinationProvider;
import org.springframework.ws.support.MarshallingUtils;
/**
* An outbound Messaging Gateway for invoking Web Services that also supports
* marshalling and unmarshalling of the request and response messages.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @see Marshaller
* @see Unmarshaller
*/
public class MarshallingWebServiceOutboundGateway extends AbstractWebServiceOutboundGateway {
private volatile Marshaller marshaller;
private volatile Unmarshaller unmarshaller;
public MarshallingWebServiceOutboundGateway(DestinationProvider destinationProvider, Marshaller marshaller, Unmarshaller unmarshaller, WebServiceMessageFactory messageFactory) {
super(destinationProvider, messageFactory);
@@ -43,8 +52,7 @@ public class MarshallingWebServiceOutboundGateway extends AbstractWebServiceOutb
}
public MarshallingWebServiceOutboundGateway(DestinationProvider destinationProvider, Marshaller marshaller, WebServiceMessageFactory messageFactory) {
super(destinationProvider, messageFactory);
this.configureMarshallers(marshaller);
this(destinationProvider, marshaller, null, messageFactory);
}
public MarshallingWebServiceOutboundGateway(DestinationProvider destinationProvider, Marshaller marshaller) {
@@ -61,8 +69,7 @@ public class MarshallingWebServiceOutboundGateway extends AbstractWebServiceOutb
}
public MarshallingWebServiceOutboundGateway(String uri, Marshaller marshaller, WebServiceMessageFactory messageFactory) {
super(uri, messageFactory);
this.configureMarshallers(marshaller);
this(uri, marshaller, null, messageFactory);
}
public MarshallingWebServiceOutboundGateway(String uri, Marshaller marshaller) {
@@ -76,28 +83,43 @@ public class MarshallingWebServiceOutboundGateway extends AbstractWebServiceOutb
*/
private void configureMarshallers(Marshaller marshaller, Unmarshaller unmarshaller) {
Assert.notNull(marshaller, "marshaller must not be null");
if (unmarshaller == null){
Assert.isInstanceOf(Unmarshaller.class, marshaller,
"Marshaller [" + marshaller + "] does not implement the Unmarshaller interface. " +
"Please set an Unmarshaller explicitly by using one of the constructors that accepts " +
"both Marshaller and Unmarshaller arguments.");
unmarshaller = (Unmarshaller) marshaller;
}
Assert.notNull(unmarshaller, "unmarshaller must not be null");
this.getWebServiceTemplate().setMarshaller(marshaller);
this.getWebServiceTemplate().setUnmarshaller(unmarshaller);
}
/**
* Sets the provided Marshaller on this gateway's WebServiceTemplate as both its
* Marshaller and Unmarshaller. Therefore, it must implement both, and it must not be null.
*/
private void configureMarshallers(Marshaller marshaller) {
Assert.notNull(marshaller, "marshaller must not be null");
Assert.isInstanceOf(Unmarshaller.class, marshaller,
"Marshaller [" + marshaller + "] does not implement the Unmarshaller interface. " +
"Please set an Unmarshaller explicitly by using one of the constructors that accepts " +
"both Marshaller and Unmarshaller arguments.");
this.getWebServiceTemplate().setMarshaller(marshaller);
this.getWebServiceTemplate().setUnmarshaller((Unmarshaller) marshaller);
this.marshaller = marshaller;
this.unmarshaller = unmarshaller;
}
@Override
protected Object doHandle(String uri, Object requestPayload, WebServiceMessageCallback requestCallback) {
return this.getWebServiceTemplate().marshalSendAndReceive(uri, requestPayload, requestCallback);
protected Object doHandle(String uri, Message<?> requestMessage, WebServiceMessageCallback requestCallback) {
Object reply = this.getWebServiceTemplate().sendAndReceive(uri,
new MarshallingRequestMessageCallback(requestCallback, requestMessage), new MarshallingResponseMessageExtractor());
return reply;
}
private class MarshallingRequestMessageCallback extends RequestMessageCallback {
public MarshallingRequestMessageCallback(WebServiceMessageCallback requestCallback, Message<?> requestMessage){
super(requestCallback, requestMessage);
}
@Override
public void doWithMessageInternal(WebServiceMessage message, Object payload) throws IOException{
MarshallingUtils.marshal(marshaller, payload, message);
}
}
private class MarshallingResponseMessageExtractor extends ResponseMessageExtractor {
@Override
public Object doExtractData(WebServiceMessage message) throws IOException{
return MarshallingUtils.unmarshal(unmarshaller, message);
}
}
}

View File

@@ -18,6 +18,7 @@ package org.springframework.integration.ws;
import java.io.IOException;
import javax.xml.transform.Result;
import javax.xml.transform.Source;
import javax.xml.transform.TransformerException;
import javax.xml.transform.dom.DOMResult;
@@ -25,7 +26,10 @@ import javax.xml.transform.dom.DOMSource;
import org.w3c.dom.Document;
import org.springframework.integration.Message;
import org.springframework.integration.MessagingException;
import org.springframework.util.Assert;
import org.springframework.ws.WebServiceMessage;
import org.springframework.ws.WebServiceMessageFactory;
import org.springframework.ws.client.core.SourceExtractor;
import org.springframework.ws.client.core.WebServiceMessageCallback;
@@ -41,10 +45,9 @@ import org.springframework.xml.transform.TransformerObjectSupport;
* @author Oleg Zhurakousky
*/
public class SimpleWebServiceOutboundGateway extends AbstractWebServiceOutboundGateway {
private final SourceExtractor<?> sourceExtractor;
public SimpleWebServiceOutboundGateway(DestinationProvider destinationProvider) {
this(destinationProvider, null, null);
}
@@ -73,30 +76,92 @@ public class SimpleWebServiceOutboundGateway extends AbstractWebServiceOutboundG
@Override
protected Object doHandle(String uri, Object requestPayload, WebServiceMessageCallback requestCallback) {
if (requestPayload instanceof Source) {
return this.getWebServiceTemplate().sendSourceAndReceive(
uri, (Source) requestPayload, requestCallback, this.sourceExtractor);
}
protected Object doHandle(String uri, final Message<?> requestMessage, final WebServiceMessageCallback requestCallback) {
Object requestPayload = requestMessage.getPayload();
Result responseResultInstance = null;
if (requestPayload instanceof String) {
StringResult result = new StringResult();
this.getWebServiceTemplate().sendSourceAndReceiveToResult(
uri, new StringSource((String) requestPayload), requestCallback, result);
return result.toString();
responseResultInstance = new StringResult();
}
if (requestPayload instanceof Document) {
DOMResult result = new DOMResult();
this.getWebServiceTemplate().sendSourceAndReceiveToResult(
uri, new DOMSource((Document) requestPayload), requestCallback, result);
return result.getNode();
else if (requestPayload instanceof Document) {
responseResultInstance = new DOMResult();
}
throw new MessagingException("Unsupported payload type '" + requestPayload.getClass() +
"'. " + this.getClass().getName() + " only supports 'java.lang.String', '" + Source.class.getName() +
"', and '" + Document.class.getName() + "'. Consider either using the '"
+ MarshallingWebServiceOutboundGateway.class.getName() + "' or a Message Transformer.");
Object reply = this.getWebServiceTemplate().sendAndReceive(uri,
new SimpleRequestMessageCallback(requestCallback, requestMessage), new SimpleResponseMessageExtractor(responseResultInstance));
return reply;
}
private class SimpleRequestMessageCallback extends RequestMessageCallback {
public SimpleRequestMessageCallback(WebServiceMessageCallback requestCallback, Message<?> requestMessage){
super(requestCallback, requestMessage);
}
@Override
public void doWithMessageInternal(WebServiceMessage message, Object payload) throws IOException, TransformerException {
Source source = this.extractSource(payload);
this.transform(source, message.getPayloadResult());
}
private Source extractSource(Object requestPayload) throws IOException, TransformerException{
Source source = null;
if (requestPayload instanceof Source) {
source = (Source) requestPayload;
Object o = sourceExtractor.extractData(source);
Assert.isInstanceOf(Source.class, o);
source = (Source) o;
}
else if (requestPayload instanceof String) {
source = new StringSource((String) requestPayload);
}
else if (requestPayload instanceof Document) {
source = new DOMSource((Document) requestPayload);
}
else {
throw new MessagingException("Unsupported payload type '" + requestPayload.getClass() +
"'. " + this.getClass().getName() + " only supports 'java.lang.String', '" + Source.class.getName() +
"', and '" + Document.class.getName() + "'. Consider either using the '"
+ MarshallingWebServiceOutboundGateway.class.getName() + "' or a Message Transformer.");
}
return source;
}
}
private class SimpleResponseMessageExtractor extends ResponseMessageExtractor {
private final Result result;
public SimpleResponseMessageExtractor(Result result){
super();
this.result = result;
}
@Override
public Object doExtractData(WebServiceMessage message) throws IOException, TransformerException{
Source payloadSource = message.getPayloadSource();
Object payload = null;
if (this.result != null){
this.transform(payloadSource, this.result);
if (this.result instanceof StringResult){
payload = this.result.toString();
}
else if (this.result instanceof DOMResult){
payload = ((DOMResult)this.result).getNode();
}
else {
payload = this.result;
}
}
else {
payload = payloadSource;
}
return payload;
}
}
private static class DefaultSourceExtractor extends TransformerObjectSupport implements SourceExtractor<DOMSource> {
public DOMSource extractData(Source source) throws IOException, TransformerException {
@@ -108,5 +173,4 @@ public class SimpleWebServiceOutboundGateway extends AbstractWebServiceOutboundG
return new DOMSource(result.getNode());
}
}
}

View File

@@ -0,0 +1,20 @@
/**
*
*/
package org.springframework.integration.ws;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.mapping.RequestReplyHeaderMapper;
import org.springframework.ws.soap.SoapHeader;
import org.springframework.ws.soap.SoapMessage;
/**
* A convenience interface that extends {@link HeaderMapper}
* but parameterized with {@link SoapHeader}.
*
* @author Oleg Zhurakousky
* @since 2.1
*/
public interface SoapHeaderMapper extends RequestReplyHeaderMapper<SoapMessage>{
}

View File

@@ -16,14 +16,15 @@
package org.springframework.integration.ws.config;
import org.w3c.dom.Element;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractInboundGatewayParser;
import org.springframework.util.Assert;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ws.DefaultSoapHeaderMapper;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
/**
* @author Iwein Fuld
@@ -66,12 +67,10 @@ public class WebServiceInboundGatewayParser extends AbstractInboundGatewayParser
logger.warn("Setting 'extract-payload' attribute has no effect when used with a marshalling Web Service Inbound Gateway.");
}
}
String headerMapperRef = element.getAttribute("header-mapper");
if (StringUtils.hasText(headerMapperRef)) {
Assert.isTrue(!StringUtils.hasText(marshallerRef),
"The 'header-mapper' attribute cannot be used when a 'marshaller' is provided.");
builder.addPropertyReference("headerMapper", headerMapperRef);
}
}
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
super.doParse(element, parserContext, builder);
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultSoapHeaderMapper.class, null);
}
}

View File

@@ -28,6 +28,7 @@ import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.ExpressionFactoryBean;
import org.springframework.integration.config.xml.AbstractOutboundGatewayParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.ws.DefaultSoapHeaderMapper;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
import org.springframework.util.xml.DomUtils;
@@ -84,6 +85,9 @@ public class WebServiceOutboundGatewayParser extends AbstractOutboundGatewayPars
IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel");
IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "ignore-empty-responses");
this.postProcessGateway(builder, element, parserContext);
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultSoapHeaderMapper.class, null);
return builder;
}

View File

@@ -204,6 +204,38 @@
</xsd:attribute>
<xsd:attribute name="order" type="xsd:string"/>
<xsd:attribute name="auto-startup" type="xsd:string" default="true"/>
<xsd:attribute name="header-mapper">
<xsd:annotation>
<xsd:documentation>
Reference to a HeaderMapper&lt;SoapHeader&gt; implementation
that this gateway will use to map between Spring Integration
MessageHeaders and the SoapHeader.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.integration.mapping.HeaderMapper"/>
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of SOAP Headers to be mapped from the SOAP request into the MessageHeaders.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-reply-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of MessageHeaders to be mapped into the SOAP Headers of the SOAP reply.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>
@@ -295,8 +327,7 @@
<xsd:documentation>
Reference to a HeaderMapper&lt;SoapHeader&gt; implementation
that this gateway will use to map between Spring Integration
MessageHeaders and the SoapHeader. This strategy can only be
applied when a 'marshaller' is not being configured.
MessageHeaders and the SoapHeader.
</xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
@@ -305,6 +336,24 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of SOAP Headers to be mapped from the SOAP request into the MessageHeaders.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-reply-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of MessageHeaders to be mapped into the SOAP Headers of the SOAP reply.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
</xsd:element>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -25,6 +25,7 @@ import javax.xml.transform.Source;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.springframework.beans.DirectFieldAccessor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
@@ -33,17 +34,17 @@ import org.springframework.integration.MessageChannel;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.core.PollableChannel;
import org.springframework.integration.history.MessageHistory;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.ws.MarshallingWebServiceInboundGateway;
import org.springframework.integration.ws.SimpleWebServiceInboundGateway;
import org.springframework.integration.ws.SoapHeaderMapper;
import org.springframework.oxm.Unmarshaller;
import org.springframework.oxm.support.AbstractMarshaller;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.ws.context.DefaultMessageContext;
import org.springframework.ws.context.MessageContext;
import org.springframework.ws.soap.SoapHeader;
import org.springframework.ws.soap.SoapMessage;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertNotNull;
@@ -149,7 +150,7 @@ public class WebServiceInboundGatewayParserTests {
assertNotNull(history);
Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "marshalling", 0);
assertNotNull(componentHistoryRecord);
assertEquals("ws:outbound-gateway", componentHistoryRecord.get("type"));
assertEquals("ws:inbound-gateway", componentHistoryRecord.get("type"));
}
@Test
@@ -162,14 +163,14 @@ public class WebServiceInboundGatewayParserTests {
Properties componentHistoryRecord = TestUtils.locateComponentInHistory(history, "extractsPayload", 0);
System.out.println(componentHistoryRecord);
assertNotNull(componentHistoryRecord);
assertEquals("ws:outbound-gateway", componentHistoryRecord.get("type"));
assertEquals("ws:inbound-gateway", componentHistoryRecord.get("type"));
}
@Autowired
private SimpleWebServiceInboundGateway headerMappingGateway;
@Autowired
private HeaderMapper<SoapHeader> testHeaderMapper;
private SoapHeaderMapper testHeaderMapper;
@Test
public void testHeaderMapperReference() throws Exception {
@@ -180,12 +181,20 @@ public class WebServiceInboundGatewayParserTests {
@SuppressWarnings("unused")
private static class TestHeaderMapper implements HeaderMapper<SoapHeader> {
public void fromHeaders(MessageHeaders headers, SoapHeader target) {
private static class TestHeaderMapper implements SoapHeaderMapper {
public void fromHeadersToRequest(MessageHeaders headers,
SoapMessage target) {
}
public Map<String, ?> toHeaders(SoapHeader source) {
public void fromHeadersToReply(MessageHeaders headers, SoapMessage target) {
}
public Map<String, Object> toHeadersFromRequest(SoapMessage source) {
return Collections.emptyMap();
}
public Map<String, Object> toHeadersFromReply(SoapMessage source) {
return Collections.emptyMap();
}
}

View File

@@ -16,9 +16,6 @@
package org.springframework.integration.ws.config;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import org.junit.Test;
import org.springframework.beans.DirectFieldAccessor;
@@ -28,6 +25,7 @@ import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.endpoint.AbstractEndpoint;
import org.springframework.integration.endpoint.EventDrivenConsumer;
import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.ws.MarshallingWebServiceOutboundGateway;
import org.springframework.integration.ws.SimpleWebServiceOutboundGateway;
import org.springframework.oxm.Marshaller;
@@ -40,6 +38,9 @@ import org.springframework.ws.client.core.WebServiceMessageCallback;
import org.springframework.ws.client.support.interceptor.ClientInterceptor;
import org.springframework.ws.transport.WebServiceMessageSender;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
/**
* @author Mark Fisher
*/
@@ -261,14 +262,10 @@ public class WebServiceOutboundGatewayParserTests {
"marshallingWebServiceOutboundGatewayParserTests.xml", this.getClass());
AbstractEndpoint endpoint = (AbstractEndpoint) context.getBean("gatewayWithAllInOneMarshaller");
assertEquals(EventDrivenConsumer.class, endpoint.getClass());
Object gateway = new DirectFieldAccessor(endpoint).getPropertyValue("handler");
assertEquals(MarshallingWebServiceOutboundGateway.class, gateway.getClass());
DirectFieldAccessor gatewayAccessor = new DirectFieldAccessor(gateway);
DirectFieldAccessor templateAccessor = new DirectFieldAccessor(
gatewayAccessor.getPropertyValue("webServiceTemplate"));
MarshallingWebServiceOutboundGateway gateway = (MarshallingWebServiceOutboundGateway) new DirectFieldAccessor(endpoint).getPropertyValue("handler");
Marshaller marshaller = (Marshaller) context.getBean("marshallerAndUnmarshaller");
assertEquals(marshaller, templateAccessor.getPropertyValue("marshaller"));
assertEquals(marshaller, templateAccessor.getPropertyValue("unmarshaller"));
assertEquals(marshaller, TestUtils.getPropertyValue(gateway, "marshaller", Marshaller.class));
assertEquals(marshaller, TestUtils.getPropertyValue(gateway, "unmarshaller", Unmarshaller.class));
}
@Test
@@ -277,15 +274,11 @@ public class WebServiceOutboundGatewayParserTests {
"marshallingWebServiceOutboundGatewayParserTests.xml", this.getClass());
AbstractEndpoint endpoint = (AbstractEndpoint) context.getBean("gatewayWithSeparateMarshallerAndUnmarshaller");
assertEquals(EventDrivenConsumer.class, endpoint.getClass());
Object gateway = new DirectFieldAccessor(endpoint).getPropertyValue("handler");
assertEquals(MarshallingWebServiceOutboundGateway.class, gateway.getClass());
DirectFieldAccessor gatewayAccessor = new DirectFieldAccessor(gateway);
DirectFieldAccessor templateAccessor = new DirectFieldAccessor(
gatewayAccessor.getPropertyValue("webServiceTemplate"));
MarshallingWebServiceOutboundGateway gateway = (MarshallingWebServiceOutboundGateway) new DirectFieldAccessor(endpoint).getPropertyValue("handler");
Marshaller marshaller = (Marshaller) context.getBean("marshaller");
Unmarshaller unmarshaller = (Unmarshaller) context.getBean("unmarshaller");
assertEquals(marshaller, templateAccessor.getPropertyValue("marshaller"));
assertEquals(unmarshaller, templateAccessor.getPropertyValue("unmarshaller"));
assertEquals(marshaller, TestUtils.getPropertyValue(gateway, "marshaller", Marshaller.class));
assertEquals(unmarshaller, TestUtils.getPropertyValue(gateway, "unmarshaller", Unmarshaller.class));
}
@Test
@@ -307,16 +300,13 @@ public class WebServiceOutboundGatewayParserTests {
"marshallingWebServiceOutboundGatewayParserTests.xml", this.getClass());
AbstractEndpoint endpoint = (AbstractEndpoint) context.getBean("gatewayWithAllInOneMarshallerAndMessageFactory");
assertEquals(EventDrivenConsumer.class, endpoint.getClass());
Object gateway = new DirectFieldAccessor(endpoint).getPropertyValue("handler");
assertEquals(MarshallingWebServiceOutboundGateway.class, gateway.getClass());
DirectFieldAccessor gatewayAccessor = new DirectFieldAccessor(gateway);
DirectFieldAccessor templateAccessor = new DirectFieldAccessor(
gatewayAccessor.getPropertyValue("webServiceTemplate"));
MarshallingWebServiceOutboundGateway gateway = (MarshallingWebServiceOutboundGateway) new DirectFieldAccessor(endpoint).getPropertyValue("handler");
Marshaller marshaller = (Marshaller) context.getBean("marshallerAndUnmarshaller");
assertEquals(marshaller, templateAccessor.getPropertyValue("marshaller"));
assertEquals(marshaller, templateAccessor.getPropertyValue("unmarshaller"));
assertEquals(marshaller, TestUtils.getPropertyValue(gateway, "marshaller", Marshaller.class));
assertEquals(marshaller, TestUtils.getPropertyValue(gateway, "unmarshaller", Unmarshaller.class));
WebServiceMessageFactory messageFactory = (WebServiceMessageFactory) context.getBean("messageFactory");
assertEquals(messageFactory, templateAccessor.getPropertyValue("messageFactory"));
assertEquals(messageFactory, TestUtils.getPropertyValue(gateway, "webServiceTemplate.messageFactory"));
}
@Test
@@ -325,17 +315,14 @@ public class WebServiceOutboundGatewayParserTests {
"marshallingWebServiceOutboundGatewayParserTests.xml", this.getClass());
AbstractEndpoint endpoint = (AbstractEndpoint) context.getBean("gatewayWithSeparateMarshallerAndUnmarshallerAndMessageFactory");
assertEquals(EventDrivenConsumer.class, endpoint.getClass());
Object gateway = new DirectFieldAccessor(endpoint).getPropertyValue("handler");
assertEquals(MarshallingWebServiceOutboundGateway.class, gateway.getClass());
DirectFieldAccessor gatewayAccessor = new DirectFieldAccessor(gateway);
DirectFieldAccessor templateAccessor = new DirectFieldAccessor(
gatewayAccessor.getPropertyValue("webServiceTemplate"));
MarshallingWebServiceOutboundGateway gateway = (MarshallingWebServiceOutboundGateway) new DirectFieldAccessor(endpoint).getPropertyValue("handler");
Marshaller marshaller = (Marshaller) context.getBean("marshaller");
Unmarshaller unmarshaller = (Unmarshaller) context.getBean("unmarshaller");
assertEquals(marshaller, templateAccessor.getPropertyValue("marshaller"));
assertEquals(unmarshaller, templateAccessor.getPropertyValue("unmarshaller"));
assertEquals(marshaller, TestUtils.getPropertyValue(gateway, "marshaller", Marshaller.class));
assertEquals(unmarshaller, TestUtils.getPropertyValue(gateway, "unmarshaller", Unmarshaller.class));
WebServiceMessageFactory messageFactory = (WebServiceMessageFactory) context.getBean("messageFactory");
assertEquals(messageFactory, templateAccessor.getPropertyValue("messageFactory"));
assertEquals(messageFactory, TestUtils.getPropertyValue(gateway, "webServiceTemplate.messageFactory"));
}
@Test

View File

@@ -0,0 +1,215 @@
/*
* Copyright 2002-2011 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.ws.config;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.net.URI;
import java.util.List;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.transform.Source;
import javax.xml.transform.dom.DOMSource;
import org.junit.Test;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;
import org.springframework.integration.Message;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.ws.AbstractWebServiceOutboundGateway;
import org.springframework.integration.ws.DefaultSoapHeaderMapper;
import org.springframework.integration.ws.SimpleWebServiceOutboundGateway;
import org.springframework.integration.ws.WebServiceHeaders;
import org.springframework.oxm.Unmarshaller;
import org.springframework.oxm.XmlMappingException;
import org.springframework.util.xml.DomUtils;
import org.springframework.ws.WebServiceMessage;
import org.springframework.ws.WebServiceMessageFactory;
import org.springframework.ws.soap.SoapHeader;
import org.springframework.ws.soap.SoapMessage;
import org.springframework.ws.soap.SoapMessageFactory;
import org.springframework.ws.transport.WebServiceConnection;
import org.springframework.ws.transport.WebServiceMessageSender;
import org.springframework.xml.namespace.QNameUtils;
import static junit.framework.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
/**
* @author Oleg Zhurakousky
*
*/
public class WebServiceOutboundGatewayWithHeaderMapperTests {
String responseMessage = "<?xml version=\"1.0\" encoding=\"UTF-8\" ?> " +
"<SOAP-ENV:Envelope xmlns:SOAP-ENV=\"http://schemas.xmlsoap.org/soap/envelope/\"> " +
"<SOAP-ENV:Header/>" +
"<SOAP-ENV:Body> " +
"<root><name>jane</name></root>" +
"</SOAP-ENV:Body> " +
"</SOAP-ENV:Envelope>";
@SuppressWarnings("unchecked")
@Test
public void headerMapperParserTest() throws Exception{
ApplicationContext context = new ClassPathXmlApplicationContext("ws-outbound-gateway-with-headermappers.xml", this.getClass());
SimpleWebServiceOutboundGateway gateway = TestUtils.getPropertyValue(context.getBean("withHeaderMapper"), "handler", SimpleWebServiceOutboundGateway.class);
DefaultSoapHeaderMapper headerMapper = TestUtils.getPropertyValue(gateway, "headerMapper", DefaultSoapHeaderMapper.class);
assertNotNull(headerMapper);
List<String> requestHeaderNames = TestUtils.getPropertyValue(headerMapper, "requestHeaderNames", List.class);
assertEquals(2, requestHeaderNames.size());
assertEquals("foo*", requestHeaderNames.get(0));
assertEquals("*baz*", requestHeaderNames.get(1));
List<String> responseHeaderNames = TestUtils.getPropertyValue(headerMapper, "replyHeaderNames", List.class);
assertEquals(1, responseHeaderNames.size());
assertEquals("bar*", responseHeaderNames.get(0));
}
@Test
public void withHeaderMapperString() throws Exception{
String payload = "<root><name>bill</name></root>";
this.process(payload, "withHeaderMapper", "inputChannel");
}
@Test
public void withHeaderMapperSource() throws Exception{
DocumentBuilderFactory dbfac = DocumentBuilderFactory.newInstance();
DocumentBuilder docBuilder = dbfac.newDocumentBuilder();
Document document = docBuilder.parse(new ByteArrayInputStream("<root><name>bill</name></root>".getBytes()));
DOMSource payload = new DOMSource(document);
this.process(payload, "withHeaderMapper", "inputChannel");
}
@Test
public void withHeaderMapperDocument() throws Exception{
DocumentBuilderFactory dbfac = DocumentBuilderFactory.newInstance();
DocumentBuilder docBuilder = dbfac.newDocumentBuilder();
Document payload = docBuilder.parse(new ByteArrayInputStream("<root><name>bill</name></root>".getBytes()));
this.process(payload, "withHeaderMapper", "inputChannel");
}
@Test
public void withHeaderMapperAndMarshaller() throws Exception{
Person person = new Person();
person.setName("Bill Clinton");
this.process(person, "marshallingWithHeaderMapper", "inputMarshallingChannel");
}
@SuppressWarnings("rawtypes")
public void process(Object payload, String gatewayName, String channelName) throws Exception{
ApplicationContext context = new ClassPathXmlApplicationContext("ws-outbound-gateway-with-headermappers.xml", this.getClass());
AbstractWebServiceOutboundGateway gateway = TestUtils.getPropertyValue(context.getBean(gatewayName), "handler", AbstractWebServiceOutboundGateway.class);
WebServiceMessageSender messageSender = Mockito.mock(WebServiceMessageSender.class);
WebServiceConnection wsConnection = Mockito.mock(WebServiceConnection.class);
Mockito.when(messageSender.createConnection(Mockito.any(URI.class))).thenReturn(wsConnection);
Mockito.when(messageSender.supports(Mockito.any(URI.class))).thenReturn(true);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
SoapMessage soapMessage = (SoapMessage) args[0];
// try { // uncomment if you want to see a pretty-print of SOAP message
// Transformer transformer = TransformerFactory.newInstance().newTransformer();
// transformer.setOutputProperty(OutputKeys.INDENT, "yes");
// transformer.transform(new DOMSource(soapMessage.getDocument()), new StreamResult(System.out));
// } catch (Exception e) {
// // ignore
// }
SoapHeader soapHeader = soapMessage.getSoapHeader();
assertNotNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("foo")));
assertNotNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("foobar")));
assertNotNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("abaz")));
assertNull(soapHeader.getAttributeValue(QNameUtils.parseQNameString("bar")));
return null;
}})
.when(wsConnection).send(Mockito.any(WebServiceMessage.class));
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) throws Exception{
Object[] args = invocation.getArguments();
SoapMessageFactory factory = (SoapMessageFactory) args[0];
SoapMessage soapMessage = factory.createWebServiceMessage(new ByteArrayInputStream(responseMessage.getBytes()));
soapMessage.getSoapHeader().addAttribute(QNameUtils.parseQNameString("bar"), "bar");
soapMessage.getSoapHeader().addAttribute(QNameUtils.parseQNameString("baz"), "baz");
// try { // uncomment if you want to see a pretty-print of SOAP message
// Transformer transformer = TransformerFactory.newInstance().newTransformer();
// transformer.setOutputProperty(OutputKeys.INDENT, "yes");
// transformer.transform(new DOMSource(soapMessage.getDocument()), new StreamResult(System.out));
// } catch (Exception e) {
// // ignore
// }
return soapMessage;
}})
.when(wsConnection).receive(Mockito.any(WebServiceMessageFactory.class));
gateway.setMessageSender(messageSender);
MessageChannel inputChannel = context.getBean(channelName, MessageChannel.class);
Message<?> message =
MessageBuilder.withPayload(payload).
setHeader("foo", "foo").setHeader("foobar", "foobar").setHeader("abaz", "abaz").setHeader("bar", "bar").
setHeader(WebServiceHeaders.SOAP_ACTION, "someAction").build();
inputChannel.send(message);
QueueChannel outputChannel = context.getBean("outputChannel", QueueChannel.class);
Message<?> replyMessage = outputChannel.receive(0);
assertEquals("bar", replyMessage.getHeaders().get("bar"));
assertNull(replyMessage.getHeaders().get("baz"));
}
public static class Person{
private String name;
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
}
public static class SampleUnmarshaller implements Unmarshaller {
public boolean supports(Class<?> clazz) {
return true;
}
public Object unmarshal(Source source) throws IOException, XmlMappingException {
Element documentElement = (Element) ((DOMSource) source).getNode();
String name = DomUtils.getChildElementValueByTagName(documentElement, "name");
Person person = new Person();
person.setName(name);
return person;
}
}
}

View File

@@ -0,0 +1,38 @@
<?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:int-ws="http://www.springframework.org/schema/integration/ws"
xmlns:int="http://www.springframework.org/schema/integration"
xsi:schemaLocation="http://www.springframework.org/schema/integration/ws http://www.springframework.org/schema/integration/ws/spring-integration-ws-2.1.xsd
http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<int:channel id="inputChannel"/>
<int:channel id="inputMarshallingChannel"/>
<int-ws:outbound-gateway id="withHeaderMapper"
request-channel="inputChannel"
reply-channel="outputChannel"
uri="http://example.org"
mapped-request-headers="foo*, *baz*"
mapped-reply-headers="bar*"/>
<int-ws:outbound-gateway id="marshallingWithHeaderMapper"
request-channel="inputMarshallingChannel"
reply-channel="outputChannel"
uri="http://example.org"
marshaller="marshaller"
unmarshaller="ubmarshaller"
mapped-request-headers="foo*, *baz*"
mapped-reply-headers="bar*"/>
<bean id="marshaller" class="org.springframework.oxm.xstream.XStreamMarshaller"/>
<bean id="ubmarshaller" class="org.springframework.integration.ws.config.WebServiceOutboundGatewayWithHeaderMapperTests.SampleUnmarshaller"/>
<int:channel id="outputChannel">
<int:queue/>
</int:channel>
</beans>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -27,13 +27,29 @@ package org.springframework.integration.xmpp;
*/
public class XmppHeaders {
private static final String PREFIX = "xmpp_";
public static final String PREFIX = "xmpp_";
public static final String CHAT = PREFIX + "chatKey";
public static final String CHAT = PREFIX + "chat";
public static final String CHAT_TO = PREFIX + "chatTo";
public static final String TO = PREFIX + "to";
public static final String CHAT_THREAD_ID = PREFIX + "chatThreadId";
/**
* {@link Deprecated} use {@link #TO} instead
*/
@Deprecated
public static final String CHAT_TO = TO;
public static final String FROM = PREFIX + "from";
public static final String THREAD = PREFIX + "thread";
/**
* {@link Deprecated} use {@link #THREAD} instead
*/
@Deprecated
public static final String CHAT_THREAD_ID = THREAD;
public static final String SUBJECT = PREFIX + "subject";
public static final String TYPE = PREFIX + "type";

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -23,6 +23,7 @@ import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.xmpp.support.DefaultXmppHeaderMapper;
import org.springframework.util.StringUtils;
/**
@@ -45,6 +46,9 @@ public abstract class AbstractXmppInboundChannelAdapterParser extends AbstractSi
@Override
protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultXmppHeaderMapper.class, null);
String connectionName = element.getAttribute("xmpp-connection");
if (StringUtils.hasText(connectionName)){

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -21,6 +21,8 @@ import org.springframework.beans.factory.support.AbstractBeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.xml.ParserContext;
import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
import org.springframework.integration.xmpp.support.DefaultXmppHeaderMapper;
import org.springframework.util.StringUtils;
import org.w3c.dom.Element;
@@ -36,6 +38,9 @@ public abstract class AbstractXmppOutboundChannelAdapterParser extends AbstractO
@Override
protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(this.getHandlerClassName());
IntegrationNamespaceUtils.configureHeaderMapper(element, builder, parserContext, DefaultXmppHeaderMapper.class, null);
String connectionName = element.getAttribute("xmpp-connection");
if (StringUtils.hasText(connectionName)){
builder.addConstructorArgReference(connectionName);

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -22,14 +22,14 @@ import org.springframework.integration.xmpp.XmppHeaders;
/**
* Parser for 'xmpp:header-enricher' element
* @author Josh Long
* @author Oleg ZHurakousky
* @author Oleg Zhurakousky
* @since 2.0
*/
public class XmppHeaderEnricherParser extends HeaderEnricherParserSupport {
public XmppHeaderEnricherParser() {
this.addElementToHeaderMapping("chat-to", XmppHeaders.CHAT_TO);
this.addElementToHeaderMapping("chat-thread-id", XmppHeaders.CHAT_THREAD_ID);
this.addElementToHeaderMapping("chat-to", XmppHeaders.TO);
this.addElementToHeaderMapping("chat-thread-id", XmppHeaders.THREAD);
}
}

View File

@@ -16,14 +16,16 @@
package org.springframework.integration.xmpp.inbound;
import org.jivesoftware.smack.Chat;
import java.util.Map;
import org.jivesoftware.smack.PacketListener;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.packet.Packet;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.xmpp.XmppHeaders;
import org.springframework.integration.xmpp.core.AbstractXmppConnectionAwareEndpoint;
import org.springframework.integration.xmpp.support.DefaultXmppHeaderMapper;
import org.springframework.integration.xmpp.support.XmppHeaderMapper;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -41,7 +43,8 @@ public class ChatMessageListeningEndpoint extends AbstractXmppConnectionAwareEnd
private volatile boolean extractPayload = true;
private final PacketListener packetListener = new ChatMessagePublishingPacketListener();
private volatile XmppHeaderMapper headerMapper = new DefaultXmppHeaderMapper();
public ChatMessageListeningEndpoint() {
super();
@@ -50,6 +53,10 @@ public class ChatMessageListeningEndpoint extends AbstractXmppConnectionAwareEnd
public ChatMessageListeningEndpoint(XMPPConnection xmppConnection) {
super(xmppConnection);
}
public void setHeaderMapper(XmppHeaderMapper headerMapper) {
this.headerMapper = headerMapper;
}
/**
@@ -85,8 +92,8 @@ public class ChatMessageListeningEndpoint extends AbstractXmppConnectionAwareEnd
public void processPacket(final Packet packet) {
if (packet instanceof org.jivesoftware.smack.packet.Message) {
org.jivesoftware.smack.packet.Message xmppMessage = (org.jivesoftware.smack.packet.Message) packet;
Chat chat = xmppConnection.getChatManager().getThreadChat(xmppMessage.getThread());
Map<String, ?> mappedHeaders = headerMapper.toHeadersFromRequest(xmppMessage);
String messageBody = xmppMessage.getBody();
/*
* Since there are several types of chat messages with different ChatState (e.g., composing, paused etc)
@@ -98,9 +105,8 @@ public class ChatMessageListeningEndpoint extends AbstractXmppConnectionAwareEnd
*/
if (StringUtils.hasText(messageBody)){
Object payload = (extractPayload ? messageBody : xmppMessage);
MessageBuilder<?> messageBuilder = MessageBuilder.withPayload(payload)
.setHeader(XmppHeaders.TYPE, xmppMessage.getType())
.setHeader(XmppHeaders.CHAT, chat);
MessageBuilder<?> messageBuilder = MessageBuilder.withPayload(payload).copyHeaders(mappedHeaders);
sendMessage(messageBuilder.build());
}
}

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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.
@@ -22,6 +22,8 @@ import org.springframework.integration.Message;
import org.springframework.integration.MessageHandlingException;
import org.springframework.integration.xmpp.XmppHeaders;
import org.springframework.integration.xmpp.core.AbstractXmppConnectionAwareMessageHandler;
import org.springframework.integration.xmpp.support.DefaultXmppHeaderMapper;
import org.springframework.integration.xmpp.support.XmppHeaderMapper;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -35,7 +37,10 @@ import org.springframework.util.StringUtils;
* @since 2.0
*/
public class ChatMessageSendingMessageHandler extends AbstractXmppConnectionAwareMessageHandler {
private volatile XmppHeaderMapper headerMapper = new DefaultXmppHeaderMapper();
public ChatMessageSendingMessageHandler() {
super();
}
@@ -44,6 +49,10 @@ public class ChatMessageSendingMessageHandler extends AbstractXmppConnectionAwar
super(xmppConnection);
}
public void setHeaderMapper(XmppHeaderMapper headerMapper) {
Assert.notNull(headerMapper, "headerMapper must not be null");
this.headerMapper = headerMapper;
}
@Override
protected void handleMessageInternal(Message<?> message) throws Exception {
@@ -54,12 +63,11 @@ public class ChatMessageSendingMessageHandler extends AbstractXmppConnectionAwar
xmppMessage = (org.jivesoftware.smack.packet.Message) messageBody;
}
else if (messageBody instanceof String) {
String chatTo = message.getHeaders().get(XmppHeaders.CHAT_TO, String.class);
Assert.state(StringUtils.hasText(chatTo), "The '" + XmppHeaders.CHAT_TO + "' header must not be null");
xmppMessage = new org.jivesoftware.smack.packet.Message(chatTo);
String threadId = message.getHeaders().get(XmppHeaders.CHAT_THREAD_ID, String.class);
if (StringUtils.hasText(threadId)) {
xmppMessage.setThread(threadId);
String to = message.getHeaders().get(XmppHeaders.TO, String.class);
Assert.state(StringUtils.hasText(to), "The '" + XmppHeaders.TO + "' header must not be null");
xmppMessage = new org.jivesoftware.smack.packet.Message(to);
if (this.headerMapper != null) {
this.headerMapper.fromHeadersToRequest(message.getHeaders(), xmppMessage);
}
xmppMessage.setBody((String) messageBody);
}

View File

@@ -0,0 +1,148 @@
/*
* Copyright 2002-2011 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.xmpp.support;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.jivesoftware.smack.packet.Message;
import org.springframework.integration.mapping.AbstractHeaderMapper;
import org.springframework.integration.xmpp.XmppHeaders;
import org.springframework.util.StringUtils;
/**
* Default implementation of {@link XmppHeaderMapper}.
*
* @author Mark Fisher
* @author Oleg Zhurakousky
* @since 2.1
*/
public class DefaultXmppHeaderMapper extends AbstractHeaderMapper<Message> implements XmppHeaderMapper {
private static final List<String> STANDARD_HEADER_NAMES = new ArrayList<String>();
static {
STANDARD_HEADER_NAMES.add(XmppHeaders.FROM);
STANDARD_HEADER_NAMES.add(XmppHeaders.SUBJECT);
STANDARD_HEADER_NAMES.add(XmppHeaders.THREAD);
STANDARD_HEADER_NAMES.add(XmppHeaders.TO);
STANDARD_HEADER_NAMES.add(XmppHeaders.TYPE);
}
@Override
protected Map<String, Object> extractStandardHeaders(Message source) {
Map<String, Object> headers = new HashMap<String, Object>();
/*Collection<PacketExtension> extensions = source.getExtensions();
if (!CollectionUtils.isEmpty(extensions)) {
for (PacketExtension extension : extensions) {
String name = extension.getElementName();
String namespace = extension.getNamespace();
if (StringUtils.hasText(namespace)) {
name = namespace + ":" + name;
}
headers.put(name, extension.toXML());
}
}*/
String from = source.getFrom();
if (StringUtils.hasText(from)) {
headers.put(XmppHeaders.FROM, from);
}
String subject = source.getSubject();
if (StringUtils.hasText(subject)) {
headers.put(XmppHeaders.SUBJECT, subject);
}
String thread = source.getThread();
if (StringUtils.hasText(thread)) {
headers.put(XmppHeaders.THREAD, thread);
}
String to = source.getTo();
if (StringUtils.hasText(to)) {
headers.put(XmppHeaders.TO, to);
}
Message.Type type = source.getType();
if (type != null) {
headers.put(XmppHeaders.TYPE, type);
}
return headers;
}
@Override
protected Map<String, Object> extractUserDefinedHeaders(Message source) {
Map<String, Object> headers = new HashMap<String, Object>();
for (String propertyName : source.getPropertyNames()) {
headers.put(propertyName, source.getProperty(propertyName));
}
return headers;
}
@Override
protected void populateStandardHeaders(Map<String, Object> headers, Message target) {
String threadId = getHeaderIfAvailable(headers, XmppHeaders.THREAD, String.class);
if (StringUtils.hasText(threadId)) {
target.setThread(threadId);
}
String to = getHeaderIfAvailable(headers, XmppHeaders.TO, String.class);
if (StringUtils.hasText(to)) {
target.setTo(to);
}
String from = getHeaderIfAvailable(headers, XmppHeaders.FROM, String.class);
if (StringUtils.hasText(from)) {
target.setFrom(from);
}
String subject = getHeaderIfAvailable(headers, XmppHeaders.SUBJECT, String.class);
if (StringUtils.hasText(subject)) {
target.setSubject(subject);
}
Object typeHeader = getHeaderIfAvailable(headers, XmppHeaders.TYPE, Object.class);
if (typeHeader instanceof String) {
try {
typeHeader = Message.Type.valueOf((String) typeHeader);
}
catch (Exception e) {
if (logger.isWarnEnabled()) {
logger.warn("XMPP Type must be either a valid [Message.Type] " +
"enum value or a String representation of such.");
}
}
}
if (typeHeader instanceof Message.Type) {
target.setType((Message.Type) typeHeader);
}
}
@Override
protected void populateUserDefinedHeader(String headerName, Object headerValue, Message target) {
target.setProperty(headerName, headerValue);
}
@Override
protected List<String> getStandardReplyHeaderNames() {
return STANDARD_HEADER_NAMES;
}
@Override
protected List<String> getStandardRequestHeaderNames() {
return STANDARD_HEADER_NAMES;
}
@Override
protected String getStandardHeaderPrefix() {
return XmppHeaders.PREFIX;
}
}

View File

@@ -0,0 +1,32 @@
/*
* Copyright 2002-2011 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.xmpp.support;
import org.jivesoftware.smack.packet.Message;
import org.springframework.integration.mapping.HeaderMapper;
import org.springframework.integration.mapping.RequestReplyHeaderMapper;
/**
* A convenience interface that extends {@link HeaderMapper}
* but parameterized with {@link MessageProperties}.
*
* @author Mark Fisher
* @since 2.1
*/
public interface XmppHeaderMapper extends RequestReplyHeaderMapper<Message> {
}

View File

@@ -195,6 +195,22 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper">
<xsd:annotation>
<xsd:documentation>
Allows you to reference custom implementation of HeaderMapper.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of MessageHeaders to be mapped into the XMPP Headers of the request.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:complexType name="xmppOutboundAdapterType">
@@ -236,6 +252,22 @@
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="header-mapper">
<xsd:annotation>
<xsd:documentation>
Allows you to reference custom implementation of HeaderMapper.
</xsd:documentation>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="mapped-request-headers" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Comma-separated list of names of MessageHeaders to be mapped into the XMPP Headers of the request.
This can only be provided if the 'header-mapper' reference is not being set directly. The values in
this list can also be simple patterns to be matched against the header names (e.g. "foo*" or "*foo").
]]></xsd:documentation>
</xsd:annotation>
</xsd:attribute>
</xsd:complexType>
<xsd:element name="header-enricher">

View File

@@ -22,9 +22,13 @@
<beans:constructor-arg value="org.jivesoftware.smack.XMPPConnection"/>
</beans:bean>
<channel id="xmppInbound"/>
<channel id="xmppInbound">
<queue/>
</channel>
<xmpp:inbound-channel-adapter id="xmppInboundAdapter" channel="xmppInbound"
xmpp-connection="testConnection" extract-payload="false" auto-startup="false" error-channel="errorChannel"/>
xmpp-connection="testConnection" extract-payload="false"
auto-startup="false" error-channel="errorChannel"
mapped-request-headers="foo*, xmpp*"/>
</beans:beans>

View File

@@ -16,21 +16,29 @@
package org.springframework.integration.xmpp.config;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertFalse;
import java.lang.reflect.Field;
import org.jivesoftware.smack.Chat;
import org.jivesoftware.smack.ChatManager;
import org.jivesoftware.smack.PacketListener;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.packet.Message;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.integration.MessageChannel;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.xmpp.inbound.ChatMessageListeningEndpoint;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import org.springframework.util.ReflectionUtils;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertFalse;
/**
* @author Oleg Zhurakousky
@@ -42,6 +50,9 @@ public class ChatMessageInboundChannelAdapterParserTests {
@Autowired
private ApplicationContext context;
@Autowired
private QueueChannel xmppInbound;
@Test
public void testInboundAdapter(){
@@ -49,10 +60,37 @@ public class ChatMessageInboundChannelAdapterParserTests {
MessageChannel errorChannel = (MessageChannel) TestUtils.getPropertyValue(adapter, "errorChannel");
assertEquals(context.getBean("errorChannel"), errorChannel);
assertFalse(adapter.isAutoStartup());
DirectChannel channel = (DirectChannel) TestUtils.getPropertyValue(adapter, "outputChannel");
QueueChannel channel = (QueueChannel) TestUtils.getPropertyValue(adapter, "outputChannel");
assertEquals("xmppInbound", channel.getComponentName());
XMPPConnection connection = (XMPPConnection)TestUtils.getPropertyValue(adapter, "xmppConnection");
assertEquals(connection, context.getBean("testConnection"));
}
@Test
public void testInboundAdapterUsageWithHeaderMapper() {
XMPPConnection xmppConnection = Mockito.mock(XMPPConnection.class);
ChatManager chatManager = Mockito.mock(ChatManager.class);
Mockito.when(xmppConnection.getChatManager()).thenReturn(chatManager);
Chat chat = Mockito.mock(Chat.class);
Mockito.when(chatManager.getThreadChat(Mockito.any(String.class))).thenReturn(chat);
ChatMessageListeningEndpoint adapter = context.getBean("xmppInboundAdapter", ChatMessageListeningEndpoint.class);
Field xmppConnectionField = ReflectionUtils.findField(ChatMessageListeningEndpoint.class, "xmppConnection");
xmppConnectionField.setAccessible(true);
ReflectionUtils.setField(xmppConnectionField, adapter, xmppConnection);
PacketListener packetListener = TestUtils.getPropertyValue(adapter, "packetListener", PacketListener.class);
Message message = new Message();
message.setBody("hello");
message.setTo("oleg");
message.setProperty("foo", "foo");
message.setProperty("bar", "bar");
packetListener.processPacket(message);
org.springframework.integration.Message<?> siMessage = xmppInbound.receive(0);
assertEquals("foo", siMessage.getHeaders().get("foo"));
assertEquals("oleg", siMessage.getHeaders().get("xmpp_to"));
}
}

View File

@@ -15,21 +15,32 @@
<int-xmpp:outbound-channel-adapter id="outboundEventAdapter"
channel="outboundEventChannel"
xmpp-connection="testConnection"/>
xmpp-connection="testConnection"
mapped-request-headers="foo*, bar*"/>
<int:channel id="outboundPollingChannel">
<int:queue/>
</int:channel>
<int-xmpp:outbound-channel-adapter id="outboundPollingAdapter"
<int-xmpp:outbound-channel-adapter id="pollingConsumer"
channel="outboundPollingChannel"
xmpp-connection="testConnection">
<int:poller fixed-rate="1000" max-messages-per-poll="1"/>
<int:poller fixed-rate="5000" max-messages-per-poll="1"/>
</int-xmpp:outbound-channel-adapter>
<int-xmpp:outbound-channel-adapter id="withHeaderMapper"
channel="outboundPollingChannel"
xmpp-connection="testConnection"
header-mapper="headerMapper">
<int:poller fixed-rate="5000" max-messages-per-poll="1"/>
</int-xmpp:outbound-channel-adapter>
<bean id="headerMapper" class="org.springframework.integration.xmpp.support.DefaultXmppHeaderMapper">
<property name="requestHeaderNames" value="foo*"/>
</bean>
<int-xmpp:outbound-channel-adapter id="outboundNoChannelAdapter"
xmpp-connection="testConnection">
<!-- <int:poller fixed-rate="1000" max-messages-per-poll="1"/>-->
</int-xmpp:outbound-channel-adapter>
</beans>

View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2010 the original author or authors.
* Copyright 2002-2011 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,15 +16,14 @@
package org.springframework.integration.xmpp.config;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertTrue;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.util.List;
import org.jivesoftware.smack.XMPPConnection;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
@@ -37,9 +36,16 @@ import org.springframework.integration.endpoint.PollingConsumer;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.test.util.TestUtils;
import org.springframework.integration.xmpp.XmppHeaders;
import org.springframework.integration.xmpp.support.DefaultXmppHeaderMapper;
import org.springframework.integration.xmpp.support.XmppHeaderMapper;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
import static junit.framework.Assert.assertEquals;
import static junit.framework.Assert.assertTrue;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
/**
* @author Oleg Zhurakousky
* @author Mark Fisher
@@ -50,10 +56,13 @@ public class ChatMessageOutboundChannelAdapterParserTests {
@Autowired
private ApplicationContext context;
@Autowired
private XmppHeaderMapper headerMapper;
@Test
public void testPollingConsumer() {
Object pollingConsumer = context.getBean("outboundPollingAdapter");
Object pollingConsumer = context.getBean("withHeaderMapper");
QueueChannel channel = (QueueChannel) TestUtils.getPropertyValue(pollingConsumer, "inputChannel");
assertEquals("outboundPollingChannel", channel.getComponentName());
assertTrue(pollingConsumer instanceof PollingConsumer);
@@ -65,20 +74,43 @@ public class ChatMessageOutboundChannelAdapterParserTests {
assertTrue(eventConsumer instanceof SubscribableChannel);
}
@SuppressWarnings("unchecked")
@Test
public void testEventConsumer() {
Object eventConsumer = context.getBean("outboundEventAdapter");
DefaultXmppHeaderMapper headerMapper =
TestUtils.getPropertyValue(eventConsumer, "handler.headerMapper", DefaultXmppHeaderMapper.class);
List<String> requestHeaderNames = TestUtils.getPropertyValue(headerMapper, "requestHeaderNames", List.class);
assertEquals(2, requestHeaderNames.size());
assertEquals("foo*", requestHeaderNames.get(0));
assertEquals("bar*", requestHeaderNames.get(1));
assertTrue(eventConsumer instanceof EventDrivenConsumer);
}
@SuppressWarnings("rawtypes")
@Test
public void testPollingConsumerUsage() throws Exception{
Object pollingConsumer = context.getBean("outboundPollingAdapter");
public void withHeaderMapper() throws Exception{
Object pollingConsumer = context.getBean("withHeaderMapper");
assertTrue(pollingConsumer instanceof PollingConsumer);
assertEquals(headerMapper, TestUtils.getPropertyValue(pollingConsumer, "handler.headerMapper"));
MessageChannel channel = context.getBean("outboundEventChannel", MessageChannel.class);
Message<?> message = MessageBuilder.withPayload("hello").setHeader(XmppHeaders.CHAT_TO, "oleg").build();
Message<?> message = MessageBuilder.withPayload("hello").setHeader(XmppHeaders.TO, "oleg").
setHeader("foobar", "foobar").build();
XMPPConnection connection = context.getBean("testConnection", XMPPConnection.class);
Mockito.doAnswer(new Answer() {
public Object answer(InvocationOnMock invocation) {
Object[] args = invocation.getArguments();
org.jivesoftware.smack.packet.Message xmppMessage = (org.jivesoftware.smack.packet.Message) args[0];
assertEquals("oleg", xmppMessage.getTo());
assertEquals("foobar", xmppMessage.getProperty("foobar"));
assertEquals("oleg", xmppMessage.getTo());
return null;
}})
.when(connection).sendPacket(Mockito.any(org.jivesoftware.smack.packet.Message.class));
channel.send(message);
verify(connection, times(1)).sendPacket(Mockito.any(org.jivesoftware.smack.packet.Message.class));
}

View File

@@ -0,0 +1,144 @@
/*
* Copyright 2002-2011 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.xmpp.support;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import java.util.HashMap;
import java.util.Map;
import org.jivesoftware.smack.packet.Message;
import org.junit.Test;
import org.springframework.integration.MessageHeaders;
import org.springframework.integration.xmpp.XmppHeaders;
/**
* @author Mark Fisher
* @since 2.1
*/
public class DefaultXmppHeaderMapperTests {
@Test
public void fromHeadersStandardOutbound() {
DefaultXmppHeaderMapper mapper = new DefaultXmppHeaderMapper();
Map<String, Object> headerMap = new HashMap<String, Object>();
headerMap.put("userDefined1", "foo");
headerMap.put("userDefined2", "bar");
headerMap.put(XmppHeaders.THREAD, "test.thread");
headerMap.put(XmppHeaders.TO, "test.to");
headerMap.put(XmppHeaders.FROM, "test.from");
headerMap.put(XmppHeaders.SUBJECT, "test.subject");
headerMap.put(XmppHeaders.TYPE, "headline");
MessageHeaders headers = new MessageHeaders(headerMap);
Message target = new Message();
mapper.fromHeadersToRequest(headers, target);
// "standard" XMPP headers
assertEquals("test.thread", target.getThread());
assertEquals("test.to", target.getTo());
assertEquals("test.from", target.getFrom());
assertEquals("test.subject", target.getSubject());
assertEquals(Message.Type.headline, target.getType());
// user-defined headers not included by default
assertNull(target.getProperty("userDefined1"));
assertNull(target.getProperty("userDefined2"));
// transient headers should not be copied
assertNull(target.getProperty("id"));
assertNull(target.getProperty("timestamp"));
}
@Test
public void fromHeadersUserDefinedOnly() {
DefaultXmppHeaderMapper mapper = new DefaultXmppHeaderMapper();
mapper.setRequestHeaderNames(new String[] { "userDefined1", "userDefined2" });
Map<String, Object> headerMap = new HashMap<String, Object>();
headerMap.put("userDefined1", "foo");
headerMap.put("userDefined2", "bar");
headerMap.put("userDefined3", "baz");
headerMap.put(XmppHeaders.THREAD, "test.thread");
headerMap.put(XmppHeaders.TO, "test.to");
headerMap.put(XmppHeaders.FROM, "test.from");
headerMap.put(XmppHeaders.SUBJECT, "test.subject");
headerMap.put(XmppHeaders.TYPE, "headline");
MessageHeaders headers = new MessageHeaders(headerMap);
Message target = new Message();
mapper.fromHeadersToRequest(headers, target);
// "standard" XMPP headers not included
assertNull(target.getThread());
assertNull(target.getTo());
assertNull(target.getFrom());
assertNull(target.getSubject());
assertEquals(Message.Type.normal, target.getType());
// user-defined headers are included if in the list
assertEquals("foo", target.getProperty("userDefined1"));
assertEquals("bar", target.getProperty("userDefined2"));
// user-defined headers are not included if not in the list
assertNull(target.getProperty("userDefined3"));
// transient headers should not be copied
assertNull(target.getProperty("id"));
assertNull(target.getProperty("timestamp"));
}
@Test
public void toHeadersStandardOnly() {
DefaultXmppHeaderMapper mapper = new DefaultXmppHeaderMapper();
Message source = new Message("test.to", Message.Type.headline);
source.setFrom("test.from");
source.setSubject("test.subject");
source.setThread("test.thread");
source.setProperty("userDefined1", "foo");
source.setProperty("userDefined2", "bar");
Map<String, Object> headers = mapper.toHeadersFromRequest(source);
assertEquals("test.to", headers.get(XmppHeaders.TO));
assertEquals("test.from", headers.get(XmppHeaders.FROM));
assertEquals("test.subject", headers.get(XmppHeaders.SUBJECT));
assertEquals("test.thread", headers.get(XmppHeaders.THREAD));
assertEquals(Message.Type.headline, headers.get(XmppHeaders.TYPE));
assertNull(headers.get("userDefined1"));
assertNull(headers.get("userDefined2"));
}
@Test
public void toHeadersUserDefinedOnly() {
DefaultXmppHeaderMapper mapper = new DefaultXmppHeaderMapper();
mapper.setReplyHeaderNames(new String[] { "userDefined*" });
Message source = new Message("test.to", Message.Type.headline);
source.setFrom("test.from");
source.setSubject("test.subject");
source.setThread("test.thread");
source.setProperty("userDefined1", "foo");
source.setProperty("userDefined2", "bar");
Map<String, Object> headers = mapper.toHeadersFromReply(source);
assertNull(headers.get(XmppHeaders.TO));
assertNull(headers.get(XmppHeaders.FROM));
assertNull(headers.get(XmppHeaders.SUBJECT));
assertNull(headers.get(XmppHeaders.THREAD));
assertNull(headers.get(XmppHeaders.TYPE));
assertEquals("foo", headers.get("userDefined1"));
assertEquals("bar", headers.get("userDefined2"));
}
}