Message IDs are now maintained when creating a new Message from an existing Message or copying another Message's headers (INT-576). The AbstractMessageSplitter will generate a new ID for each split Message, and the BroadcastingDispatcher (used by Publish Subscribe Channels) will also generate a new ID if its 'applySequence' value is true.
This commit is contained in:
@@ -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<String, Object>, Serializable {
|
||||
this.headers = (headers != null)
|
||||
? new HashMap<String, Object>(headers)
|
||||
: new HashMap<String, Object>();
|
||||
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()));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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<T> implements Message<T>, 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;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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<T> {
|
||||
* <code>null</code>, the header will be removed.
|
||||
*/
|
||||
public MessageBuilder<T> 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<T> {
|
||||
return new GenericMessage<T>(this.payload, this.headers);
|
||||
}
|
||||
|
||||
private boolean isReadOnly(String key) {
|
||||
return (key.equals(MessageHeaders.ID) || key.equals(MessageHeaders.TIMESTAMP));
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in New Issue
Block a user