diff --git a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DefaultFileNameGenerator.java b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DefaultFileNameGenerator.java index aae2a9b7f6..f447089c31 100644 --- a/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DefaultFileNameGenerator.java +++ b/org.springframework.integration.adapter/src/main/java/org/springframework/integration/adapter/file/DefaultFileNameGenerator.java @@ -30,7 +30,7 @@ public class DefaultFileNameGenerator implements FileNameGenerator { public String generateFileName(Message message) { String filenameProperty = message.getHeaders().get(FILENAME_PROPERTY_KEY, String.class); return StringUtils.hasText(filenameProperty) ? - filenameProperty : message.getId() + ".msg"; + filenameProperty : message.getHeaders().getId() + ".msg"; } } diff --git a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/DefaultFileNameGeneratorTests.java b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/DefaultFileNameGeneratorTests.java index af9f299a26..21c5bf2a71 100644 --- a/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/DefaultFileNameGeneratorTests.java +++ b/org.springframework.integration.adapter/src/test/java/org/springframework/integration/adapter/file/DefaultFileNameGeneratorTests.java @@ -44,7 +44,7 @@ public class DefaultFileNameGeneratorTests { Message message = new GenericMessage("testing"); FileNameGenerator generator = new DefaultFileNameGenerator(); String filename = generator.generateFileName(message); - assertTrue(filename.startsWith("" + message.getId())); + assertTrue(filename.startsWith("" + message.getHeaders().getId())); assertTrue(filename.endsWith(".msg")); } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java index a66aa4643d..182ea83388 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/MessageStoringInterceptor.java @@ -56,7 +56,7 @@ public class MessageStoringInterceptor extends ChannelInterceptorAdapter { @Override public Message preSend(Message message, MessageChannel channel) { if (message != null) { - this.messageStore.put(message.getId(), message); + this.messageStore.put(message.getHeaders().getId(), message); } return message; } @@ -86,7 +86,7 @@ public class MessageStoringInterceptor extends ChannelInterceptorAdapter { @Override public Message postReceive(Message message, MessageChannel channel) { if (message != null) { - this.messageStore.remove(message.getId()); + this.messageStore.remove(message.getHeaders().getId()); } return message; } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java index 4cd2d3a4cb..cf6052d939 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/channel/interceptor/WireTap.java @@ -101,7 +101,7 @@ public class WireTap extends ChannelInterceptorAdapter implements Lifecycle { public Message preSend(Message message, MessageChannel channel) { if (this.running && this.selectorsAccept(message)) { Message duplicate = MessageBuilder.fromMessage(message) - .setHeader(ORIGINAL_MESSAGE_ID_KEY, message.getId()) + .setHeader(ORIGINAL_MESSAGE_ID_KEY, message.getHeaders().getId()) .build(); if (!this.secondaryChannel.send(duplicate, 0)) { if (logger.isWarnEnabled()) { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java index 0b31d74748..5f3f8a8c70 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/endpoint/HandlerEndpoint.java @@ -117,7 +117,7 @@ public class HandlerEndpoint extends AbstractEndpoint { Object correlationId = replyMessage.getHeaders().getCorrelationId(); if (correlationId == null) { replyMessage = MessageBuilder.fromMessage(replyMessage) - .setHeader(MessageHeaders.CORRELATION_ID, message.getId()).build(); + .setHeader(MessageHeaders.CORRELATION_ID, message.getHeaders().getId()).build(); } if (replyMessage != null) { MessageTarget returnAddress = resolveReturnAddress(replyMessage); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java index bab5e83aa7..6dcd6e7b90 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/gateway/SimpleMessagingGateway.java @@ -192,8 +192,8 @@ public class SimpleMessagingGateway extends MessagingGatewaySupport implements M 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()); + ? this.replyMessageCorrelator.getReply(message.getHeaders().getId(), this.replyTimeout) + : this.replyMessageCorrelator.getReply(message.getHeaders().getId()); } private void registerReplyMessageCorrelator() { diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java index b7cc9053ec..908d6d19ed 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/handler/AbstractMessageHandlerAdapter.java @@ -130,7 +130,7 @@ public abstract class AbstractMessageHandlerAdapter extends AbstractMethodInvoki return null; } return MessageBuilder.fromMessage(reply).copyHeadersIfAbsent(originalMessage.getHeaders()) - .setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, originalMessage.getId()).build(); + .setHeaderIfAbsent(MessageHeaders.CORRELATION_ID, originalMessage.getHeaders().getId()).build(); } /** diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessage.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessage.java index ae3c2a791e..58399c6ae5 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessage.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/AsyncMessage.java @@ -65,28 +65,6 @@ public class AsyncMessage implements Future>, Message { return this.future.isDone(); } - public MessageHeaders getHeaders() { - try { - return this.future.get().getHeaders(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return null; - } catch (ExecutionException e) { - throw new MessagingException("failure occurred in AsyncMessage", e); - } - } - - public Object getId() { - try { - return this.future.get().getId(); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return null; - } catch (ExecutionException e) { - throw new MessagingException("failure occurred in AsyncMessage", e); - } - } - public T getPayload() { try { return this.future.get().getPayload(); @@ -98,4 +76,15 @@ public class AsyncMessage implements Future>, Message { } } + public MessageHeaders getHeaders() { + try { + return this.future.get().getHeaders(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return null; + } catch (ExecutionException e) { + throw new MessagingException("failure occurred in AsyncMessage", e); + } + } + } 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 548924f13a..fea4b7fd38 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 @@ -21,7 +21,6 @@ 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; /** @@ -31,14 +30,10 @@ import org.springframework.util.Assert; */ public class GenericMessage implements Message, Serializable { - 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 payload. The id will be generated by @@ -67,25 +62,20 @@ public class GenericMessage implements Message, Serializable { else if (headers instanceof MessageHeaders) { headers = new HashMap(headers); } - headers.put(ID_HEADER_KEY, this.defaultIdGenerator.generateId()); this.headers = new MessageHeaders(headers); } - public Object getId() { - return this.headers.get(ID_HEADER_KEY); + public T getPayload() { + return this.payload; } public MessageHeaders getHeaders() { return this.headers; } - public T getPayload() { - return this.payload; - } - public String toString() { - return "[ID=" + this.getId() + "][Headers=" + this.headers + "][Payload='" + this.payload + "']"; + return "[Payload=" + this.payload + "][Headers=" + this.headers + "]"; } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/Message.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/Message.java index c93cfae38d..59b64447a0 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/Message.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/Message.java @@ -24,8 +24,6 @@ package org.springframework.integration.message; */ public interface Message { - Object getId(); - T getPayload(); MessageHeaders getHeaders(); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java index 673292e816..5281e1350e 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/message/MessageHeaders.java @@ -24,6 +24,9 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; +import org.springframework.integration.util.IdGenerator; +import org.springframework.integration.util.RandomUuidGenerator; + /** * The headers for a {@link Message}. * @@ -32,6 +35,8 @@ import java.util.Set; */ public final class MessageHeaders implements Map, Serializable { + public static final String ID = "internal.header.id"; + public static final String TIMESTAMP = "internal.header.timestamp"; public static final String CORRELATION_ID = "internal.header.correlationId"; @@ -51,13 +56,20 @@ public final class MessageHeaders implements Map, Serializable { private final Map headers; + private transient final IdGenerator idGenerator = new RandomUuidGenerator(); + public MessageHeaders(Map headers) { this.headers = (headers != null ? headers : new HashMap()); + this.headers.put(ID, this.idGenerator.generateId()); this.headers.put(TIMESTAMP, new Long(System.currentTimeMillis())); } + public Object getId() { + return this.get(ID); + } + public Long getTimestamp() { return this.get(TIMESTAMP, Long.class); } @@ -169,7 +181,7 @@ public final class MessageHeaders implements Map, Serializable { } public String toString() { - return headers.toString(); + return this.headers.toString(); } } diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java b/org.springframework.integration/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java index 855c37b0af..cb30f6d79a 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/router/SplitterMessageHandlerAdapter.java @@ -85,7 +85,7 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter Message splitMessage = (item instanceof Message) ? (Message) item : this.createReplyMessage(item, originalMessage); splitMessage = MessageBuilder.fromMessage(splitMessage) - .setCorrelationId(originalMessage.getId()) + .setCorrelationId(originalMessage.getHeaders().getId()) .setSequenceNumber(++sequenceNumber) .setSequenceSize(sequenceSize).build(); this.sendMessage(splitMessage, this.outputChannelName); @@ -99,7 +99,7 @@ public class SplitterMessageHandlerAdapter extends AbstractMessageHandlerAdapter Message splitMessage = (item instanceof Message) ? (Message) item : this.createReplyMessage(item, originalMessage); splitMessage = MessageBuilder.fromMessage(splitMessage) - .setCorrelationId(originalMessage.getId()) + .setCorrelationId(originalMessage.getHeaders().getId()) .setSequenceNumber(++sequenceNumber) .setSequenceSize(sequenceSize).build(); this.sendMessage(splitMessage, this.outputChannelName); diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/util/RandomUuidGenerator.java b/org.springframework.integration/src/main/java/org/springframework/integration/util/RandomUuidGenerator.java index 9a15d66c19..4ec757e254 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/util/RandomUuidGenerator.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/util/RandomUuidGenerator.java @@ -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. diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java index 5eccd0101c..a73a4edbeb 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/channel/interceptor/WireTapTests.java @@ -88,9 +88,9 @@ public class WireTapTests { mainChannel.send(new StringMessage("testing")); Message original = mainChannel.receive(0); Message duplicate = secondaryChannel.receive(0); - Object duplicateId = duplicate.getId(); + Object duplicateId = duplicate.getHeaders().getId(); assertNotNull(duplicateId); - assertFalse("message ids should not match", original.getId().equals(duplicateId)); + assertFalse("message ids should not match", original.getHeaders().getId().equals(duplicateId)); } @Test @@ -103,7 +103,7 @@ public class WireTapTests { Message duplicate = secondaryChannel.receive(0); Object originalIdAttribute = duplicate.getHeaders().get(WireTap.ORIGINAL_MESSAGE_ID_KEY); assertNotNull(originalIdAttribute); - assertEquals(original.getId(), originalIdAttribute); + assertEquals(original.getHeaders().getId(), originalIdAttribute); } @Test diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java index 721238ea72..10e96a11d9 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/endpoint/HandlerEndpointTests.java @@ -273,7 +273,7 @@ public class HandlerEndpointTests { .setReturnAddress(replyChannel).build(); endpoint.send(message); Message reply = replyChannel.receive(500); - assertEquals(message.getId(), reply.getHeaders().getCorrelationId()); + assertEquals(message.getHeaders().getId(), reply.getHeaders().getCorrelationId()); } @Test @@ -290,7 +290,7 @@ public class HandlerEndpointTests { endpoint.send(message); Message reply = replyChannel.receive(500); Object correlationId = reply.getHeaders().getCorrelationId(); - assertFalse(message.getId().equals(correlationId)); + assertFalse(message.getHeaders().getId().equals(correlationId)); assertEquals("ABC-123", correlationId); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java index bbfda67eab..c5c08e4ae2 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/SimpleMessagingGatewayTests.java @@ -161,7 +161,7 @@ public class SimpleMessagingGatewayTests { expect(replyChannel.getName()).andReturn("replyChannel").anyTimes(); expect(messageMock.getHeaders()).andReturn(messageHeadersMock); expect(requestChannel.send(messageMock)).andReturn(true); - expect(messageMock.getId()).andReturn(1); + expect(messageHeadersMock.getId()).andReturn(1); //play scenario replay(allmocks); @@ -198,7 +198,7 @@ public class SimpleMessagingGatewayTests { expect(messageMock.getHeaders()).andReturn(messageHeadersMock); expect(messageHeadersMock.getReturnAddress()).andReturn(replyChannel); expect(requestChannel.send(messageMock)).andReturn(true); - expect(messageMock.getId()).andReturn(1); + expect(messageHeadersMock.getId()).andReturn(1); replay(allmocks); this.simpleMessagingGateway.sendAndReceiveMessage(messageMock); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/config/GatewayParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/config/GatewayParserTests.java index c802744214..96d0df0a06 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/gateway/config/GatewayParserTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/gateway/config/GatewayParserTests.java @@ -95,7 +95,7 @@ public class GatewayParserTests { public void run() { Message request = requestChannel.receive(); Message reply = MessageBuilder.fromPayload(request.getPayload() + "bar") - .setCorrelationId(request.getId()).build(); + .setCorrelationId(request.getHeaders().getId()).build(); replyChannel.send(reply); } }); diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java index 43643ca705..3c3bf1b885 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/CorrelationIdTests.java @@ -56,7 +56,7 @@ public class CorrelationIdTests { adapter.setMethodName("upperCase"); adapter.afterPropertiesSet(); Message reply = adapter.handle(message); - assertEquals(message.getId(), reply.getHeaders().getCorrelationId()); + assertEquals(message.getHeaders().getId(), reply.getHeaders().getCorrelationId()); } @Test @@ -127,8 +127,8 @@ public class CorrelationIdTests { splitter.handle(message); Message reply1 = testChannel.receive(100); Message reply2 = testChannel.receive(100); - assertEquals(message.getId(), reply1.getHeaders().getCorrelationId()); - assertEquals(message.getId(), reply2.getHeaders().getCorrelationId()); + assertEquals(message.getHeaders().getId(), reply1.getHeaders().getCorrelationId()); + assertEquals(message.getHeaders().getId(), reply2.getHeaders().getCorrelationId()); } diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java index 465bbbdaf5..1d7384ac94 100644 --- a/org.springframework.integration/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java +++ b/org.springframework.integration/src/test/java/org/springframework/integration/router/SplitterMessageHandlerAdapterTests.java @@ -225,12 +225,12 @@ public class SplitterMessageHandlerAdapterTests { assertNotNull(reply1); assertEquals(new Integer(2), reply1.getHeaders().getSequenceSize()); assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); - assertEquals(message.getId(), reply1.getHeaders().getCorrelationId()); + assertEquals(message.getHeaders().getId(), reply1.getHeaders().getCorrelationId()); Message reply2 = testChannel.receive(0); assertNotNull(reply2); assertEquals(new Integer(2), reply2.getHeaders().getSequenceSize()); assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); - assertEquals(message.getId(), reply2.getHeaders().getCorrelationId()); + assertEquals(message.getHeaders().getId(), reply2.getHeaders().getCorrelationId()); } @Test @@ -242,12 +242,12 @@ public class SplitterMessageHandlerAdapterTests { assertNotNull(reply1); assertEquals(new Integer(2), reply1.getHeaders().getSequenceSize()); assertEquals(new Integer(1), reply1.getHeaders().getSequenceNumber()); - assertEquals(message.getId(), reply1.getHeaders().getCorrelationId()); + assertEquals(message.getHeaders().getId(), reply1.getHeaders().getCorrelationId()); Message reply2 = testChannel.receive(0); assertNotNull(reply2); assertEquals(new Integer(2), reply2.getHeaders().getSequenceSize()); assertEquals(new Integer(2), reply2.getHeaders().getSequenceNumber()); - assertEquals(message.getId(), reply2.getHeaders().getCorrelationId()); + assertEquals(message.getHeaders().getId(), reply2.getHeaders().getCorrelationId()); }