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

Added MessageHeaders and MessageBuilder. Messages are now immutable (including header maps).

Browse Source
This commit is contained in:
Mark Fisher
2008-07-17 22:58:56 +00:00
parent b9ea75ea98
commit 2c95306f63
91 changed files with 1154 additions and 1083 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
org.springframework.integration/src/main/java/org/springframework/integration/channel/PriorityChannel.java
View File

@@ -22,7 +22,6 @@ import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHeader;
import org.springframework.integration.message.MessagePriority;
/**
@@ -90,8 +89,10 @@ public class PriorityChannel extends QueueChannel {
private static class MessagePriorityComparator implements Comparator<Message<?>> {
public int compare(Message<?> message1, Message<?> message2) {
MessagePriority priority1 = message1.getHeader().getPriority();
MessagePriority priority2 = message2.getHeader().getPriority();
MessagePriority priority1 = message1.getHeaders().getPriority();
MessagePriority priority2 = message2.getHeaders().getPriority();
priority1 = priority1 != null ? priority1 : MessagePriority.NORMAL;
priority2 = priority2 != null ? priority2 : MessagePriority.NORMAL;
return priority1.compareTo(priority2);
}
}

7
org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java
View File

@@ -25,8 +25,8 @@ import org.apache.commons.logging.LogFactory;
import org.springframework.context.Lifecycle;
import org.springframework.integration.channel.ChannelInterceptor;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.selector.MessageSelector;
import org.springframework.util.Assert;
@@ -100,8 +100,9 @@ public class WireTap extends ChannelInterceptorAdapter implements Lifecycle {
@Override
public Message<?> preSend(Message<?> message, MessageChannel channel) {
if (this.running && this.selectorsAccept(message)) {
Message<?> duplicate = new GenericMessage<Object>(message.getPayload(), message.getHeader());
duplicate.getHeader().setAttribute(ORIGINAL_MESSAGE_ID_KEY, message.getId());
Message<?> duplicate = MessageBuilder.fromMessage(message)
.setHeader(ORIGINAL_MESSAGE_ID_KEY, message.getId())
.build();
if (!this.secondaryChannel.send(duplicate, 0)) {
if (logger.isWarnEnabled()) {
logger.warn("Failed to send message to secondary channel '" + this.secondaryChannel.getName()

32
org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java
View File

@@ -22,9 +22,10 @@ import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.ReplyHandler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageHeader;
import org.springframework.integration.message.MessageHeaders;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -82,9 +83,9 @@ public class HandlerEndpoint extends AbstractEndpoint {
super.initialize();
}
private MessageChannel resolveReplyChannel(MessageHeader originalMessageHeader) {
private MessageChannel resolveReplyChannel(Message<?> originalMessage) {
if (this.returnAddressOverrides) {
MessageChannel channel = this.getReturnAddress(originalMessageHeader);
MessageChannel channel = this.getReturnAddress(originalMessage);
if (channel == null) {
channel = this.getOutputChannel();
}
@@ -93,14 +94,14 @@ public class HandlerEndpoint extends AbstractEndpoint {
else {
MessageChannel channel = this.getOutputChannel();
if (channel == null) {
channel = this.getReturnAddress(originalMessageHeader);
channel = this.getReturnAddress(originalMessage);
}
return channel;
}
}
private MessageChannel getReturnAddress(MessageHeader originalMessageHeader) {
Object returnAddress = originalMessageHeader.getReturnAddress();
private MessageChannel getReturnAddress(Message<?> originalMessage) {
Object returnAddress = originalMessage.getHeaders().getReturnAddress();
if (returnAddress != null) {
if (returnAddress instanceof MessageChannel) {
return (MessageChannel) returnAddress;
@@ -120,10 +121,12 @@ public class HandlerEndpoint extends AbstractEndpoint {
protected Message<?> handleMessage(Message<?> message) {
Message<?> replyMessage = this.handler.handle(message);
if (replyMessage != null) {
if (replyMessage.getHeader().getCorrelationId() == null) {
replyMessage.getHeader().setCorrelationId(message.getId());
Object correlationId = replyMessage.getHeaders().getCorrelationId();
if (correlationId == null) {
replyMessage = MessageBuilder.fromMessage(replyMessage)
.setHeader(MessageHeaders.CORRELATION_ID, message.getId()).build();
}
this.replyHandler.handle(replyMessage, message.getHeader());
this.replyHandler.handle(replyMessage, message);
}
return null;
}
@@ -131,14 +134,17 @@ public class HandlerEndpoint extends AbstractEndpoint {
private class EndpointReplyHandler implements ReplyHandler {
public void handle(Message<?> replyMessage, MessageHeader originalMessageHeader) {
public void handle(Message<?> replyMessage, Message<?> originalMessage) {
if (replyMessage == null) {
return;
}
MessageChannel replyChannel = resolveReplyChannel(originalMessageHeader);
MessageChannel replyChannel = resolveReplyChannel(replyMessage);
if (replyChannel == null) {
throw new MessageHandlingException(replyMessage, "Unable to determine reply channel for message. " +
"Provide an 'outputChannelName' on the message endpoint or a 'returnAddress' in the message header");
replyChannel = resolveReplyChannel(originalMessage);
if (replyChannel == null) {
throw new MessageHandlingException(replyMessage, "Unable to determine reply channel for message. " +
"Provide an 'outputChannelName' on the message endpoint or a 'returnAddress' in the message header");
}
}
if (logger.isDebugEnabled()) {
logger.debug("endpoint '" + HandlerEndpoint.this + "' replying to channel '" + replyChannel + "' with message: " + replyMessage);

17
org.springframework.integration/src/main/java/org/springframework/integration/gateway/RequestReplyTemplate.java
View File

@@ -28,7 +28,9 @@ import org.springframework.integration.endpoint.HandlerEndpoint;
import org.springframework.integration.handler.ReplyHandler;
import org.springframework.integration.handler.ReplyMessageCorrelator;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageDeliveryException;
import org.springframework.integration.message.MessageHeaders;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.message.selector.MessageSelector;
@@ -157,8 +159,9 @@ public class RequestReplyTemplate implements MessageBusAware {
*/
public boolean request(Message<?> message, ReplyHandler replyHandler) {
MessageChannel replyChannelAdapter = new ReplyHandlingChannelAdapter(message, replyHandler);
message.getHeader().setReturnAddress(replyChannelAdapter);
return this.send(message);
Message<?> requestMessage = MessageBuilder.fromMessage(message)
.setReturnAddress(replyChannelAdapter).build();
return this.send(requestMessage);
}
/**
@@ -186,7 +189,8 @@ public class RequestReplyTemplate implements MessageBusAware {
if (this.replyMessageCorrelator == null) {
this.registerReplyMessageCorrelator();
}
message.getHeader().setReturnAddress(this.replyChannel);
message = MessageBuilder.fromMessage(message)
.setReturnAddress(this.replyChannel).build();
this.send(message);
return (this.replyTimeout >= 0) ? this.replyMessageCorrelator.getReply(message.getId(), this.replyTimeout) :
this.replyMessageCorrelator.getReply(message.getId());
@@ -194,8 +198,9 @@ public class RequestReplyTemplate implements MessageBusAware {
private Message<?> sendAndReceiveWithTemporaryChannel(Message<?> message) {
RendezvousChannel temporaryChannel = new RendezvousChannel();
message.getHeader().setReturnAddress(temporaryChannel);
this.send(message);
Message<?> requestMessage = MessageBuilder.fromMessage(message)
.setReturnAddress(temporaryChannel).build();
this.send(requestMessage);
return this.receiveResponse(temporaryChannel);
}
@@ -258,7 +263,7 @@ public class RequestReplyTemplate implements MessageBusAware {
}
public boolean send(Message<?> message) {
this.replyHandler.handle(message, originalMessage.getHeader());
this.replyHandler.handle(message, originalMessage);
return true;
}

15
org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java
View File

@@ -24,8 +24,10 @@ import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.message.DefaultMessageCreator;
import org.springframework.integration.message.DefaultMessageMapper;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageCreator;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageHeaders;
import org.springframework.integration.message.MessageMapper;
import org.springframework.integration.util.AbstractMethodInvokingAdapter;
import org.springframework.util.Assert;
@@ -124,16 +126,11 @@ public abstract class AbstractMessageHandlerAdapter extends AbstractMethodInvoki
protected Message<?> createReplyMessage(Object returnValue, Message<?> originalMessage) {
Message<?> reply = this.messageCreator.createMessage(returnValue);
if (reply != null) {
reply.copyHeader(originalMessage.getHeader(), false);
Object correlationId = reply.getHeader().getCorrelationId();
if (correlationId == null) {
Object orginalCorrelationId = originalMessage.getHeader().getCorrelationId();
reply.getHeader().setCorrelationId((orginalCorrelationId != null) ?
orginalCorrelationId : originalMessage.getId());
}
if (reply == null) {
return null;
}
return reply;
return MessageBuilder.fromMessage(reply).copyHeadersIfAbsent(originalMessage.getHeaders())
.setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, originalMessage.getId()).build();
}
/**

3
org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyHandler.java
View File

@@ -17,7 +17,6 @@
package org.springframework.integration.handler;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHeader;
/**
* Strategy interface for handling reply messages.
@@ -26,6 +25,6 @@ import org.springframework.integration.message.MessageHeader;
*/
public interface ReplyHandler {
void handle(Message<?> replyMessage, MessageHeader originalMessageHeader);
void handle(Message<?> replyMessage, Message<?> originalMessage);
}

