Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

RESOLVED - issue INT-1097

Browse Source
This commit is contained in:
David Syer
2010-04-26 15:23:53 +00:00
parent 417982bf03
commit dd4afd8cb5
6 changed files with 70 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
View File

@@ -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<Message<?>> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { id }, new MessageMapper());
public Message<?> get(UUID id) {
List<Message<?>> 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<Message<?>> 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 <T> Message<T> put(final Message<T> 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();
}
}

3
org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql
View File

@@ -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

10
org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
View File

@@ -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<String> 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<String> 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()));
}
}

5
org.springframework.integration.jdbc/src/test/resources/log4j.properties
View File

@@ -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
log4j.category.org.springframework.integration.jdbc=DEBUG
log4j.category.org.springframework.jdbc=TRACE

8
org.springframework.integration/src/main/java/org/springframework/integration/store/MessageStore.java
View File

@@ -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 <i>null</i> 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 <i>null</i>.
*/
Message<?> delete(Object id);
Message<?> delete(UUID id);
/**
* Return all Messages currently in the MessageStore.

19
org.springframework.integration/src/main/java/org/springframework/integration/store/SimpleMessageStore.java
View File

@@ -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<Object, Message<?>> map;
private final Map<UUID, Message<?>> 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<Object, Message<?>>();
this.map = new ConcurrentHashMap<UUID, Message<?>>();
this.upperBound = new UpperBound(capacity);
}
@@ -65,7 +66,7 @@ public class SimpleMessageStore implements MessageStore {
return (Message<T>) 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<Message<?>>(this.map.values());
}
public Message<?> delete(Object key) {
public Message<?> delete(UUID key) {
if (key != null) {
upperBound.release();
return this.map.remove(key);