From dd4afd8cb599574b4ecc3d15687db351d7d5104f Mon Sep 17 00:00:00 2001 From: David Syer Date: Mon, 26 Apr 2010 15:23:53 +0000 Subject: [PATCH] RESOLVED - issue INT-1097 --- .../integration/jdbc/JdbcMessageStore.java | 68 ++++++++++++------- .../integration/jdbc/schema-derby.sql | 3 +- .../jdbc/JdbcMessageStoreTests.java | 10 +-- .../src/test/resources/log4j.properties | 5 +- .../integration/store/MessageStore.java | 8 ++- .../integration/store/SimpleMessageStore.java | 19 +++--- 6 files changed, 70 insertions(+), 43 deletions(-) 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 2f1d9e850c..1a1dedad89 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 @@ -8,9 +8,12 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Types; import java.util.List; +import java.util.UUID; 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; @@ -37,21 +40,23 @@ import org.springframework.util.StringUtils; */ public class JdbcMessageStore implements MessageStore { + private static final Log logger = LogFactory.getLog(JdbcMessageStore.class); + /** * Default value for the table prefix property. */ 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 STORE_ID, MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES, VERSION from %PREFIX%MESSAGE where CORRELATION_KEY=?"; - private static final String LIST_ALL_MESSAGES = "SELECT MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES, VERSION from %PREFIX%MESSAGE"; + private static final String LIST_ALL_MESSAGES = "SELECT STORE_ID, MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES, VERSION from %PREFIX%MESSAGE"; - 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 STORE_ID, MESSAGE_ID, CORRELATION_KEY, MESSAGE_BYTES, VERSION 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 CREATE_MESSAGE = "INSERT into %PREFIX%MESSAGE(STORE_ID, 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=?"; @@ -174,14 +179,14 @@ public class JdbcMessageStore implements MessageStore { Assert.state(incrementer != null, "A DataFieldMaxValueIncrementer must be provided"); } - public Message delete(Object id) { + public Message delete(UUID id) { Message message = get(id); if (message == null) { return null; } - int updated = jdbcTemplate.update(getQuery(DELETE_MESSAGE), new Object[] { id }, new int[] { Types.BIGINT }); + int updated = jdbcTemplate.update(getQuery(DELETE_MESSAGE), new Object[] { getKey(id) }, new int[] { Types.VARCHAR }); if (updated != 0) { return message; @@ -191,8 +196,9 @@ public class JdbcMessageStore implements MessageStore { } - public Message get(Object id) { - List> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { id }, new MessageMapper()); + public Message get(UUID id) { + List> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { getKey(id) }, + new MessageMapper()); if (list.isEmpty()) { return null; } @@ -204,8 +210,8 @@ public class JdbcMessageStore implements MessageStore { } public List> list(Object correlationId) { - return jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_CORRELATION_KEY), - new Object[] { getCorrelationKey(correlationId) }, new MessageMapper()); + return jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_CORRELATION_KEY), new Object[] { getKey(correlationId) }, + new MessageMapper()); } public Message put(final Message message) { @@ -219,23 +225,26 @@ public class JdbcMessageStore implements MessageStore { id = (Long) message.getHeaders().get(ID_KEY); - final String correlationId = getCorrelationKey(message.getHeaders().getCorrelationId()); + 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 + " and surrogate id=" + id); ps.setString(1, correlationId); lobHandler.getLobCreator().setBlobAsBytes(ps, 2, messageBytes); ps.setInt(3, version + 1); ps.setInt(4, version); - ps.setLong(5, id); + ps.setString(5, messageId); } }); if (updated != 1) { int currentVersion = jdbcTemplate.queryForInt(getQuery(CURRENT_VERSION_MESSAGE), new Object[] { id }); - throw new OptimisticLockingFailureException("Attempt to update message id=" + id - + " with wrong version (" + version + "), where current version is " + currentVersion); + throw new OptimisticLockingFailureException("Attempt to update message id=" + + message.getHeaders().getId() + " with wrong version (" + version + + "), where current version is " + currentVersion); } } @@ -243,15 +252,18 @@ public class JdbcMessageStore implements MessageStore { id = incrementer.nextLongValue(); - final String correlationId = getCorrelationKey(message.getHeaders().getCorrelationId()); + 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 + " and surrogate id=" + id); ps.setLong(1, id); - ps.setString(2, correlationId); - lobHandler.getLobCreator().setBlobAsBytes(ps, 3, messageBytes); - ps.setInt(4, version); + ps.setString(2, messageId); + ps.setString(3, correlationId); + lobHandler.getLobCreator().setBlobAsBytes(ps, 4, messageBytes); + ps.setInt(5, version); } }); @@ -261,12 +273,20 @@ public class JdbcMessageStore implements MessageStore { } - private String getCorrelationKey(Object correlationId) { + private String getKey(Object input) { - if (correlationId == null) { + if (input == null) { return null; } + if (input instanceof UUID) { + return input.toString(); + } + + if (input instanceof String && ((String) input).length() < 100) { + return (String) input; + } + MessageDigest digest; try { digest = MessageDigest.getInstance("MD5"); @@ -275,7 +295,7 @@ public class JdbcMessageStore implements MessageStore { throw new IllegalStateException("MD5 algorithm not available. Fatal (should be in the JDK)."); } - byte[] bytes = digest.digest(SerializationUtils.serialize(correlationId)); + byte[] bytes = digest.digest(SerializationUtils.serialize(input)); return String.format("%032x", new BigInteger(1, bytes)); } @@ -293,8 +313,8 @@ 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(ID_KEY, rs.getLong("MESSAGE_ID")).setHeader( - VERSION_KEY, rs.getInt("VERSION")).build(); + return MessageBuilder.fromMessage(message).setHeader(ID_KEY, rs.getLong("STORE_ID")).setHeader(VERSION_KEY, + rs.getInt("VERSION")).build(); } } 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 0cae9d09b7..6eacf112ab 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,5 +1,6 @@ CREATE TABLE INT_MESSAGE ( - MESSAGE_ID BIGINT NOT NULL PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY, + STORE_ID BIGINT NOT NULL PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY, + MESSAGE_ID VARCHAR(100), CORRELATION_KEY VARCHAR(100), MESSAGE_BYTES BLOB, VERSION BIGINT 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 4f5f940e87..8dfc616c91 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 @@ -4,6 +4,8 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotNull; import static org.junit.Assert.assertNull; +import java.util.UUID; + import javax.sql.DataSource; import org.junit.Before; @@ -45,7 +47,7 @@ public class JdbcMessageStoreTests { @Test @Transactional public void testGetNonExistent() throws Exception { - Message result = messageStore.get(12345); + Message result = messageStore.get(UUID.randomUUID()); assertNull(result); } @@ -54,7 +56,7 @@ public class JdbcMessageStoreTests { public void testAddAndGet() throws Exception { Message message = MessageBuilder.withPayload("foo").setCorrelationId("X").build(); message = messageStore.put(message); - Message result = messageStore.get(message.getHeaders().get(JdbcMessageStore.ID_KEY)); + Message result = messageStore.get(message.getHeaders().getId()); assertNotNull(result); assertEquals(message.getPayload(), result.getPayload()); } @@ -66,7 +68,7 @@ public class JdbcMessageStoreTests { message = messageStore.put(message); message = MessageBuilder.fromMessage(message).setCorrelationId("Y").build(); message = messageStore.put(message); - assertEquals("Y", messageStore.get(message.getHeaders().get(JdbcMessageStore.ID_KEY)).getHeaders().getCorrelationId()); + assertEquals("Y", messageStore.get(message.getHeaders().getId()).getHeaders().getCorrelationId()); } @Test @@ -91,7 +93,7 @@ public class JdbcMessageStoreTests { public void testAddAndDelete() throws Exception { Message message = MessageBuilder.withPayload("foo").setCorrelationId("X").build(); message = messageStore.put(message); - assertNotNull(messageStore.delete(message.getHeaders().get(JdbcMessageStore.ID_KEY))); + assertNotNull(messageStore.delete(message.getHeaders().getId())); } } diff --git a/org.springframework.integration.jdbc/src/test/resources/log4j.properties b/org.springframework.integration.jdbc/src/test/resources/log4j.properties index ba5b291f5c..d40675320b 100644 --- a/org.springframework.integration.jdbc/src/test/resources/log4j.properties +++ b/org.springframework.integration.jdbc/src/test/resources/log4j.properties @@ -2,9 +2,10 @@ log4j.rootCategory=WARN, stdout log4j.appender.stdout=org.apache.log4j.ConsoleAppender log4j.appender.stdout.layout=org.apache.log4j.PatternLayout -log4j.appender.stdout.layout.ConversionPattern=%c{1}: %m%n +log4j.appender.stdout.layout.ConversionPattern=%d{ABSOLUTE} %5p %t %c{2}:%L - %m%n log4j.category.org.springframework=DEBUG log4j.category.org.springframework.integration=INFO -log4j.category.org.springframework.integration.jdbc=DEBUG \ No newline at end of file +log4j.category.org.springframework.integration.jdbc=DEBUG +log4j.category.org.springframework.jdbc=TRACE \ No newline at end of file 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 efe5140270..a9229eec2f 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 @@ -17,6 +17,7 @@ package org.springframework.integration.store; import java.util.List; +import java.util.UUID; import org.springframework.integration.core.Message; @@ -27,6 +28,7 @@ import org.springframework.integration.core.Message; * * @author Mark Fisher * @author Iwein Fuld + * @author Dave Syer * @since 2.0 */ public interface MessageStore { @@ -35,11 +37,11 @@ public interface MessageStore { * Return the Message with the given id, or null if no * Message with that id exists in the MessageStore. */ - Message get(Object id); + Message get(UUID id); /** * Put the provided Message into the MessageStore. Its id will - * be used as an index so that the {@link #get(Object)} and + * 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. @@ -51,7 +53,7 @@ public interface MessageStore { * if present, and return it. If no Message with that id is * present in the store, this will return null. */ - Message delete(Object id); + Message delete(UUID id); /** * Return all Messages currently in the MessageStore. diff --git a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java index 2edeca353b..be4e74c614 100644 --- a/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java +++ b/org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java @@ -16,17 +16,18 @@ package org.springframework.integration.store; -import org.springframework.integration.core.Message; -import org.springframework.integration.core.MessagingException; -import org.springframework.integration.util.UpperBound; -import org.springframework.util.Assert; - import java.util.ArrayList; import java.util.Collection; import java.util.List; import java.util.Map; +import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import org.springframework.integration.core.Message; +import org.springframework.integration.core.MessagingException; +import org.springframework.integration.util.UpperBound; +import org.springframework.util.Assert; + /** * Map-based implementation of {@link MessageStore} that enforces a maximum capacity. * @@ -36,7 +37,7 @@ import java.util.concurrent.ConcurrentHashMap; */ public class SimpleMessageStore implements MessageStore { - private final Map> map; + private final Map> map; private final UpperBound upperBound; /** @@ -44,7 +45,7 @@ public class SimpleMessageStore implements MessageStore { * size if the given capacity is less than 1. */ public SimpleMessageStore(int capacity) { - this.map = new ConcurrentHashMap>(); + this.map = new ConcurrentHashMap>(); this.upperBound = new UpperBound(capacity); } @@ -65,7 +66,7 @@ public class SimpleMessageStore implements MessageStore { return (Message) this.map.put(message.getHeaders().getId(), message); } - public Message get(Object key) { + public Message get(UUID key) { return (key != null) ? this.map.get(key) : null; } @@ -73,7 +74,7 @@ public class SimpleMessageStore implements MessageStore { return new ArrayList>(this.map.values()); } - public Message delete(Object key) { + public Message delete(UUID key) { if (key != null) { upperBound.release(); return this.map.remove(key);