diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHeaders.java b/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHeaders.java index c5332f2cc4..887de4bf16 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHeaders.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/core/MessageHeaders.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -70,8 +70,12 @@ public final class MessageHeaders implements Map, Serializable { this.headers = (headers != null) ? new HashMap(headers) : new HashMap(); - this.headers.put(ID, UUID.randomUUID()); - this.headers.put(TIMESTAMP, new Long(System.currentTimeMillis())); + if (this.headers.get(ID) == null) { + this.headers.put(ID, UUID.randomUUID()); + } + if (this.headers.get(TIMESTAMP) == null) { + this.headers.put(TIMESTAMP, new Long(System.currentTimeMillis())); + } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java index 1bb7de3e0e..4804499a69 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/dispatcher/BroadcastingDispatcher.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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,8 +16,11 @@ package org.springframework.integration.dispatcher; +import java.util.UUID; + import org.springframework.core.task.TaskExecutor; import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.message.MessageBuilder; import org.springframework.integration.message.MessageHandler; @@ -51,6 +54,7 @@ public class BroadcastingDispatcher extends AbstractDispatcher { : MessageBuilder.fromMessage(message) .setSequenceNumber(sequenceNumber++) .setSequenceSize(sequenceSize) + .setHeader(MessageHeaders.ID, UUID.randomUUID()) .build(); TaskExecutor executor = this.getTaskExecutor(); if (executor != null) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/GenericMessage.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/GenericMessage.java index 86ff64c665..1747caa5a2 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/GenericMessage.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/GenericMessage.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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 java.util.Map; import org.springframework.integration.core.Message; import org.springframework.integration.core.MessageHeaders; import org.springframework.util.Assert; +import org.springframework.util.ObjectUtils; /** * Base Message class defining common properties such as id, payload, and headers. @@ -79,4 +80,23 @@ public class GenericMessage implements Message, Serializable { return "[Payload=" + this.payload + "][Headers=" + this.headers + "]"; } + public int hashCode() { + return this.headers.hashCode() * 23 + ObjectUtils.nullSafeHashCode(this.payload); + } + + public boolean equals(Object obj) { + if (this == obj) { + return true; + } + if (obj != null && obj instanceof GenericMessage) { + GenericMessage other = (GenericMessage) obj; + if (!this.headers.getId().equals(other.headers.getId())) { + return false; + } + return this.headers.equals(other.headers) + && this.payload.equals(other.payload); + } + return false; + } + } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java index e94d987962..71cdcab15d 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageBuilder.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -86,7 +86,7 @@ public final class MessageBuilder { * null, the header will be removed. */ public MessageBuilder setHeader(String headerName, Object headerValue) { - if (StringUtils.hasLength(headerName) && !(this.isReadOnly(headerName))) { + if (StringUtils.hasLength(headerName)) { this.modified = true; if (headerValue == null) { this.headers.remove(headerName); @@ -201,8 +201,4 @@ public final class MessageBuilder { return new GenericMessage(this.payload, this.headers); } - private boolean isReadOnly(String key) { - return (key.equals(MessageHeaders.ID) || key.equals(MessageHeaders.TIMESTAMP)); - } - } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java b/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java index 22736f8d62..abfba6bd97 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/splitter/AbstractMessageSplitter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2008 the original author or authors. + * Copyright 2002-2009 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. @@ -17,8 +17,10 @@ package org.springframework.integration.splitter; import java.util.Collection; +import java.util.UUID; import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessageHeaders; import org.springframework.integration.handler.AbstractReplyProducingMessageHandler; import org.springframework.integration.handler.ReplyMessageHolder; @@ -61,7 +63,8 @@ public abstract class AbstractMessageSplitter extends AbstractReplyProducingMess private void addReply(ReplyMessageHolder replyHolder, Object item, Object correlationId, int sequenceNumber, int sequenceSize) { replyHolder.add(item).setCorrelationId(correlationId) .setSequenceNumber(sequenceNumber) - .setSequenceSize(sequenceSize); + .setSequenceSize(sequenceSize) + .setHeader(MessageHeaders.ID, UUID.randomUUID()); } /**