diff --git a/org.springframework.integration.jdbc/.springBeans b/org.springframework.integration.jdbc/.springBeans new file mode 100644 index 0000000000..34b281cf1f --- /dev/null +++ b/org.springframework.integration.jdbc/.springBeans @@ -0,0 +1,13 @@ + + + 1 + + + + + + + + + + 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 new file mode 100644 index 0000000000..2f1d9e850c --- /dev/null +++ b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -0,0 +1,302 @@ +package org.springframework.integration.jdbc; + +import java.math.BigInteger; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Types; +import java.util.List; + +import javax.sql.DataSource; + +import org.springframework.dao.OptimisticLockingFailureException; +import org.springframework.integration.core.Message; +import org.springframework.integration.jdbc.util.SerializationUtils; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.integration.store.MessageStore; +import org.springframework.jdbc.core.JdbcOperations; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.core.PreparedStatementSetter; +import org.springframework.jdbc.core.RowMapper; +import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer; +import org.springframework.jdbc.support.lob.DefaultLobHandler; +import org.springframework.jdbc.support.lob.LobHandler; +import org.springframework.util.Assert; +import org.springframework.util.StringUtils; + +/** + * Implementation of {@link MessageStore} using a relational database via JDBC. + * SQL scripts to create the necessary tables are packaged as + * org/springframework/integration/jdbc/schema-*.sql, where + * * is the target database type. + * + * @author Dave Syer + * + */ +public class JdbcMessageStore implements MessageStore { + + /** + * 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_ALL_MESSAGES = "SELECT 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 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=?"; + + 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 + */ + 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"; + + private String tablePrefix = DEFAULT_TABLE_PREFIX; + + private JdbcOperations jdbcTemplate; + + private DataFieldMaxValueIncrementer incrementer; + + private LobHandler lobHandler = new DefaultLobHandler(); + + /** + * Convenient constructor for configuration use. + */ + public JdbcMessageStore() { + } + + /** + * Create a {@link MessageStore} with all mandatory properties. + * + * @param dataSource a {@link DataSource} + * @param incrementer a {@link DataFieldMaxValueIncrementer} + */ + public JdbcMessageStore(DataSource dataSource, DataFieldMaxValueIncrementer incrementer) { + this.incrementer = incrementer; + jdbcTemplate = new JdbcTemplate(dataSource); + } + + /** + * Replace patterns in the input to produce a valid SQL query. This + * implementation replaces the table prefix. + * + * @param base the SQL query to be transformed + * @return a transformed query with replacements + */ + protected String getQuery(String base) { + return StringUtils.replace(base, "%PREFIX%", tablePrefix); + } + + /** + * Public setter for the table prefix property. This will be prefixed to all + * the table names before queries are executed. Defaults to + * {@link #DEFAULT_TABLE_PREFIX}. + * + * @param tablePrefix the tablePrefix to set + */ + public void setTablePrefix(String tablePrefix) { + this.tablePrefix = tablePrefix; + } + + /** + * The JDBC {@link DataSource} to use when interacting with the database. + * Either this property can be set or the + * {@link #setJdbcTemplate(JdbcOperations) jdbcTemplate}. + * + * @param dataSource a {@link DataSource} + */ + public void setDataSource(DataSource dataSource) { + jdbcTemplate = new JdbcTemplate(dataSource); + } + + /** + * The {@link JdbcOperations} to use when interacting with the database. + * Either this property can be set or the {@link #setDataSource(DataSource) + * dataSource}. + * + * @param dataSource a {@link DataSource} + */ + public void setJdbcTemplate(JdbcOperations jdbcTemplate) { + this.jdbcTemplate = jdbcTemplate; + } + + /** + * Setter for {@link DataFieldMaxValueIncrementer} to be used when + * generating primary keys for {@link Message} instances. The message store + * manages its own surrogate keys for messages to avoid any ambiguity. When + * you put a message in the store it should come back as a new instance with + * a header called {@link #ID_KEY}. + * + * @param incrementer the {@link DataFieldMaxValueIncrementer} + */ + public void setIncrementer(DataFieldMaxValueIncrementer incrementer) { + this.incrementer = incrementer; + } + + /** + * Override the {@link LobHandler} that is used to create and unpack large + * objects in SQL queries. The default is fine for almost all platforms, but + * some Oracle drivers require a native implementation. + * + * @param lobHandler a {@link LobHandler} + */ + public void setLobHandler(LobHandler lobHandler) { + this.lobHandler = lobHandler; + } + + /** + * Check mandatory properties (data source and incrementer). + * + * @throws Exception + */ + public void afterPropertiesSet() throws Exception { + Assert.state(jdbcTemplate != null, "A DataSource or JdbcTemplate must be provided"); + Assert.state(incrementer != null, "A DataFieldMaxValueIncrementer must be provided"); + } + + public Message delete(Object id) { + + Message message = get(id); + if (message == null) { + return null; + } + + int updated = jdbcTemplate.update(getQuery(DELETE_MESSAGE), new Object[] { id }, new int[] { Types.BIGINT }); + + if (updated != 0) { + return message; + } + + return null; + + } + + public Message get(Object id) { + List> list = jdbcTemplate.query(getQuery(GET_MESSAGE), new Object[] { id }, new MessageMapper()); + if (list.isEmpty()) { + return null; + } + return list.get(0); + } + + public List> list() { + return jdbcTemplate.query(getQuery(LIST_ALL_MESSAGES), new MessageMapper()); + } + + public List> list(Object correlationId) { + return jdbcTemplate.query(getQuery(LIST_MESSAGES_BY_CORRELATION_KEY), + new Object[] { getCorrelationKey(correlationId) }, new MessageMapper()); + } + + public Message put(final Message message) { + + final int version = message.getHeaders().containsKey(VERSION_KEY) ? (Integer) message.getHeaders().get( + VERSION_KEY) : 0; + + final long id; + + if (message.getHeaders().containsKey(ID_KEY)) { + + id = (Long) message.getHeaders().get(ID_KEY); + + final String correlationId = getCorrelationKey(message.getHeaders().getCorrelationId()); + final byte[] messageBytes = SerializationUtils.serialize(message); + + int updated = jdbcTemplate.update(getQuery(UPDATE_MESSAGE), new PreparedStatementSetter() { + public void setValues(PreparedStatement ps) throws SQLException { + ps.setString(1, correlationId); + lobHandler.getLobCreator().setBlobAsBytes(ps, 2, messageBytes); + ps.setInt(3, version + 1); + ps.setInt(4, version); + ps.setLong(5, id); + } + }); + + 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); + } + + } + else { + + id = incrementer.nextLongValue(); + + final String correlationId = getCorrelationKey(message.getHeaders().getCorrelationId()); + final byte[] messageBytes = SerializationUtils.serialize(message); + + jdbcTemplate.update(getQuery(CREATE_MESSAGE), new PreparedStatementSetter() { + public void setValues(PreparedStatement ps) throws SQLException { + ps.setLong(1, id); + ps.setString(2, correlationId); + lobHandler.getLobCreator().setBlobAsBytes(ps, 3, messageBytes); + ps.setInt(4, version); + } + }); + + } + + return MessageBuilder.fromMessage(message).setHeader(ID_KEY, id).setHeader(VERSION_KEY, version).build(); + + } + + private String getCorrelationKey(Object correlationId) { + + if (correlationId == null) { + return null; + } + + MessageDigest digest; + try { + digest = MessageDigest.getInstance("MD5"); + } + catch (NoSuchAlgorithmException e) { + throw new IllegalStateException("MD5 algorithm not available. Fatal (should be in the JDK)."); + } + + byte[] bytes = digest.digest(SerializationUtils.serialize(correlationId)); + return String.format("%032x", new BigInteger(1, bytes)); + + } + + /** + * Convenience class to be used to unpack a message from a result set row. + * Uses column named in the result set to extract the required data, so that + * select clause ordering is unimportant. + * + * @author Dave Syer + * + */ + private class MessageMapper implements RowMapper> { + + 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(); + } + + } + +} diff --git a/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/util/SerializationUtils.java b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/util/SerializationUtils.java new file mode 100644 index 0000000000..400597e7d7 --- /dev/null +++ b/org.springframework.integration.jdbc/src/main/java/org/springframework/integration/jdbc/util/SerializationUtils.java @@ -0,0 +1,78 @@ +/* + * Copyright 2006-2010 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jdbc.util; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; + + +/** + * Static utility to help with serialization. + * + * @author Dave Syer + * + */ +public class SerializationUtils { + + /** + * Serialize the object provided. + * + * @param object the object to serialize + * @return an array of bytes representing the object in a portable fashion + */ + public static byte[] serialize(Object object) { + + if (object==null) { + return null; + } + + ByteArrayOutputStream stream = new ByteArrayOutputStream(); + try { + new ObjectOutputStream(stream).writeObject(object); + } catch (IOException e) { + throw new IllegalArgumentException("Could not serialize object of type: "+object.getClass(), e); + } + + return stream.toByteArray(); + + } + + /** + * @param bytes a serialized object created + * @return the result of deserializing the bytes + */ + public static Object deserialize(byte[] bytes) { + + if (bytes==null) { + return null; + } + + try { + return new ObjectInputStream(new ByteArrayInputStream(bytes)).readObject(); + } + catch (IOException e) { + throw new IllegalArgumentException("Could not deserialize object", e); + } + catch (ClassNotFoundException e) { + throw new IllegalStateException("Could not deserialize object type", e); + } + + } + +} 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 new file mode 100644 index 0000000000..0cae9d09b7 --- /dev/null +++ b/org.springframework.integration.jdbc/src/main/resources/org/springframework/integration/jdbc/schema-derby.sql @@ -0,0 +1,7 @@ +CREATE TABLE INT_MESSAGE ( + MESSAGE_ID BIGINT NOT NULL PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY, + CORRELATION_KEY VARCHAR(100), + MESSAGE_BYTES BLOB, + VERSION BIGINT +); +CREATE TABLE INT_MESSAGE_SEQ (ID BIGINT PRIMARY KEY GENERATED BY DEFAULT AS IDENTITY, DUMMY VARCHAR(1)); 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 new file mode 100644 index 0000000000..4f5f940e87 --- /dev/null +++ b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -0,0 +1,97 @@ +package org.springframework.integration.jdbc; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import javax.sql.DataSource; + +import org.junit.Before; +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.core.Message; +import org.springframework.integration.message.MessageBuilder; +import org.springframework.jdbc.core.simple.SimpleJdbcTemplate; +import org.springframework.jdbc.support.incrementer.DataFieldMaxValueIncrementer; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.junit4.SpringJUnit4ClassRunner; +import org.springframework.test.jdbc.SimpleJdbcTestUtils; +import org.springframework.transaction.annotation.Transactional; + +@ContextConfiguration +@RunWith(SpringJUnit4ClassRunner.class) +public class JdbcMessageStoreTests { + + @Autowired + private DataSource dataSource; + + @Autowired + @Qualifier("messageIncrementer") + private DataFieldMaxValueIncrementer messageIncrementer; + + + private JdbcMessageStore messageStore; + + private SimpleJdbcTemplate jdbcTemplate; + + @Before + public void init() { + messageStore = new JdbcMessageStore(dataSource, messageIncrementer); + jdbcTemplate = new SimpleJdbcTemplate(dataSource); + } + + @Test + @Transactional + public void testGetNonExistent() throws Exception { + Message result = messageStore.get(12345); + assertNull(result); + } + + @Test + @Transactional + 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)); + assertNotNull(result); + assertEquals(message.getPayload(), result.getPayload()); + } + + @Test + @Transactional + public void testAddAndUpdate() throws Exception { + Message message = MessageBuilder.withPayload("foo").setCorrelationId("X").build(); + 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()); + } + + @Test + @Transactional + public void testAddAndList() throws Exception { + Message message = MessageBuilder.withPayload("foo").build(); + messageStore.put(message); + assertEquals(1, SimpleJdbcTestUtils.countRowsInTable(jdbcTemplate, "INT_MESSAGE")); + assertEquals(1, messageStore.list().size()); + } + + @Test + @Transactional + public void testAddAndListByCorrelationId() throws Exception { + Message message = MessageBuilder.withPayload("foo").setCorrelationId("X").build(); + messageStore.put(message); + assertEquals(1, messageStore.list("X").size()); + } + + @Test + @Transactional + 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))); + } + +} diff --git a/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/util/SerializationUtilsTests.java b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/util/SerializationUtilsTests.java new file mode 100644 index 0000000000..ab5872759f --- /dev/null +++ b/org.springframework.integration.jdbc/src/test/java/org/springframework/integration/jdbc/util/SerializationUtilsTests.java @@ -0,0 +1,70 @@ +/* + * Copyright 2006-2010 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.integration.jdbc.util; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; + +import java.math.BigInteger; + +import org.junit.Test; + +/** + * Test for static utility to help with serialization. + * + * @author Dave Syer + * + */ +public class SerializationUtilsTests { + + private static BigInteger FOO = new BigInteger( + "-9702942423549012526722364838327831379660941553432801565505143675386108883970811292563757558516603356009681061" + + "5697574744209306031461371833798723505120163874786203211176873686513374052845353833564048"); + + @Test + public void testSerializeCycleSunnyDay() throws Exception { + assertEquals("foo", SerializationUtils.deserialize(SerializationUtils.serialize("foo"))); + } + + @Test(expected = IllegalStateException.class) + public void testDeserializeUndefined() throws Exception { + byte[] bytes = FOO.toByteArray(); + Object foo = SerializationUtils.deserialize(bytes); + assertNotNull(foo); + } + + @Test(expected = IllegalArgumentException.class) + public void testSerializeNonSerializable() throws Exception { + SerializationUtils.serialize(new Object()); + } + + @Test(expected = IllegalArgumentException.class) + public void testDeserializeNonSerializable() throws Exception { + SerializationUtils.deserialize("foo".getBytes()); + } + + @Test + public void testSerializeNull() throws Exception { + assertNull(SerializationUtils.serialize(null)); + } + + @Test + public void testDeserializeNull() throws Exception { + assertNull(SerializationUtils.deserialize(null)); + } + +} diff --git a/org.springframework.integration.jdbc/src/test/resources/int-derby.properties b/org.springframework.integration.jdbc/src/test/resources/int-derby.properties new file mode 100644 index 0000000000..c971c9fe09 --- /dev/null +++ b/org.springframework.integration.jdbc/src/test/resources/int-derby.properties @@ -0,0 +1,3 @@ +# Placeholders for Derby: +int.schema.script=classpath:/org/springframework/integration/jdbc/schema-derby.sql +int.database.incrementer.class=org.springframework.jdbc.support.incrementer.DerbyMaxValueIncrementer diff --git a/org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml b/org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml new file mode 100644 index 0000000000..3e7f56ad91 --- /dev/null +++ b/org.springframework.integration.jdbc/src/test/resources/org/springframework/integration/jdbc/JdbcMessageStoreTests-context.xml @@ -0,0 +1,36 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + +