From 42113a497c5fb1c0f995db516b8f4f99f3460b82 Mon Sep 17 00:00:00 2001
From: David Syer <dsyer@vmware.com>
Date: Tue, 27 Apr 2010 11:58:37 +0000
Subject: [PATCH] RESOLVED - issue INT-1058, INT-1098: message ID is unique to
 an immutable instance

---
 ...ngWithMultipleSourcesIntegrationTests.java |  14 +--
 org.springframework.integration.ip/pom.xml    |  12 +++
 .../integration/ip/IpHeaders.java             |   5 +-
 .../ip/udp/DatagramPacketMessageMapper.java   |   3 +-
 .../udp/UnicastReceivingChannelAdapter.java   |   2 +-
 .../udp/DatagramPacketMessageMapperTests.java |  20 +++-
 .../DatagramPacketSendingHandlerTests.java    |   5 +-
 .../integration/ip/udp/testIp-out-context.xml |   2 +-
 .../integration/jdbc/JdbcMessageStore.java    | 101 ++++++------------
 .../integration/jdbc/schema-derby.sql         |   3 +-
 .../jdbc/JdbcMessageStoreTests.java           |  31 +++++-
 org.springframework.integration.jmx/pom.xml   |  71 ++++++------
 .../.settings/org.eclipse.jdt.core.prefs      |   5 +
 .../.settings/org.maven.ide.eclipse.prefs     |   9 ++
 .../test/matcher/PayloadAndHeaderMatcher.java |  68 ++++++++++++
 .../integration/core/MessageHeaders.java      |  26 ++---
 .../integration/message/MessageBuilder.java   |   4 +-
 .../integration/store/MessageStore.java       |  27 ++---
 .../CorrelatingMessageHandlerTests.java       |  19 ++--
 .../integration/config/ChainParserTests.java  |  23 ++--
 .../config/xml/BridgeParserTests.java         |  17 ++-
 .../filter/MessageFilterTests.java            |   7 +-
 .../handler/BridgeHandlerTests.java           |  14 ++-
 .../json/InboundJsonMessageMapperTests.java   |  24 +++--
 .../message/MessageBuilderTests.java          |  29 +++++
 .../message/MessageHeadersTests.java          |  22 ++++
 .../integration/message/MessageMatcher.java   |  61 +++++++++++
 27 files changed, 435 insertions(+), 189 deletions(-)
 create mode 100644 org.springframework.integration.stream/.settings/org.eclipse.jdt.core.prefs
 create mode 100644 org.springframework.integration.stream/.settings/org.maven.ide.eclipse.prefs
 create mode 100644 org.springframework.integration.test/src/main/java/org/springframework/integration/test/matcher/PayloadAndHeaderMatcher.java
 create mode 100644 org.springframework.integration/src/test/java/org/springframework/integration/message/MessageMatcher.java

diff --git a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests.java b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests.java
index 04230bd16e..45ac321826 100644
--- a/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests.java
+++ b/org.springframework.integration.file/src/test/java/org/springframework/integration/file/locking/FileLockingWithMultipleSourcesIntegrationTests.java
@@ -15,6 +15,13 @@
  */
 package org.springframework.integration.file.locking;
 
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.junit.Assert.assertThat;
+import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload;
+
+import java.io.File;
+import java.io.IOException;
+
 import org.junit.Before;
 import org.junit.BeforeClass;
 import org.junit.Test;
@@ -24,13 +31,6 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.integration.file.FileReadingMessageSource;
 import org.springframework.test.context.ContextConfiguration;
 
-import java.io.File;
-import java.io.IOException;
-
-import static org.hamcrest.CoreMatchers.nullValue;
-import static org.junit.Assert.assertThat;
-import static org.springframework.integration.test.matcher.PayloadMatcher.hasPayload;
-
 /**
  * @author Iwein Fuld
  */
diff --git a/org.springframework.integration.ip/pom.xml b/org.springframework.integration.ip/pom.xml
index b913ac4888..ac9ab90976 100644
--- a/org.springframework.integration.ip/pom.xml
+++ b/org.springframework.integration.ip/pom.xml
@@ -54,6 +54,18 @@
       <version>${project.version}</version>
       <scope>compile</scope>
     </dependency>
+    <dependency>
+      <groupId>org.springframework.integration</groupId>
+      <artifactId>spring-integration-stream</artifactId>
+      <version>${project.version}</version>
+      <scope>runtime</scope>
+    </dependency>
+    <dependency>
+      <groupId>org.springframework.integration</groupId>
+      <artifactId>spring-integration-test</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
     <dependency>
       <groupId>org.springframework.integration</groupId>
       <artifactId>spring-integration-adapter</artifactId>
diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/IpHeaders.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/IpHeaders.java
index f6890177ed..e2eae2c845 100644
--- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/IpHeaders.java
+++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/IpHeaders.java
@@ -19,10 +19,11 @@ package org.springframework.integration.ip;
 import org.springframework.integration.core.MessageHeaders;
 
 /**
- * Headers for Messages mapped from UDP datagram packets.
+ * Headers for Messages mapped from IP datagram packets.
  * 
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Dave Syer
  * @since 2.0
  */
 public abstract class IpHeaders {
@@ -41,6 +42,8 @@ public abstract class IpHeaders {
 
 	public static final String ACK_ADDRESS = IP + "ackTo";
 
+	public static final String ACK_ID = IP + "ackId";
+
 	public static final String REMOTE_PORT = TCP + "remote_port";
 
 }
diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java
index b1cfd3bfbe..266d768295 100644
--- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java
+++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapper.java
@@ -53,6 +53,7 @@ import org.springframework.util.Assert;
  * 
  * @author Mark Fisher
  * @author Gary Russell
+ * @author Dave Syer
  * @since 2.0
  */
 public class DatagramPacketMessageMapper implements InboundMessageMapper<DatagramPacket>,
@@ -187,7 +188,7 @@ public class DatagramPacketMessageMapper implements InboundMessageMapper<Datagra
 					payload = new byte[length];
 					System.arraycopy(packet.getData(), offset + matcher.end(), payload, 0, length);
 					message = MessageBuilder.withPayload(payload)
-							.setHeader(MessageHeaders.ID, UUID.fromString(matcher.group(2)))
+							.setHeader(IpHeaders.ACK_ID, UUID.fromString(matcher.group(2)))
 							.setHeader(IpHeaders.ACK_ADDRESS, matcher.group(1))
 							.setHeader(IpHeaders.HOSTNAME, packet.getAddress().getHostName())
 							.setHeader(IpHeaders.IP_ADDRESS, packet.getAddress().getHostAddress())
diff --git a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
index 1347b3fa42..0701a529f8 100644
--- a/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
+++ b/org.springframework.integration.ip/src/main/java/org/springframework/integration/ip/udp/UnicastReceivingChannelAdapter.java
@@ -125,7 +125,7 @@ public class UnicastReceivingChannelAdapter extends AbstractInternetProtocolRece
 
 	protected void sendAck(Message<byte[]> message) {
 		MessageHeaders headers = message.getHeaders();
-		Object id = headers.getId();
+		Object id = headers.get(IpHeaders.ACK_ID);
 		byte[] ack = id.toString().getBytes();
 		String ackAddress = ((String) headers.get(IpHeaders.ACK_ADDRESS)).trim();
 		Matcher mat = addressPattern.matcher(ackAddress);
diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java
index 401b9a0b34..9d23901df1 100644
--- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java
+++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketMessageMapperTests.java
@@ -28,20 +28,34 @@ import org.junit.Test;
 
 import org.springframework.integration.adapter.MessageMappingException;
 import org.springframework.integration.core.Message;
+import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.ip.udp.DatagramPacketMessageMapper;
 import org.springframework.integration.message.MessageBuilder;
 
 /**
  * @author Gary Russell
+ * @author Dave Syer
  * @since 2.0
  */
 public class DatagramPacketMessageMapperTests {
 
 	@Test
-	public void testFromToMessage() throws Exception {
+	public void testFromToMessageNoAckNoLengthCheck() throws Exception {
 		test(false, false);
+	}
+
+	@Test
+	public void testFromToMessageAckNoLengthCheck() throws Exception {
 		test(true, false);
+	}
+
+	@Test
+	public void testFromToMessageNoAckLengthCheck() throws Exception {
 		test(false, true);
+	}
+
+	@Test
+	public void testFromToMessageAckLengthCheck() throws Exception {
 		test(true, true);
 	}
 
@@ -56,8 +70,8 @@ public class DatagramPacketMessageMapperTests {
 		Message<byte[]> messageOut = mapper.toMessage(packet);
 		assertEquals(new String(message.getPayload()), new String(messageOut.getPayload()));
 		if (ack) {
-			assertEquals(message.getHeaders().getId().toString(), 
-					     messageOut.getHeaders().getId().toString());
+			assertEquals(messageOut.getHeaders().get(IpHeaders.ACK_ID).toString(), 
+					     message.getHeaders().getId().toString());
 		}
 	}
 
diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java
index 02de1b4f96..e3bf27e951 100644
--- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java
+++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/DatagramPacketSendingHandlerTests.java
@@ -34,6 +34,7 @@ import org.junit.Ignore;
 import org.junit.Test;
 
 import org.springframework.integration.core.Message;
+import org.springframework.integration.ip.IpHeaders;
 import org.springframework.integration.ip.udp.DatagramPacketMessageMapper;
 import org.springframework.integration.ip.udp.MulticastSendingMessageHandler;
 import org.springframework.integration.ip.udp.UnicastSendingMessageHandler;
@@ -100,7 +101,7 @@ public class DatagramPacketSendingHandlerTests {
 					mapper.setAcknowledge(true);
 					mapper.setLengthCheck(true);
 					Message<byte[]> message = mapper.toMessage(receivedPacket);
-					Object id = message.getHeaders().getId();
+					Object id = message.getHeaders().get(IpHeaders.ACK_ID);
 					byte[] ack = id.toString().getBytes();
 					DatagramPacket ackPack = new DatagramPacket(ack, ack.length, 
 							                        new InetSocketAddress("localHost", ackPort));
@@ -201,7 +202,7 @@ public class DatagramPacketSendingHandlerTests {
 					mapper.setAcknowledge(true);
 					mapper.setLengthCheck(true);
 					Message<byte[]> message = mapper.toMessage(receivedPacket);
-					Object id = message.getHeaders().getId();
+					Object id = message.getHeaders().get(IpHeaders.ACK_ID);
 					byte[] ack = id.toString().getBytes();
 					DatagramPacket ackPack = new DatagramPacket(ack, ack.length, 
 							                        new InetSocketAddress("localHost", ackPort));
diff --git a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-context.xml b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-context.xml
index f0786fde4d..dd203b878d 100644
--- a/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-context.xml
+++ b/org.springframework.integration.ip/src/test/java/org/springframework/integration/ip/udp/testIp-out-context.xml
@@ -35,7 +35,7 @@
 		acknowledge="true"
 		ack-host="localhost"
 		ack-port="22222"
-		ack-timeout="10000"
+		ack-timeout="2000"
 		channel="outputChannel"/>
 
 	<beans:import resource="testIp-common-context.xml" />
diff --git a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
index 8c13f49331..c1e4cdd8c3 100644
--- a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
+++ b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
@@ -14,7 +14,6 @@ import javax.sql.DataSource;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
-import org.springframework.dao.OptimisticLockingFailureException;
 import org.springframework.integration.core.Message;
 import org.springframework.integration.jdbc.util.SerializationUtils;
 import org.springframework.integration.message.MessageBuilder;
@@ -46,32 +45,22 @@ public class JdbcMessageStore implements MessageStore {
 	 */
 	public static final String DEFAULT_TABLE_PREFIX = "INT_";
 
-	private static final String LIST_MESSAGES_BY_CORRELATION_KEY = "SELECT MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES, VERSION from %PREFIX%MESSAGE where CORRELATION_KEY=?";
+	private static final String LIST_MESSAGES_BY_CORRELATION_KEY = "SELECT MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES from %PREFIX%MESSAGE where CORRELATION_KEY=?";
 
-	private static final String GET_MESSAGE = "SELECT MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES, VERSION from %PREFIX%MESSAGE where MESSAGE_ID=?";
+	private static final String GET_MESSAGE = "SELECT MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES from %PREFIX%MESSAGE where MESSAGE_ID=?";
 
 	private static final String DELETE_MESSAGE = "DELETE from %PREFIX%MESSAGE where MESSAGE_ID=?";
 
-	private static final String CREATE_MESSAGE = "INSERT into %PREFIX%MESSAGE(MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES, VERSION)"
-			+ " values (?, ?, ?, ?)";
-
-	private static final String UPDATE_MESSAGE = "UPDATE %PREFIX%MESSAGE set CORRELATION_KEY=?, MESSAGE_BYTES=?, VERSION=? where VERSION=? and MESSAGE_ID=?";
-
-	private static final String CURRENT_VERSION_MESSAGE = "SELECT VERSION from %PREFIX%MESSAGE where MESSAGE_ID=?";
+	private static final String CREATE_MESSAGE = "INSERT into %PREFIX%MESSAGE(MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES)"
+			+ " values (?, ?, ?)";
 
 	public static final int DEFAULT_LONG_STRING_LENGTH = 2500;
 
 	/**
-	 * The name of the message header that stores the surrogate key used by this
-	 * message store
+	 * The name of the message header that stores a flag to indicate that the
+	 * message has been saved. This is an optimization for the put method.
 	 */
-	public static final String ID_KEY = JdbcMessageStore.class.getSimpleName() + ".ID";
-
-	/**
-	 * The name of the message header that stores the version used by this
-	 * message store to implement optimistic locking
-	 */
-	public static final String VERSION_KEY = JdbcMessageStore.class.getSimpleName() + ".VERSION";
+	public static final String SAVED_KEY = JdbcMessageStore.class.getSimpleName() + ".SAVED";
 
 	private String tablePrefix = DEFAULT_TABLE_PREFIX;
 
@@ -79,6 +68,8 @@ public class JdbcMessageStore implements MessageStore {
 
 	private LobHandler lobHandler = new DefaultLobHandler();
 
+	private MessageMapper mapper = new MessageMapper();
+
 	/**
 	 * Convenient constructor for configuration use.
 	 */
@@ -177,8 +168,7 @@ public class JdbcMessageStore implements MessageStore {
 	}
 
 	public Message<?> get(UUID id) {
-		List<Message<?>> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id) },
-				new MessageMapper());
+		List<Message<?>> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id) }, mapper);
 		if (list.isEmpty()) {
 			return null;
 		}
@@ -187,59 +177,36 @@ public class JdbcMessageStore implements MessageStore {
 
 	public List<Message<?>> list(Object correlationId) {
 		return jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_CORRELATION_KEY), new Object[] { getKey(correlationId) },
-				new MessageMapper());
+				mapper);
 	}
 
 	public <T> Message<T> put(final Message<T> message) {
 
-		boolean alreadySaved = message.getHeaders().containsKey(VERSION_KEY);
-		final int version = alreadySaved ? (Integer) message.getHeaders().get(
-				VERSION_KEY) : 0;
-
-		if (alreadySaved) {
-
-			final String correlationId = getKey(message.getHeaders().getCorrelationId());
-			final String messageId = getKey(message.getHeaders().getId());
-			final byte[] messageBytes = SerializationUtils.serialize(message);
-
-			int updated = jdbcTemplate.update(getQuery(UPDATE_MESSAGE), new PreparedStatementSetter() {
-				public void setValues(PreparedStatement ps) throws SQLException {
-					logger.debug("Updating message with id key=" + messageId);
-					ps.setString(1, correlationId);
-					lobHandler.getLobCreator().setBlobAsBytes(ps, 2, messageBytes);
-					ps.setInt(3, version + 1);
-					ps.setInt(4, version);
-					ps.setString(5, messageId);
-				}
-			});
-
-			if (updated != 1) {
-				int currentVersion = jdbcTemplate.queryForInt(getQuery(CURRENT_VERSION_MESSAGE), new Object[] { messageId });
-				throw new OptimisticLockingFailureException("Attempt to update message id="
-						+ message.getHeaders().getId() + " with wrong version (" + version
-						+ "), where current version is " + currentVersion);
+		if (message.getHeaders().containsKey(SAVED_KEY)) {
+			@SuppressWarnings("unchecked")
+			Message<T> saved = (Message<T>) get(message.getHeaders().getId());
+			if (saved != null) {
+				if (saved.equals(message)) {
+					return message;
+				} // We need to save it under its own id
 			}
-
-		}
-		else {
-
-			final String messageId = getKey(message.getHeaders().getId());
-			final String correlationId = getKey(message.getHeaders().getCorrelationId());
-			final byte[] messageBytes = SerializationUtils.serialize(message);
-
-			jdbcTemplate.update(getQuery(CREATE_MESSAGE), new PreparedStatementSetter() {
-				public void setValues(PreparedStatement ps) throws SQLException {
-					logger.debug("Inserting message with id key=" + messageId);
-					ps.setString(1, messageId);
-					ps.setString(2, correlationId);
-					lobHandler.getLobCreator().setBlobAsBytes(ps, 3, messageBytes);
-					ps.setInt(4, version);
-				}
-			});
-
 		}
 
-		return MessageBuilder.fromMessage(message).setHeader(VERSION_KEY, version).build();
+		Message<T> result = MessageBuilder.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE).build();
+		final String messageId = getKey(result.getHeaders().getId());
+		final String correlationId = getKey(result.getHeaders().getCorrelationId());
+		final byte[] messageBytes = SerializationUtils.serialize(result);
+
+		jdbcTemplate.update(getQuery(CREATE_MESSAGE), new PreparedStatementSetter() {
+			public void setValues(PreparedStatement ps) throws SQLException {
+				logger.debug("Inserting message with id key=" + messageId);
+				ps.setString(1, messageId);
+				ps.setString(2, correlationId);
+				lobHandler.getLobCreator().setBlobAsBytes(ps, 3, messageBytes);
+			}
+		});
+
+		return result;
 
 	}
 
@@ -283,7 +250,7 @@ public class JdbcMessageStore implements MessageStore {
 		public Message<?> mapRow(ResultSet rs, int rowNum) throws SQLException {
 			Message<?> message = (Message<?>) SerializationUtils.deserialize(lobHandler.getBlobAsBytes(rs,
 					"MESSAGE_BYTES"));
-			return MessageBuilder.fromMessage(message).setHeader(VERSION_KEY, rs.getInt("VERSION")).build();
+			return message;
 		}
 
 	}
diff --git a/org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql b/org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql
index 4d0e82fee6..9900200c49 100644
--- a/org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql
+++ b/org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql
@@ -1,6 +1,5 @@
 CREATE TABLE INT_MESSAGE  (
 	MESSAGE_ID VARCHAR(100) NOT NULL PRIMARY KEY,
 	CORRELATION_KEY VARCHAR(100),
-	MESSAGE_BYTES BLOB,
-	VERSION BIGINT  
+	MESSAGE_BYTES BLOB
 );
diff --git a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
index 3947a55cd6..f6731081b1 100644
--- a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
+++ b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
@@ -2,7 +2,10 @@ package org.springframework.integration.jdbc;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
+import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptImmutableHeaders;
 
 import java.util.UUID;
 
@@ -43,10 +46,11 @@ public class JdbcMessageStoreTests {
 	@Transactional
 	public void testAddAndGet() throws Exception {
 		Message<String> message = MessageBuilder.withPayload("foo").setCorrelationId("X").build();
-		message = messageStore.put(message);
-		Message<?> result = messageStore.get(message.getHeaders().getId());
+		Message<String> saved = messageStore.put(message);
+		assertNull(messageStore.get(message.getHeaders().getId()));
+		Message<?> result = messageStore.get(saved.getHeaders().getId());
 		assertNotNull(result);
-		assertEquals(message.getPayload(), result.getPayload());
+		assertThat(saved, sameExceptImmutableHeaders(result));
 	}
 
 	@Test
@@ -59,6 +63,27 @@ public class JdbcMessageStoreTests {
 		assertEquals("Y", messageStore.get(message.getHeaders().getId()).getHeaders().getCorrelationId());
 	}
 
+	@Test
+	@Transactional
+	public void testAddAndUpdateAlreadySaved() throws Exception {
+		Message<String> message = MessageBuilder.withPayload("foo").build();
+		message = messageStore.put(message);
+		Message<String> result = messageStore.put(message);
+		assertEquals(message, result);
+	}
+
+	@Test
+	@Transactional
+	public void testAddAndUpdateAlreadySavedAndCopied() throws Exception {
+		Message<String> message = MessageBuilder.withPayload("foo").build();
+		Message<String> saved = messageStore.put(message);
+		Message<String> copy = MessageBuilder.fromMessage(saved).build();
+		Message<String> result = messageStore.put(copy);
+		assertNotSame(copy, result);
+		assertThat(saved, sameExceptImmutableHeaders(result));
+		assertNotNull(messageStore.get(saved.getHeaders().getId()));
+	}
+
 	@Test
 	@Transactional
 	public void testAddAndListByCorrelationId() throws Exception {
diff --git a/org.springframework.integration.jmx/pom.xml b/org.springframework.integration.jmx/pom.xml
index 2bdad2f1d3..d8ee8a2b09 100644
--- a/org.springframework.integration.jmx/pom.xml
+++ b/org.springframework.integration.jmx/pom.xml
@@ -1,34 +1,41 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
-  <modelVersion>4.0.0</modelVersion>
-  <groupId>org.springframework.integration</groupId>
-  <artifactId>spring-integration-jmx</artifactId>
-  <packaging>jar</packaging>
-  <name>Spring Integration JMX Support</name>
-  <parent>
-    <groupId>org.springframework.integration</groupId>
-    <artifactId>spring-integration-parent</artifactId>
-    <version>2.0.0.BUILD-SNAPSHOT</version>
-    <relativePath>../spring-integration-parent</relativePath>
-  </parent>
-  <dependencies>
-    <dependency>
-      <groupId>junit</groupId>
-      <artifactId>junit</artifactId>
-      <version>${junit.version}</version>
-      <scope>test</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework</groupId>
-      <artifactId>spring-context</artifactId>
-      <version>${spring.version}</version>
-      <scope>compile</scope>
-    </dependency>
-    <dependency>
-      <groupId>org.springframework.integration</groupId>
-      <artifactId>spring-integration-core</artifactId>
-      <version>${project.version}</version>
-      <scope>compile</scope>
-    </dependency>
-  </dependencies>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+	<modelVersion>4.0.0</modelVersion>
+	<groupId>org.springframework.integration</groupId>
+	<artifactId>spring-integration-jmx</artifactId>
+	<packaging>jar</packaging>
+	<name>Spring Integration JMX Support</name>
+	<parent>
+		<groupId>org.springframework.integration</groupId>
+		<artifactId>spring-integration-parent</artifactId>
+		<version>2.0.0.BUILD-SNAPSHOT</version>
+		<relativePath>../spring-integration-parent</relativePath>
+	</parent>
+	<dependencies>
+		<dependency>
+			<groupId>junit</groupId>
+			<artifactId>junit</artifactId>
+			<version>${junit.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-context</artifactId>
+			<version>${spring.version}</version>
+			<scope>compile</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework</groupId>
+			<artifactId>spring-test</artifactId>
+			<version>${spring.version}</version>
+			<scope>test</scope>
+		</dependency>
+		<dependency>
+			<groupId>org.springframework.integration</groupId>
+			<artifactId>spring-integration-core</artifactId>
+			<version>${project.version}</version>
+			<scope>compile</scope>
+		</dependency>
+	</dependencies>
 </project>
diff --git a/org.springframework.integration.stream/.settings/org.eclipse.jdt.core.prefs b/org.springframework.integration.stream/.settings/org.eclipse.jdt.core.prefs
new file mode 100644
index 0000000000..d259debf5e
--- /dev/null
+++ b/org.springframework.integration.stream/.settings/org.eclipse.jdt.core.prefs
@@ -0,0 +1,5 @@
+#Mon Mar 01 13:40:53 GMT 2010
+eclipse.preferences.version=1
+org.eclipse.jdt.core.compiler.codegen.targetPlatform=1.5
+org.eclipse.jdt.core.compiler.compliance=1.5
+org.eclipse.jdt.core.compiler.source=1.5
diff --git a/org.springframework.integration.stream/.settings/org.maven.ide.eclipse.prefs b/org.springframework.integration.stream/.settings/org.maven.ide.eclipse.prefs
new file mode 100644
index 0000000000..85f2635658
--- /dev/null
+++ b/org.springframework.integration.stream/.settings/org.maven.ide.eclipse.prefs
@@ -0,0 +1,9 @@
+#Mon Mar 01 13:38:53 GMT 2010
+activeProfiles=
+eclipse.preferences.version=1
+fullBuildGoals=process-test-resources
+includeModules=false
+resolveWorkspaceProjects=true
+resourceFilterGoals=process-resources resources\:testResources
+skipCompilerPlugin=true
+version=1
diff --git a/org.springframework.integration.test/src/main/java/org/springframework/integration/test/matcher/PayloadAndHeaderMatcher.java b/org.springframework.integration.test/src/main/java/org/springframework/integration/test/matcher/PayloadAndHeaderMatcher.java
new file mode 100644
index 0000000000..6fbc9cb6c7
--- /dev/null
+++ b/org.springframework.integration.test/src/main/java/org/springframework/integration/test/matcher/PayloadAndHeaderMatcher.java
@@ -0,0 +1,68 @@
+/**
+ * 
+ */
+package org.springframework.integration.test.matcher;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.hamcrest.Factory;
+import org.hamcrest.Matcher;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessageHeaders;
+
+/**
+ * Matcher to make assertions about message equality easier.  Usage:
+ * 
+ * <pre>
+ * &#064;Test
+ * public void testSomething() {
+ *   Message<String> expected = ...;
+ *   Message<String> result = ...;
+ *   assertThat(result, sameExceptImmutableHeaders(expected));
+ * }
+ * 
+ * &#064;Factory
+ * public static Matcher<Message<?>> sameExceptImmutableHeaders(Message<?> expected) {
+ *   return new PayloadAndHeaderMatcher(expected);
+ * }
+ * </pre>
+ *
+ * @author Dave Syer
+ *
+ */
+public class PayloadAndHeaderMatcher extends BaseMatcher<Message<?>> {
+
+	private final Object payload;
+	private final Map<String, Object> headers;
+
+	@Factory
+	public static Matcher<Message<?>> sameExceptImmutableHeaders(Message<?> expected) {
+		return new PayloadAndHeaderMatcher(expected);
+	}
+
+	private PayloadAndHeaderMatcher(Message<?> expected) {
+		this.payload = expected.getPayload();
+		this.headers = getHeaders(expected);
+	}
+
+	private Map<String, Object> getHeaders(Message<?> operand) {
+		HashMap<String, Object> headers = new HashMap<String, Object>(operand.getHeaders());
+		headers.remove(MessageHeaders.ID);
+		headers.remove(MessageHeaders.TIMESTAMP);
+		return headers;
+	}
+
+	public boolean matches(Object arg) {
+		Message<?> input = (Message<?>) arg;
+		Map<String, Object> inputHeaders = getHeaders(input);
+		return input.getPayload().equals(payload) && inputHeaders.equals(headers);
+	}
+
+	public void describeTo(Description description) {
+		description.appendText("a Message with Headers that match except ID and timestamp for payload: ").appendValue(payload).appendText(" and headers: ").appendValue(headers);
+	}
+	
+}
\ No newline at end of file
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 5b4fe998cf..7d6ef01e27 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
@@ -47,9 +47,10 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
 	public static final String PREFIX = "$";
 
 	/**
-	 * The key for the Message ID. This is an automatically generated UUID and should
-	 * never be explicitly set in the header map <b>except</b> in the case of Message
-	 * deserialization where the serialized Message's generated UUID is being restored.
+	 * The key for the Message ID. This is an automatically generated UUID and
+	 * should never be explicitly set in the header map <b>except</b> in the
+	 * case of Message deserialization where the serialized Message's generated
+	 * UUID is being restored.
 	 */
 	public static final String ID = PREFIX + "id";
 
@@ -71,26 +72,17 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
 
 	public static final String SEQUENCE_SIZE = PREFIX + "sequenceSize";
 
-
 	private final Map<String, Object> headers;
 
-
 	public MessageHeaders(Map<String, Object> headers) {
-		this.headers = (headers != null)
-				? new HashMap<String, Object>(headers)
-				: new HashMap<String, Object>();
-		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()));
-		}
+		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(HISTORY) == null) {
 			this.headers.put(HISTORY, new MessageHistory());
 		}
 	}
 
-
 	public UUID getId() {
 		return this.get(ID, UUID.class);
 	}
@@ -140,8 +132,8 @@ public final class MessageHeaders implements Map<String, Object>, Serializable {
 			return null;
 		}
 		if (!type.isAssignableFrom(value.getClass())) {
-			throw new IllegalArgumentException("Incorrect type specified for header '" + key
-					+ "'. Expected [" + type + "] but actual type is [" + value.getClass() + "]");
+			throw new IllegalArgumentException("Incorrect type specified for header '" + key + "'. Expected [" + type
+					+ "] but actual type is [" + value.getClass() + "]");
 		}
 		return (T) value;
 	}
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 ab3b3ae6a7..5eeb30b3a6 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
@@ -87,7 +87,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)) {
+		if (StringUtils.hasLength(headerName) && !headerName.equals(MessageHeaders.ID) && !headerName.equals(MessageHeaders.TIMESTAMP)) {
 			this.verifyType(headerName, headerValue);
 			this.modified = true;
 			if (headerValue == null) {
@@ -115,7 +115,7 @@ public final class MessageBuilder<T> {
 	 * Remove the value for the given header name.
 	 */
 	public MessageBuilder<T> removeHeader(String headerName) {
-		if (StringUtils.hasLength(headerName)) {
+		if (StringUtils.hasLength(headerName) && !headerName.equals(MessageHeaders.ID) && !headerName.equals(MessageHeaders.TIMESTAMP)) {
 			this.modified = true;
 			this.headers.remove(headerName);
 		}
diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java
index 94993307eb..9eae40b5f2 100644
--- a/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java
+++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java
@@ -34,30 +34,33 @@ import org.springframework.integration.core.Message;
 public interface MessageStore {
 
 	/**
-	 * Return the Message with the given id, or <i>null</i> if no
-	 * Message with that id exists in the MessageStore.
+	 * Return the Message with the given id, or <i>null</i> if no Message with
+	 * that id exists in the MessageStore.
 	 */
 	Message<?> get(UUID id);
 
 	/**
-	 * Put the provided Message into the MessageStore. Its id will
-	 * be used as an index so that the {@link #get(UUID)} and
-	 * {@link #delete(Object)} behave properly. If available, its
-	 * correlationId header will also be stored so that the
-	 * {@link #list(Object)} method behaves properly.
+	 * Put the provided Message into the MessageStore. The store may need to
+	 * mutate the message internally, and if it does then the return value can
+	 * be different than the input. The id of the return value will be used as
+	 * an index so that the {@link #get(UUID)} and {@link #delete(Object)}
+	 * behave properly. Since messages are immutable, putting the same message
+	 * more than once is a no-op.
+	 * 
+	 * @return the message that was stored
 	 */
 	<T> Message<T> put(Message<T> message);
 
 	/**
-	 * Remove the Message with the given id from the MessageStore,
-	 * if present, and return it. If no Message with that id is
-	 * present in the store, this will return <i>null</i>.
+	 * Remove the Message with the given id from the MessageStore, if present,
+	 * and return it. If no Message with that id is present in the store, this
+	 * will return <i>null</i>.
 	 */
 	Message<?> delete(UUID id);
 
 	/**
-	 * Return all Messages currently in the MessageStore that
-	 * contain the provided correlationId header value.
+	 * Return all Messages currently in the MessageStore that contain the
+	 * provided correlationId header value.
 	 * @see org.springframework.integration.core.MessageHeaders#getCorrelationId()
 	 */
 	List<Message<?>> list(Object correlationId);
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
index 2bfe546982..de9405fd75 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/aggregator/CorrelatingMessageHandlerTests.java
@@ -90,8 +90,8 @@ public class CorrelatingMessageHandlerTests {
         String correlationKey = "key";
         UUID id1 = UUID.randomUUID();
         UUID id2 = UUID.randomUUID();
-        Message<?> message1 = testMessage(correlationKey, id1, 1);
-        Message<?> message2 = testMessage(correlationKey, id2, 2);
+        Message<?> message1 = testMessage(correlationKey, 1);
+        Message<?> message2 = testMessage(correlationKey, 2);
         List<Message<?>> storedMessages = new ArrayList<Message<?>>();
 
         when(store.list(correlationKey)).thenReturn(storedMessages);
@@ -128,10 +128,10 @@ public class CorrelatingMessageHandlerTests {
     @Test
     public void shouldNotPruneWhileCompleting() throws Exception {
         String correlationKey = "key";
-        UUID id1 = UUID.randomUUID();
-        UUID id2 = UUID.randomUUID();
-        final Message<?> message1 = testMessage(correlationKey, id1, 1);
-        final Message<?> message2 = testMessage(correlationKey, id2, 2);
+        final Message<?> message1 = testMessage(correlationKey, 1);
+        final Message<?> message2 = testMessage(correlationKey, 2);
+        UUID id1 = message1.getHeaders().getId();
+        UUID id2 = message2.getHeaders().getId();
         final List<Message<?>> storedMessages = new ArrayList<Message<?>>();
 
         final CountDownLatch bothMessagesHandled = new CountDownLatch(2);
@@ -170,10 +170,9 @@ public class CorrelatingMessageHandlerTests {
         verify(store).delete(id2);
     }
 
-    private Message<?> testMessage(String correllationKey, UUID id, int sequenceNumber) {
-        return MessageBuilder.withPayload("test" + id)
-                .setHeader(MessageHeaders.ID, id)
-                .setCorrelationId(correllationKey)
+    private Message<?> testMessage(String correlationKey, int sequenceNumber) {
+        return MessageBuilder.withPayload("test" + sequenceNumber)
+                .setCorrelationId(correlationKey)
                 .setSequenceNumber(sequenceNumber).build();
     }
 
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java
index 34d9f9939f..16d65c623e 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/ChainParserTests.java
@@ -19,12 +19,14 @@ package org.springframework.integration.config;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThat;
 
 import java.util.List;
 
+import org.hamcrest.Factory;
+import org.hamcrest.Matcher;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.integration.channel.PollableChannel;
@@ -32,6 +34,7 @@ import org.springframework.integration.core.Message;
 import org.springframework.integration.core.MessageChannel;
 import org.springframework.integration.handler.AbstractReplyProducingMessageHandler;
 import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.MessageMatcher;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
 import org.springframework.util.StringUtils;
@@ -42,7 +45,7 @@ import org.springframework.util.StringUtils;
  */
 @ContextConfiguration
 @RunWith(SpringJUnit4ClassRunner.class)
-public class ChainParserTests  {
+public class ChainParserTests {
 
 	@Autowired
 	@Qualifier("filterInput")
@@ -85,12 +88,15 @@ public class ChainParserTests  {
 	@Autowired
 	private PollableChannel numbers;
 
-
 	public static Message<?> successMessage = MessageBuilder.withPayload("success").build();
 
+	@Factory
+	public static Matcher<Message<?>> sameExceptImmutableHeaders(Message<?> expected) {
+		return new MessageMatcher(expected);
+	}
 
 	@Test
-	public void chainWithAcceptingFilter() {	
+	public void chainWithAcceptingFilter() {
 		Message<?> message = MessageBuilder.withPayload("test").build();
 		this.filterInput.send(message);
 		Message<?> reply = this.output.receive(0);
@@ -143,7 +149,7 @@ public class ChainParserTests  {
 		this.beanInput.send(message);
 		Message reply = this.output.receive(3000);
 		assertNotNull(reply);
-		assertEquals(reply, successMessage);
+		assertThat(reply, sameExceptImmutableHeaders(successMessage));
 	}
 
 	@Test
@@ -170,17 +176,16 @@ public class ChainParserTests  {
 		assertEquals(123, reply2.getPayload());
 	}
 
-
 	public static class StubHandler extends AbstractReplyProducingMessageHandler {
 		@Override
 		protected Object handleRequestMessage(Message<?> requestMessage) {
 			return successMessage;
 		}
-		
+
 	}
-	
+
 	public static class StubAggregator {
-		public String aggregate(List<String> strings){
+		public String aggregate(List<String> strings) {
 			return StringUtils.collectionToCommaDelimitedString(strings);
 		}
 	}
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/BridgeParserTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/BridgeParserTests.java
index de698ef91b..5a45ec9772 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/BridgeParserTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/config/xml/BridgeParserTests.java
@@ -16,10 +16,11 @@
 
 package org.springframework.integration.config.xml;
 
-import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThat;
 
+import org.hamcrest.Factory;
+import org.hamcrest.Matcher;
 import org.junit.Test;
-
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.integration.channel.PollableChannel;
@@ -28,6 +29,7 @@ import org.springframework.integration.core.Message;
 import org.springframework.integration.core.MessageChannel;
 import org.springframework.integration.core.MessagingException;
 import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.MessageMatcher;
 import org.springframework.integration.message.StringMessage;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
@@ -59,12 +61,17 @@ public class BridgeParserTests extends AbstractJUnit4SpringContextTests {
 	@Qualifier("output2")
 	private PollableChannel output2;
 
+	@Factory
+    public static Matcher<Message<?>> sameExceptImmutableHeaders(Message<?> expected) {
+        return new MessageMatcher(expected);
+    }
+
 	@Test
 	public void pollableChannel() {
 		Message<?> message = new StringMessage("test1");
 		this.pollableChannel.send(message);
 		Message<?> reply = this.output1.receive(1000);
-		assertEquals(message, reply);
+		assertThat(message, sameExceptImmutableHeaders(reply));
 	}
 
 	@Test
@@ -72,7 +79,7 @@ public class BridgeParserTests extends AbstractJUnit4SpringContextTests {
 		Message<?> message = new StringMessage("test2");
 		this.subscribableChannel.send(message);
 		Message<?> reply = this.output2.receive(0);
-		assertEquals(message, reply);
+		assertThat(message, sameExceptImmutableHeaders(reply));
 	}
 
 	@Test
@@ -81,7 +88,7 @@ public class BridgeParserTests extends AbstractJUnit4SpringContextTests {
 		Message<?> message = MessageBuilder.withPayload("test3").setReplyChannel(replyChannel).build();
 		this.stopperChannel.send(message);
 		Message<?> reply = replyChannel.receive(0);
-		assertEquals(message, reply);
+		assertThat(message, sameExceptImmutableHeaders(reply));
 	}
 
 	@Test(expected = MessagingException.class)
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/filter/MessageFilterTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/filter/MessageFilterTests.java
index e76486c5d8..7e0a4e3cd4 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/filter/MessageFilterTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/filter/MessageFilterTests.java
@@ -18,6 +18,7 @@ package org.springframework.integration.filter;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -48,7 +49,9 @@ public class MessageFilterTests {
 		QueueChannel output = new QueueChannel();
 		filter.setOutputChannel(output);
 		filter.handleMessage(message);
-		assertEquals(message, output.receive(0));
+		Message<?> received = output.receive(0);
+		assertEquals(message.getPayload(), received.getPayload());
+		assertNotSame(message.getHeaders().getId(), received.getHeaders().getId());
 	}
 
 	@Test
@@ -93,7 +96,7 @@ public class MessageFilterTests {
 		assertTrue(inputChannel.send(message));
 		Message<?> reply = outputChannel.receive(0);
 		assertNotNull(reply);
-		assertEquals(message, reply);
+		assertEquals(message.getPayload(), reply.getPayload());
 	}
 
 	@Test
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/handler/BridgeHandlerTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/handler/BridgeHandlerTests.java
index e70bdc0165..a6030d3363 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/handler/BridgeHandlerTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/handler/BridgeHandlerTests.java
@@ -16,17 +16,18 @@
 
 package org.springframework.integration.handler;
 
-import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertThat;
 
+import org.hamcrest.Factory;
+import org.hamcrest.Matcher;
 import org.junit.Test;
 import org.springframework.integration.channel.PollableChannel;
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.core.Message;
 import org.springframework.integration.message.MessageBuilder;
 import org.springframework.integration.message.MessageHandlingException;
+import org.springframework.integration.message.MessageMatcher;
 import org.springframework.integration.message.StringMessage;
 
 /**
@@ -37,6 +38,11 @@ public class BridgeHandlerTests {
 
 	private BridgeHandler handler= new BridgeHandler();
 
+	@Factory
+    public static Matcher<Message<?>> sameExceptImmutableHeaders(Message<?> expected) {
+        return new MessageMatcher(expected);
+    }
+
 	@Test
 	public void simpleBridge() {
 		QueueChannel outputChannel = new QueueChannel();
@@ -45,7 +51,7 @@ public class BridgeHandlerTests {
 		handler.handleMessage(request);
 		Message<?> reply = outputChannel.receive(0);
 		assertNotNull(reply);
-		assertEquals(request, reply);
+		assertThat(reply, sameExceptImmutableHeaders(request));
 	}
 
 	@Test(expected = MessageHandlingException.class)
@@ -60,7 +66,7 @@ public class BridgeHandlerTests {
 		PollableChannel replyChannel = new QueueChannel();
 		Message request = MessageBuilder.withPayload("tst").setReplyChannel(replyChannel ).build();
 		handler.handleMessage(request );
-		assertThat(replyChannel.receive(), is(request));
+		assertThat(replyChannel.receive(), sameExceptImmutableHeaders(request));
 	}
 
 }
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/json/InboundJsonMessageMapperTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/json/InboundJsonMessageMapperTests.java
index 1bafdbc09e..466fa01187 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/json/InboundJsonMessageMapperTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/json/InboundJsonMessageMapperTests.java
@@ -16,7 +16,6 @@
 
 package org.springframework.integration.json;
 
-import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThat;
 import static org.junit.Assert.fail;
@@ -33,18 +32,27 @@ import org.codehaus.jackson.JsonGenerationException;
 import org.codehaus.jackson.map.JsonMappingException;
 import org.codehaus.jackson.map.ObjectMapper;
 import org.codehaus.jackson.type.TypeReference;
+import org.hamcrest.Factory;
+import org.hamcrest.Matcher;
 import org.junit.Test;
 import org.springframework.integration.core.Message;
 import org.springframework.integration.core.MessageHeaders;
 import org.springframework.integration.message.MessageBuilder;
+import org.springframework.integration.message.MessageMatcher;
 
 /**
  * @author Jeremy Grelle
  * @author Mark Fisher
+ * @author Dave Syer
  */
 public class InboundJsonMessageMapperTests {
 
-	ObjectMapper mapper = new ObjectMapper();
+	private ObjectMapper mapper = new ObjectMapper();
+	
+	@Factory
+    public static Matcher<Message<?>> sameExceptImmutableHeaders(Message<?> operand) {
+        return new MessageMatcher(operand);
+    }
 
 	@Test
 	public void testToMessageWithHeadersAndStringPayload() throws Exception {
@@ -53,7 +61,7 @@ public class InboundJsonMessageMapperTests {
 		Message<String> expected = MessageBuilder.withPayload("myPayloadStuff").setHeader(MessageHeaders.TIMESTAMP, new Long(1)).setHeader(MessageHeaders.ID, id).build();
 		InboundJsonMessageMapper mapper = new InboundJsonMessageMapper(String.class);
 		Message<String> result = (Message<String>) mapper.toMessage(jsonMessage);
-		assertThat(result, is(expected));
+		assertThat(result, sameExceptImmutableHeaders(expected));
 	}
 	
 	@Test
@@ -74,7 +82,7 @@ public class InboundJsonMessageMapperTests {
 		Message<TestBean> expected = MessageBuilder.withPayload(bean).setHeader(MessageHeaders.TIMESTAMP, new Long(1)).setHeader(MessageHeaders.ID, id).build();
 		InboundJsonMessageMapper mapper = new InboundJsonMessageMapper(TestBean.class);
 		Message<?> result = mapper.toMessage(jsonMessage);
-		assertEquals(expected, result);
+		assertThat(result, sameExceptImmutableHeaders(expected));
 	}
 	
 	@Test
@@ -99,7 +107,7 @@ public class InboundJsonMessageMapperTests {
 		headerTypes.put("myHeader", TestBean.class);
 		mapper.setHeaderTypes(headerTypes);
 		Message<?> result = mapper.toMessage(jsonMessage);
-		assertEquals(expected, result);
+		assertThat(result, sameExceptImmutableHeaders(expected));
 	}
 	
 	@Test 
@@ -110,7 +118,7 @@ public class InboundJsonMessageMapperTests {
 		Message<List<String>> expected = MessageBuilder.withPayload(expectedList).setHeader(MessageHeaders.TIMESTAMP, new Long(1)).setHeader(MessageHeaders.ID, id).build();
 		InboundJsonMessageMapper mapper = new InboundJsonMessageMapper(new TypeReference<List<String>>(){});
 		Message<?> result = mapper.toMessage(jsonMessage);
-		assertEquals(expected, result);
+		assertThat(result, sameExceptImmutableHeaders(expected));
 	}
 	
 	@Test 
@@ -123,7 +131,7 @@ public class InboundJsonMessageMapperTests {
 		Message<List<TestBean>> expected = MessageBuilder.withPayload(expectedList).setHeader(MessageHeaders.TIMESTAMP, new Long(1)).setHeader(MessageHeaders.ID, id).build();
 		InboundJsonMessageMapper mapper = new InboundJsonMessageMapper(new TypeReference<List<TestBean>>(){});
 		Message<?> result = mapper.toMessage(jsonMessage);
-		assertEquals(expected, result);
+		assertThat(result, sameExceptImmutableHeaders(expected));
 	}
 	
 	@Test
@@ -236,6 +244,6 @@ public class InboundJsonMessageMapperTests {
 		StringWriter writer = new StringWriter();
 		mapper.writeValue(writer, bean);
 		return writer.toString();
-	}
+	};
 
 }
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageBuilderTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageBuilderTests.java
index 07a547891f..36c7a9d5a5 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageBuilderTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageBuilderTests.java
@@ -18,8 +18,10 @@ package org.springframework.integration.message;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotSame;
 
 import java.util.Date;
+import java.util.UUID;
 
 import org.junit.Test;
 
@@ -67,6 +69,24 @@ public class MessageBuilderTests {
 		assertEquals("2", message2.getHeaders().get("bar"));
 	}
 
+	@Test
+	public void testIdHeaderValues() {
+		UUID id = UUID.randomUUID();
+		Message<String> message = MessageBuilder.withPayload("test")
+				.setHeader(MessageHeaders.ID, id)
+				.build();
+		assertNotSame(id, message.getHeaders().getId());
+	}
+
+	@Test
+	public void testTimestampHeaderValues() {
+		Long timestamp = 12345L;
+		Message<String> message = MessageBuilder.withPayload("test")
+				.setHeader(MessageHeaders.TIMESTAMP, timestamp)
+				.build();
+		assertNotSame(timestamp, message.getHeaders().getTimestamp());
+	}
+
 	@Test
 	public void copyHeadersIfAbsent() {
 		Message<String> message1 = MessageBuilder.withPayload("test1")
@@ -88,6 +108,15 @@ public class MessageBuilderTests {
 		assertEquals("bar", message2.getHeaders().get("foo"));
 	}
 
+	@Test
+	public void createIdRegenerated() {
+		Message<String> message1 = MessageBuilder.withPayload("test")
+				.setHeader("foo", "bar").build();
+		Message<String> message2 = MessageBuilder.fromMessage(message1).build();
+		assertEquals("bar", message2.getHeaders().get("foo"));
+		assertNotSame(message1.getHeaders().getId(), message2.getHeaders().getId());
+	}
+
 	@Test
 	public void testPriority() {
 		Message<Integer> importantMessage = MessageBuilder.withPayload(1)
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageHeadersTests.java b/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageHeadersTests.java
index a474f1b726..86647a850d 100644
--- a/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageHeadersTests.java
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageHeadersTests.java
@@ -18,6 +18,7 @@ package org.springframework.integration.message;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNotSame;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -44,6 +45,27 @@ public class MessageHeadersTests {
 		assertNotNull(headers.getTimestamp());
 	}
 
+	@Test
+	public void testTimestampOverwritten() throws Exception {
+		MessageHeaders headers1 = new MessageHeaders(null);
+		Thread.sleep(50L);
+		MessageHeaders headers2 = new MessageHeaders(headers1);
+		assertNotSame(headers1.getTimestamp(), headers2.getTimestamp());
+	}
+
+	@Test
+	public void testIdOverwritten() throws Exception {
+		MessageHeaders headers1 = new MessageHeaders(null);
+		MessageHeaders headers2 = new MessageHeaders(headers1);
+		assertNotSame(headers1.getId(), headers2.getId());
+	}
+
+	@Test
+	public void testId() {
+		MessageHeaders headers = new MessageHeaders(null);
+		assertNotNull(headers.getId());
+	}
+
 	@Test
 	public void testNonTypedAccessOfHeaderValue() {
 		Integer value = new Integer(123);
diff --git a/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageMatcher.java b/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageMatcher.java
new file mode 100644
index 0000000000..d81fe9b959
--- /dev/null
+++ b/org.springframework.integration/src/test/java/org/springframework/integration/message/MessageMatcher.java
@@ -0,0 +1,61 @@
+/**
+ * 
+ */
+package org.springframework.integration.message;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.hamcrest.BaseMatcher;
+import org.hamcrest.Description;
+import org.springframework.integration.core.Message;
+import org.springframework.integration.core.MessageHeaders;
+
+/**
+ * Matcher to make assertions about message equality easier.  Usage:
+ * 
+ * <pre>
+ * &#064;Test
+ * public void testSomething() {
+ *   Message<String> expected = ...;
+ *   Message<String> result = ...;
+ *   assertThat(result, sameExceptImmutableHeaders(expected));
+ * }
+ * 
+ * &#064;Factory
+ * public static Matcher<Message<?>> sameExceptImmutableHeaders(Message<?> expected) {
+ *   return new MessageMatcher(expected);
+ * }
+ * </pre>
+ *
+ * @author Dave Syer
+ *
+ */
+public class MessageMatcher extends BaseMatcher<Message<?>> {
+
+	private final Object payload;
+	private final Map<String, Object> headers;
+
+	public MessageMatcher(Message<?> operand) {
+		this.payload = operand.getPayload();
+		this.headers = getHeaders(operand);
+	}
+
+	private Map<String, Object> getHeaders(Message<?> operand) {
+		HashMap<String, Object> headers = new HashMap<String, Object>(operand.getHeaders());
+		headers.remove(MessageHeaders.ID);
+		headers.remove(MessageHeaders.TIMESTAMP);
+		return headers;
+	}
+
+	public boolean matches(Object arg) {
+		Message<?> input = (Message<?>) arg;
+		Map<String, Object> inputHeaders = getHeaders(input);
+		return input.getPayload().equals(payload) && inputHeaders.equals(headers);
+	}
+
+	public void describeTo(Description description) {
+		description.appendText("Headers match except ID and timestamp for payload: ").appendValue(payload).appendText(" and headers: ").appendValue(headers);
+	}
+	
+}
\ No newline at end of file