2
org.springframework.integration/src/main/java/org/springframework/integration/handler/ReplyMessageCorrelator.java
View File

@@ -70,7 +70,7 @@ public class ReplyMessageCorrelator implements MessageHandler {
* returns the 'correlationId' from the message header.
*/
protected Object getCorrelationId(final Message<?> message) {
return message.getHeader().getCorrelationId();
return message.getHeaders().getCorrelationId();
}
}

34
org.springframework.integration/src/main/java/org/springframework/integration/handler/annotation/AnnotationMethodMessageMapper.java
View File

@@ -17,10 +17,8 @@
package org.springframework.integration.handler.annotation;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import org.springframework.core.GenericTypeResolver;
import org.springframework.core.LocalVariableTableParameterNameDiscoverer;
@@ -29,7 +27,7 @@ import org.springframework.core.ParameterNameDiscoverer;
import org.springframework.integration.ConfigurationException;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageHandlingException;
import org.springframework.integration.message.MessageHeader;
import org.springframework.integration.message.MessageHeaders;
import org.springframework.integration.message.MessageMapper;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
@@ -124,7 +122,7 @@ public class AnnotationMethodMessageMapper implements MessageMapper {
MethodParameterMetadata metadata = this.parameterMetadata[i];
Class<?> expectedType = metadata.type;
if (expectedType.equals(HeaderAttribute.class)) {
Object value = message.getHeader().getAttribute(metadata.key);
Object value = message.getHeaders().get(metadata.key);
if (value == null && metadata.required) {
throw new MessageHandlingException(message,
"required attribute '" + metadata.key + "' not available");
@@ -132,7 +130,7 @@ public class AnnotationMethodMessageMapper implements MessageMapper {
args[i] = value;
}
else if (expectedType.equals(HeaderProperty.class)) {
Object value = message.getHeader().getProperty(metadata.key);
Object value = message.getHeaders().get(metadata.key);
if (value == null && metadata.required) {
throw new MessageHandlingException(message,
"required property '" + metadata.key + "' not available");
@@ -146,10 +144,10 @@ public class AnnotationMethodMessageMapper implements MessageMapper {
args[i] = message.getPayload();
}
else if (expectedType.equals(Map.class)) {
args[i] = this.getHeaderAttributes(message);
args[i] = message.getHeaders();
}
else if (expectedType.equals(Properties.class)) {
args[i] = this.getHeaderProperties(message);
args[i] = this.getStringTypedHeaders(message);
}
else {
args[i] = message.getPayload();
@@ -158,22 +156,14 @@ public class AnnotationMethodMessageMapper implements MessageMapper {
return args;
}
private Map<String, Object> getHeaderAttributes(Message<?> message) {
Map<String, Object> attributes = new HashMap<String, Object>();
MessageHeader header = message.getHeader();
Set<String> attributeNames = header.getAttributeNames();
for (String name : attributeNames) {
attributes.put(name, header.getAttribute(name));
}
return attributes;
}
private Properties getHeaderProperties(Message<?> message) {
private Properties getStringTypedHeaders(Message<?> message) {
Properties properties = new Properties();
MessageHeader header = message.getHeader();
Set<String> propertyNames = header.getPropertyNames();
for (String name : propertyNames) {
properties.setProperty(name, header.getProperty(name));
MessageHeaders headers = message.getHeaders();
for (String key : headers.keySet()) {
Object value = headers.get(key);
if (value instanceof String) {
properties.setProperty(key, (String) value);
}
}
return properties;
}

162
org.springframework.integration/src/main/java/org/springframework/integration/message/DefaultMessageHeader.java
View File

@@ -1,162 +0,0 @@
/*
* Copyright 2002-2008 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.message;
import java.io.Serializable;
import java.util.Date;
import java.util.HashSet;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* The default implementation of the {@link MessageHeader} interface.
*
* @author Mark Fisher
*/
public class DefaultMessageHeader implements MessageHeader, Serializable {
private final long timestamp = System.currentTimeMillis();
private volatile Date expiration;
private volatile Object correlationId;
private transient volatile Object returnAddress;
private volatile int sequenceNumber = 1;
private volatile int sequenceSize = 1;
private volatile MessagePriority priority = MessagePriority.NORMAL;
private final Properties properties = new Properties();
private final ConcurrentMap<String, Object> attributes = new ConcurrentHashMap<String, Object>();
/**
* Return the creation time of this message (in milliseconds).
*/
public long getTimestamp() {
return this.timestamp;
}
/**
* Return the expiration date for this message or <code>null</code> to
* indicate 'never expire'.
*/
public Date getExpiration() {
return this.expiration;
}
/**
* Set the expiration date for this message or <code>null</code> to
* indicate 'never expire'. The default is <code>null</code>.
*/
public void setExpiration(Date expiration) {
this.expiration = expiration;
}
public Object getCorrelationId() {
return this.correlationId;
}
public void setCorrelationId(Object correlationId) {
this.correlationId = correlationId;
}
public Object getReturnAddress() {
return this.returnAddress;
}
public void setReturnAddress(Object returnAddress) {
this.returnAddress = returnAddress;
}
public int getSequenceNumber() {
return this.sequenceNumber;
}
public void setSequenceNumber(int sequenceNumber) {
this.sequenceNumber = sequenceNumber;
}
public int getSequenceSize() {
return this.sequenceSize;
}
public void setSequenceSize(int sequenceSize) {
this.sequenceSize = sequenceSize;
}
public MessagePriority getPriority() {
return this.priority;
}
public void setPriority(MessagePriority priority) {
this.priority = priority;
}
public String getProperty(String key) {
return this.properties.getProperty(key);
}
public String setProperty(String key, String value) {
return (String) this.properties.setProperty(key, value);
}
public String removeProperty(String key) {
return (String) this.properties.remove(key);
}
public Set<String> getPropertyNames() {
Set<String> propertyNames = new HashSet<String>();
for (Object key : this.properties.keySet()) {
propertyNames.add((String) key);
}
return propertyNames;
}
public Object getAttribute(String key) {
return this.attributes.get(key);
}
public Object setAttribute(String key, Object value) {
return this.attributes.put(key, value);
}
public Object setAttributeIfAbsent(String key, Object value) {
return this.attributes.putIfAbsent(key, value);
}
public Object removeAttribute(String key) {
return this.attributes.remove(key);
}
public Set<String> getAttributeNames() {
return this.attributes.keySet();
}
public String toString() {
return "[CorrelationID=" + this.correlationId + "][Properties=" + this.properties + "][Attributes=" + this.attributes +
"][Timestamp=" + this.timestamp + "][Expiration=" + this.expiration + "][Priority=" + this.priority +
"][Sequence #" + this.sequenceNumber + " (of " + this.sequenceSize + ")]";
}
}

6
org.springframework.integration/src/main/java/org/springframework/integration/message/ErrorMessage.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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,8 +27,4 @@ public class ErrorMessage extends GenericMessage<Throwable> {
super(payload);
}
public ErrorMessage(Object id, Throwable payload) {
super(id, payload);
}
}

106
org.springframework.integration/src/main/java/org/springframework/integration/message/GenericMessage.java
View File

@@ -16,42 +16,30 @@
package org.springframework.integration.message;
import java.util.Date;
import java.util.Set;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import org.springframework.integration.util.IdGenerator;
import org.springframework.integration.util.RandomUuidGenerator;
import org.springframework.util.Assert;
/**
* Base Message class defining common properties such as id, header, and lock.
* Base Message class defining common properties such as id, payload, and headers.
*
* @author Mark Fisher
*/
public class GenericMessage<T> implements Message<T> {
public class GenericMessage<T> implements Message<T>, Serializable {
private final Object id;
private final MessageHeader header = new DefaultMessageHeader();
private final static String ID_HEADER_KEY = "id";
private volatile T payload;
private final MessageHeaders headers;
private transient final IdGenerator defaultIdGenerator = new RandomUuidGenerator();
/**
* Create a new message with the given id and payload.
*
* @param id unique identifier for this message
* @param payload the message payload
*/
public GenericMessage(Object id, T payload) {
Assert.notNull(id, "id must not be null");
Assert.notNull(payload, "payload must not be null");
this.id = id;
this.payload = payload;
}
/**
* Create a new message with the given payload. The id will be generated by
* the default {@link IdGenerator} strategy.
@@ -59,85 +47,45 @@ public class GenericMessage<T> implements Message<T> {
* @param payload the message payload
*/
public GenericMessage(T payload) {
Assert.notNull(payload, "payload must not be null");
this.id = this.defaultIdGenerator.generateId();
this.payload = payload;
this(payload, null);
}
/**
* Create a new message with the given payload. The id will be generated by
* the default {@link IdGenerator} strategy. The header will be populated
* with the attributes and properties of the provided header.
* the default {@link IdGenerator} strategy. The headers will be populated
* with the provided header values.
*
* @param payload the message payload
* @param headerToCopy message header whose attributes and properties should
* be copied into the new message's header
* @param headers message headers
*/
public GenericMessage(T payload, MessageHeader headerToCopy) {
this(payload);
this.copyHeader(headerToCopy, true);
public GenericMessage(T payload, Map<String, Object> headers) {
Assert.notNull(payload, "payload must not be null");
this.payload = payload;
if (headers == null) {
headers = new HashMap<String, Object>();
}
else if (headers instanceof MessageHeaders) {
headers = new HashMap<String, Object>(headers);
}
headers.put(ID_HEADER_KEY, this.defaultIdGenerator.generateId());
this.headers = new MessageHeaders(headers);
}
public Object getId() {
return this.id;
return this.headers.get(ID_HEADER_KEY);
}
public MessageHeader getHeader() {
return this.header;
public MessageHeaders getHeaders() {
return this.headers;
}
public T getPayload() {
return this.payload;
}
public void setPayload(T payload) {
Assert.notNull(payload, "payload must not be null");
this.payload = payload;
}
public boolean isExpired() {
Date expiration = this.header.getExpiration();
return (expiration != null) ? expiration.getTime() < System.currentTimeMillis() : false;
}
public String toString() {
return "[ID=" + this.id + "][Header=" + this.header + "][Payload='" + this.payload + "']";
}
public void copyHeader(final MessageHeader headerToCopy, boolean overrideExistingValues) {
Set<String> propertyNames = headerToCopy.getPropertyNames();
for (String key : propertyNames) {
if (overrideExistingValues) {
this.header.setProperty(key, headerToCopy.getProperty(key));
}
else if (this.header.getProperty(key) == null) {
this.header.setProperty(key, headerToCopy.getProperty(key));
}
}
Set<String> attributeNames = headerToCopy.getAttributeNames();
for (String key : attributeNames) {
if (overrideExistingValues) {
this.header.setAttribute(key, headerToCopy.getAttribute(key));
}
else {
this.header.setAttributeIfAbsent(key, headerToCopy.getAttribute(key));
}
}
if (overrideExistingValues) {
this.header.setSequenceNumber(headerToCopy.getSequenceNumber());
this.header.setSequenceSize(headerToCopy.getSequenceSize());
this.header.setReturnAddress(headerToCopy.getReturnAddress());
}
else {
if (headerToCopy.getSequenceSize() > 1 && this.header.getSequenceSize() == 1) {
this.header.setSequenceSize(headerToCopy.getSequenceSize());
this.header.setSequenceNumber(headerToCopy.getSequenceNumber());
}
if (headerToCopy.getReturnAddress() != null && this.header.getReturnAddress() == null) {
this.header.setReturnAddress(headerToCopy.getReturnAddress());
}
}
return "[ID=" + this.getId() + "][Headers=" + this.headers + "][Payload='" + this.payload + "']";
}
}

11
org.springframework.integration/src/main/java/org/springframework/integration/message/Message.java
View File

@@ -16,23 +16,18 @@
package org.springframework.integration.message;
import java.io.Serializable;
/**
* The central interface that any Message type must implement.
*
* @author Mark Fisher
* @author Arjen Poutsma
*/
public interface Message<T> extends Serializable {
public interface Message<T> {
Object getId();
MessageHeader getHeader();
T getPayload();
boolean isExpired();
void copyHeader(MessageHeader header, boolean overwriteExistingValues);
MessageHeaders getHeaders();
}

144
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java Normal file
View File

@@ -0,0 +1,144 @@
/*
* Copyright 2002-2008 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.message;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import org.springframework.util.Assert;
/**
* @author Arjen Poutsma
* @author Mark Fisher
*/
public final class MessageBuilder<T> {
private final T payload;
private final Map<String, Object> headers = new HashMap<String, Object>();
/**
* Create a new {@link Message} instance with no header values using
* the provided payload instance.
*/
private MessageBuilder(T payload) {
Assert.notNull(payload, "payload must not be null");
this.payload = payload;
}
/**
* Create a builder for a new {@link Message} instance pre-populated with
* all of the headers copied from the provided message. The payload will
* also be taken from the provided message.
*
* @param messageToCopy the Message from which all headers should be copied
*/
public static <T> MessageBuilder<T> fromMessage(Message<T> message) {
MessageBuilder<T> builder = new MessageBuilder<T>(message.getPayload());
builder.headers.putAll(message.getHeaders());
return builder;
}
/**
* Create a builder for a new {@link Message} instance with no header
* values using the provided payload instance.
*
* @param payload the payload for the new message
*/
public static <T> MessageBuilder<T> fromPayload(T payload) {
MessageBuilder<T> builder = new MessageBuilder<T>(payload);
return builder;
}
public MessageBuilder<T> setHeader(String headerName, Object headerValue) {
this.headers.put(headerName, headerValue);
return this;
}
public MessageBuilder<T> setHeaderIfAbsent(String headerName, Object headerValue) {
if (this.headers.get(headerName) == null) {
this.headers.put(headerName, headerValue);
}
return this;
}
public MessageBuilder<T> copyHeadersFromMessage(Message<?> message) {
return this.copyHeaders(message.getHeaders());
}
public MessageBuilder<T> copyHeaders(MessageHeaders headersToCopy) {
Set<String> keys = headersToCopy.keySet();
for (String key : keys) {
if (key.equals(MessageHeaders.TIMESTAMP)) {
continue;
}
this.setHeader(key, headersToCopy.get(key));
}
return this;
}
public MessageBuilder<T> copyHeadersFromMessageIfAbsent(Message<?> message) {
return this.copyHeadersIfAbsent(message.getHeaders());
}
public MessageBuilder<T> copyHeadersIfAbsent(MessageHeaders headersToCopy) {
Set<String> keys = headersToCopy.keySet();
for (String key : keys) {
if (key.equals(MessageHeaders.TIMESTAMP)) {
continue;
}
if (this.headers.get(key) == null) {
this.setHeaderIfAbsent(key, headersToCopy.get(key));
}
}
return this;
}
public MessageBuilder<T> setExpirationDate(Date expirationDate) {
return this.setHeader(MessageHeaders.EXPIRATION_DATE, expirationDate);
}
public MessageBuilder<T> setCorrelationId(Object correlationId) {
return this.setHeader(MessageHeaders.CORRELATION_ID, correlationId);
}
public MessageBuilder<T> setReturnAddress(Object returnAddress) {
return this.setHeader(MessageHeaders.RETURN_ADDRESS, returnAddress);
}
public MessageBuilder<T> setSequenceNumber(Integer sequenceNumber) {
return this.setHeader(MessageHeaders.SEQUENCE_NUMBER, sequenceNumber);
}
public MessageBuilder<T> setSequenceSize(Integer sequenceSize) {
return this.setHeader(MessageHeaders.SEQUENCE_SIZE, sequenceSize);
}
public MessageBuilder<T> setPriority(MessagePriority priority) {
return this.setHeader(MessageHeaders.PRIORITY, priority);
}
public Message<T> build() {
return new GenericMessage<T>(this.payload, this.headers);
}
}

77
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeader.java
View File

@@ -1,77 +0,0 @@
/*
* Copyright 2002-2008 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.message;
import java.util.Date;
import java.util.Set;
/**
* A holder for Message metadata. This includes information that may be used by
* the messaging system (such as <i>correlationId</i>) as well as information
* that is relevant for specific messaging endpoints. For the latter, String
* values may be stored as <i>properties</i> and Object values may be stored as
* <i>attributes</i>.
*
* @author Mark Fisher
*/
public interface MessageHeader {
long getTimestamp();
Date getExpiration();
void setExpiration(Date expiration);
Object getCorrelationId();
void setCorrelationId(Object correlationId);
Object getReturnAddress();
void setReturnAddress(Object returnAddress);
int getSequenceNumber();
void setSequenceNumber(int sequenceNumber);
int getSequenceSize();
void setSequenceSize(int sequenceSize);
MessagePriority getPriority();
void setPriority(MessagePriority priority);
String getProperty(String key);
String setProperty(String key, String value);
String removeProperty(String key);
Set<String> getPropertyNames();
Object getAttribute(String key);
Object setAttribute(String key, Object value);
Object setAttributeIfAbsent(String key, Object value);
Object removeAttribute(String key);
Set<String> getAttributeNames();
}

169
org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java Normal file
View File

@@ -0,0 +1,169 @@
/*
* Copyright 2002-2008 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.message;
import java.io.Serializable;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
/**
* The headers for a {@link Message}.
*
* @author Arjen Poutsma
* @author Mark Fisher
*/
public final class MessageHeaders implements Map<String, Object>, Serializable {
public static final String TIMESTAMP = "internal.header.timestamp";
public static final String CORRELATION_ID = "internal.header.correlationId";
public static final String RETURN_ADDRESS = "internal.header.returnAddress";
public static final String EXPIRATION_DATE = "internal.header.exprirationDate";
public static final String PRIORITY = "internal.header.priority";
public static final String SEQUENCE_NUMBER = "internal.header.sequenceNumber";
public static final String SEQUENCE_SIZE = "internal.header.sequenceSize";
private final Map<String, Object> headers;
public MessageHeaders(Map<String, Object> headers) {
this.headers = (headers != null ? headers : new HashMap<String, Object>());
this.headers.put(TIMESTAMP, new Long(System.currentTimeMillis()));
}
public Long getTimestamp() {
return this.get(TIMESTAMP, Long.class);
}
public Date getExpirationDate() {
return this.get(EXPIRATION_DATE, Date.class);
}
public Object getCorrelationId() {
return this.get(CORRELATION_ID);
}
public Object getReturnAddress() {
return this.get(RETURN_ADDRESS);
}
public Integer getSequenceNumber() {
Integer sequenceNumber = this.get(SEQUENCE_NUMBER, Integer.class);
return (sequenceNumber != null ? sequenceNumber : 0);
}
public Integer getSequenceSize() {
Integer sequenceSize = this.get(SEQUENCE_SIZE, Integer.class);
return (sequenceSize != null ? sequenceSize : 0);
}
public MessagePriority getPriority() {
return this.get(PRIORITY, MessagePriority.class);
}
public void clear() {
this.headers.clear();
}
public boolean containsKey(Object key) {
return this.headers.containsKey(key);
}
public boolean containsValue(Object value) {
return this.headers.containsValue(value);
}
public Set<Map.Entry<String, Object>> entrySet() {
return Collections.unmodifiableSet(this.headers.entrySet());
}
@SuppressWarnings("unchecked")
public <T> T get(Object key, Class<T> type) {
Object value = this.headers.get(key);
if (value == null) {
return null;
}
if (!type.isAssignableFrom(value.getClass())) {
throw new MessagingException("Type mismatch for header '" + key + "'. Expected ["
+ type + "] but actual type is [" + value.getClass() + "]");
}
return (T) value;
}
public Object get(Object key) {
return this.headers.get(key);
}
public boolean isEmpty() {
return this.headers.isEmpty();
}
public Set<String> keySet() {
return Collections.unmodifiableSet(this.headers.keySet());
}
public Object put(String key, Object value) {
throw new UnsupportedOperationException("MessageHeaders is immutable.");
}
public void putAll(Map<? extends String, ? extends Object> t) {
throw new UnsupportedOperationException("MessageHeaders is immutable.");
}
public Object remove(Object key) {
throw new UnsupportedOperationException("MessageHeaders is immutable.");
}
public int size() {
return this.headers.size();
}
public Collection<Object> values() {
return Collections.unmodifiableCollection(this.headers.values());
}
public int hashCode() {
return headers.hashCode();
}
public boolean equals(Object obj) {
if (this == obj) {
return true;
}
if (obj != null && obj instanceof MessageHeaders) {
MessageHeaders other = (MessageHeaders) obj;
return this.headers.equals(other.headers);
}
return false;
}
public String toString() {
return headers.toString();
}
}

6
org.springframework.integration/src/main/java/org/springframework/integration/message/StringMessage.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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,8 +27,4 @@ public class StringMessage extends GenericMessage<String> {
super(payload);
}
public StringMessage(Object id, String payload) {
super(id, payload);
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/message/selector/MessageSelector.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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.

10
org.springframework.integration/src/main/java/org/springframework/integration/message/selector/UnexpiredMessageSelector.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2002-2007 the original author or authors.
* Copyright 2002-2008 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,6 +16,8 @@
package org.springframework.integration.message.selector;
import java.util.Date;
import org.springframework.integration.message.Message;
/**
@@ -27,7 +29,11 @@ import org.springframework.integration.message.Message;
public class UnexpiredMessageSelector implements MessageSelector {
public boolean accept(Message<?> message) {
return (!message.isExpired());
Date expirationDate = message.getHeaders().getExpirationDate();
if (expirationDate == null) {
return true;
}
return expirationDate.getTime() > System.currentTimeMillis();
}
}

5
org.springframework.integration/src/main/java/org/springframework/integration/router/AbstractMessageBarrierHandler.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.router;
import java.util.List;
@@ -168,7 +169,7 @@ public abstract class AbstractMessageBarrierHandler implements MessageHandler, I
if (!this.initialized) {
this.afterPropertiesSet();
}
Object correlationId = message.getHeader().getCorrelationId();
Object correlationId = message.getHeaders().getCorrelationId();
if (correlationId == null) {
throw new MessageHandlingException(message,
this.getClass().getSimpleName() + " requires the 'correlationId' property");
@@ -226,7 +227,7 @@ public abstract class AbstractMessageBarrierHandler implements MessageHandler, I
}
protected MessageChannel resolveReplyChannelFromMessage(Message<?> message) {
Object returnAddress = message.getHeader().getReturnAddress();
Object returnAddress = message.getHeaders().getReturnAddress();
if (returnAddress != null) {
if (returnAddress instanceof MessageChannel) {
return (MessageChannel) returnAddress;

7
org.springframework.integration/src/main/java/org/springframework/integration/router/AggregatingMessageHandler.java
View File

@@ -20,6 +20,7 @@ import java.util.List;
import java.util.concurrent.ScheduledExecutorService;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
@@ -87,11 +88,11 @@ public class AggregatingMessageHandler extends AbstractMessageBarrierHandler {
return new Message<?>[0];
}
Message<?> result = aggregator.aggregate(messages);
if (result.getHeader().getCorrelationId() == null) {
result.getHeader().setCorrelationId(correlationId);
if (result.getHeaders().getCorrelationId() == null) {
result = MessageBuilder.fromMessage(result)
.setCorrelationId(correlationId).build();
}
return new Message<?>[] { result };
}
}

14
org.springframework.integration/src/main/java/org/springframework/integration/router/MessageSequenceComparator.java
View File

@@ -21,7 +21,7 @@ import java.util.Comparator;
import org.springframework.integration.message.Message;
/**
* A {@link Comparator} implementation based on the '<code>sequenceNumber</code>'
* A {@link Comparator} implementation based on the 'sequence number'
* property of a {@link Message Message's} header.
*
* @author Mark Fisher
@@ -29,9 +29,15 @@ import org.springframework.integration.message.Message;
public class MessageSequenceComparator implements Comparator<Message<?>> {
public int compare(Message<?> message1, Message<?> message2) {
int s1 = message1.getHeader().getSequenceNumber();
int s2 = message2.getHeader().getSequenceNumber();
return (s1 < s2) ? -1 : (s1 == s2) ? 0 : 1;
Integer s1 = message1.getHeaders().getSequenceNumber();
Integer s2 = message2.getHeaders().getSequenceNumber();
if (s1 == null) {
s1 = 0;
}
if (s2 == null) {
s2 = 0;
}
return s1.compareTo(s2);
}
}

14
org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageBarrier.java
View File

@@ -13,6 +13,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.router;
import java.util.ArrayList;
@@ -26,6 +27,7 @@ import org.springframework.integration.message.Message;
/**
* MessageBarrier implementation for resequencing. It can either
* release partial sequences as messages arrive, or the full sequence.
*
* @author Marius Bogoevici
*/
public class ResequencingMessageBarrier extends AbstractMessageBarrier {
@@ -44,7 +46,7 @@ public class ResequencingMessageBarrier extends AbstractMessageBarrier {
public ResequencingMessageBarrier(boolean releasePartialSequences) {
this.resequencingComparator = new Comparator<Message<?>>() {
public int compare(Message<?> m1, Message<?> m2) {
return m1.getHeader().getSequenceNumber() - m2.getHeader().getSequenceNumber();
return m1.getHeaders().getSequenceNumber() - m2.getHeaders().getSequenceNumber();
}
};
this.releasePartialSequences = releasePartialSequences;
@@ -66,10 +68,10 @@ public class ResequencingMessageBarrier extends AbstractMessageBarrier {
//(aggregated, this means that the last possibile partial sequence of messages has been received
Message<?> firstMessage = this.messages.get(0);
Message<?> lastMessage = this.messages.get(messages.size() - 1);
return (lastMessage.getHeader().getSequenceNumber() == lastMessage.getHeader().getSequenceSize()
&& (lastMessage.getHeader().getSequenceNumber() - firstMessage.getHeader().getSequenceNumber()
return (lastMessage.getHeaders().getSequenceNumber() == lastMessage.getHeaders().getSequenceSize()
&& (lastMessage.getHeaders().getSequenceNumber() - firstMessage.getHeaders().getSequenceNumber()
== this.messages.size() - 1
&& this.lastReleasedSequenceNumber == firstMessage.getHeader().getSequenceNumber() - 1));
&& this.lastReleasedSequenceNumber == firstMessage.getHeaders().getSequenceNumber() - 1));
}
protected List<Message<?>> releaseAvailableMessages() {
@@ -78,9 +80,9 @@ public class ResequencingMessageBarrier extends AbstractMessageBarrier {
Iterator<Message<?>> it = this.messages.iterator();
while (it.hasNext()) {
Message<?> currentMessage = it.next();
if (this.lastReleasedSequenceNumber == currentMessage.getHeader().getSequenceNumber() - 1) {
if (this.lastReleasedSequenceNumber == currentMessage.getHeaders().getSequenceNumber() - 1) {
releasedMessages.add(currentMessage);
this.lastReleasedSequenceNumber = currentMessage.getHeader().getSequenceNumber();
this.lastReleasedSequenceNumber = currentMessage.getHeaders().getSequenceNumber();
it.remove();
}
else {

4
org.springframework.integration/src/main/java/org/springframework/integration/router/ResequencingMessageHandler.java
View File

@@ -62,8 +62,8 @@ public class ResequencingMessageHandler extends AbstractMessageBarrierHandler{
}
protected boolean isBarrierRemovable(Object correlationId, List<Message<?>> releasedMessages) {
return (releasedMessages.get(releasedMessages.size() - 1).getHeader().getSequenceNumber() ==
releasedMessages.get(releasedMessages.size() - 1).getHeader().getSequenceSize());
return (releasedMessages.get(releasedMessages.size() - 1).getHeaders().getSequenceNumber() ==
releasedMessages.get(releasedMessages.size() - 1).getHeaders().getSequenceSize());
}
}

2
org.springframework.integration/src/main/java/org/springframework/integration/router/SequenceSizeCompletionStrategy.java
View File

@@ -35,7 +35,7 @@ public class SequenceSizeCompletionStrategy implements CompletionStrategy {
if (CollectionUtils.isEmpty(messages)) {
return false;
}
return messages.size() != 0 && (messages.size() >= messages.get(0).getHeader().getSequenceSize());
return messages.size() != 0 && (messages.size() >= messages.get(0).getHeaders().getSequenceSize());
}
}

17
org.springframework.integration/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java
View File

@@ -26,6 +26,7 @@ import org.springframework.integration.channel.ChannelRegistryAware;
import org.springframework.integration.channel.MessageChannel;
import org.springframework.integration.handler.AbstractMessageHandlerAdapter;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
/**
* MessageHandler adapter for methods annotated with {@link Splitter @Splitter}.
@@ -83,7 +84,10 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter
for (Object item : items) {
Message<?> splitMessage = (item instanceof Message<?>) ?
(Message<?>) item : this.createReplyMessage(item, originalMessage);
this.prepareMessage(splitMessage, originalMessage.getId(), ++sequenceNumber, sequenceSize);
splitMessage = MessageBuilder.fromMessage(splitMessage)
.setCorrelationId(originalMessage.getId())
.setSequenceNumber(++sequenceNumber)
.setSequenceSize(sequenceSize).build();
this.sendMessage(splitMessage, this.outputChannelName);
}
}
@@ -94,7 +98,10 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter
for (Object item : array) {
Message<?> splitMessage = (item instanceof Message<?>) ?
(Message<?>) item : this.createReplyMessage(item, originalMessage);
this.prepareMessage(splitMessage, originalMessage.getId(), ++sequenceNumber, sequenceSize);
splitMessage = MessageBuilder.fromMessage(splitMessage)
.setCorrelationId(originalMessage.getId())
.setSequenceNumber(++sequenceNumber)
.setSequenceSize(sequenceSize).build();
this.sendMessage(splitMessage, this.outputChannelName);
}
}
@@ -105,12 +112,6 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter
return null;
}
private void prepareMessage(Message<?> message, Object correlationId, int sequenceNumber, int sequenceSize) {
message.getHeader().setCorrelationId(correlationId);
message.getHeader().setSequenceNumber(sequenceNumber);
message.getHeader().setSequenceSize(sequenceSize);
}
private boolean sendMessage(Message<?> message, String channelName) {
ChannelRegistry channelRegistry = this.getChannelRegistry();
if (channelRegistry == null) {

13
org.springframework.integration/src/main/java/org/springframework/integration/transformer/AnnotationMethodTransformerAdapter.java
View File

@@ -22,8 +22,8 @@ import java.util.Properties;
import org.springframework.integration.handler.MessageHandler;
import org.springframework.integration.handler.annotation.AnnotationMethodMessageMapper;
import org.springframework.integration.message.DefaultMessageMapper;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessageMapper;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.util.AbstractMethodInvokingAdapter;
@@ -82,25 +82,28 @@ public class AnnotationMethodTransformerAdapter extends AbstractMethodInvokingAd
}
if (result instanceof Properties && !(message.getPayload() instanceof Properties)) {
Properties propertiesToSet = (Properties) result;
MessageBuilder builder = MessageBuilder.fromMessage(message);
for (Object keyObject : propertiesToSet.keySet()) {
String key = (String) keyObject;
message.getHeader().setProperty(key, propertiesToSet.getProperty(key));
builder.setHeader(key, propertiesToSet.getProperty(key));
}
return builder.build();
}
else if (result instanceof Map && !(message.getPayload() instanceof Map)) {
Map<String, ?> attributesToSet = (Map) result;
MessageBuilder builder = MessageBuilder.fromMessage(message);
for (String key : attributesToSet.keySet()) {
message.getHeader().setAttribute(key, attributesToSet.get(key));
builder.setHeader(key, attributesToSet.get(key));
}
return builder.build();
}
else {
return new GenericMessage(result, message.getHeader());
return MessageBuilder.fromPayload(result).copyHeadersFromMessage(message).build();
}
}
catch (Exception e) {
throw new MessagingException(message, "failed to transform message payload", e);
}
return message;
}
public Message<?> handle(Message<?> message) {

4
org.springframework.integration/src/main/java/org/springframework/integration/transformer/PayloadTransformerAdapter.java
View File

@@ -16,8 +16,8 @@
package org.springframework.integration.transformer;
import org.springframework.integration.message.GenericMessage;
import org.springframework.integration.message.Message;
import org.springframework.integration.message.MessageBuilder;
import org.springframework.integration.message.MessagingException;
import org.springframework.integration.util.AbstractMethodInvokingAdapter;
@@ -30,7 +30,7 @@ public class PayloadTransformerAdapter extends AbstractMethodInvokingAdapter imp
public Message<?> transform(Message<?> message) {
try {
Object result = this.invokeMethod(message.getPayload());
return new GenericMessage(result, message.getHeader());
return MessageBuilder.fromPayload(result).copyHeadersFromMessage(message).build();
} catch (Exception e) {
throw new MessagingException(message, "failed to transform message payload", e);
}