diff --git a/spring-integration-jdbc/pom.xml b/spring-integration-jdbc/pom.xml index 64de7a0f75..2389906543 100644 --- a/spring-integration-jdbc/pom.xml +++ b/spring-integration-jdbc/pom.xml @@ -32,6 +32,10 @@ org.springframework.integration spring-integration-core + + org.springframework.commons + spring-commons-serializer + cglib diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java index ef71577a01..9b38c20ad2 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java @@ -26,8 +26,12 @@ import javax.sql.DataSource; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.springframework.commons.serializer.DeserializingConverter; +import org.springframework.commons.serializer.InputStreamingConverter; +import org.springframework.commons.serializer.OutputStreamingConverter; +import org.springframework.commons.serializer.SerializingConverter; +import org.springframework.commons.serializer.java.JavaStreamingConverter; import org.springframework.integration.Message; -import org.springframework.integration.jdbc.util.SerializationUtils; import org.springframework.integration.store.AbstractMessageGroupStore; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageStore; @@ -103,6 +107,10 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private String tablePrefix = DEFAULT_TABLE_PREFIX; private JdbcOperations jdbcTemplate; + + private DeserializingConverter deserializer; + + private SerializingConverter serializer; private LobHandler lobHandler = new DefaultLobHandler(); @@ -112,6 +120,9 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa * Convenient constructor for configuration use. */ public JdbcMessageStore() { + JavaStreamingConverter converter = new JavaStreamingConverter(); + deserializer = new DeserializingConverter(converter); + serializer = new SerializingConverter(converter); } /** @@ -120,6 +131,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa * @param dataSource a {@link DataSource} */ public JdbcMessageStore(DataSource dataSource) { + this(); jdbcTemplate = new JdbcTemplate(dataSource); } @@ -182,6 +194,26 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa public void setLobHandler(LobHandler lobHandler) { this.lobHandler = lobHandler; } + + /** + * A converter for serializing messages to byte arrays for storage. + * + * @param serializer the serializer to set + */ + @SuppressWarnings("unchecked") + public void setSerializer(OutputStreamingConverter> serializer) { + this.serializer = new SerializingConverter((OutputStreamingConverter) serializer); + } + + /** + * A converter for deserializing byte arrays to messages. + * + * @param deserializer the deserializer to set + */ + @SuppressWarnings("unchecked") + public void setDeserializer(InputStreamingConverter> deserializer) { + this.deserializer = new DeserializingConverter((InputStreamingConverter) deserializer); + } /** * Check mandatory properties (data source and incrementer). @@ -228,7 +260,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa Message result = MessageBuilder.fromMessage(message).setHeader(SAVED_KEY, Boolean.TRUE).setHeader( CREATED_DATE_KEY, new Long(createdDate)).build(); final String messageId = getKey(result.getHeaders().getId()); - final byte[] messageBytes = SerializationUtils.serialize(result); + final byte[] messageBytes = serializer.convert(result); jdbcTemplate.update(getQuery(CREATE_MESSAGE), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { @@ -247,7 +279,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa final long createdDate = System.currentTimeMillis(); final String messageId = getKey(message.getHeaders().getId()); final String groupKey = getKey(groupId); - final byte[] messageBytes = SerializationUtils.serialize(message); + final byte[] messageBytes = serializer.convert(message); jdbcTemplate.update(getQuery(CREATE_MESSAGE_IN_GROUP), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { @@ -365,7 +397,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa private class MessageMapper implements RowMapper> { public Message mapRow(ResultSet rs, int rowNum) throws SQLException { - Message message = (Message) SerializationUtils.deserialize(lobHandler.getBlobAsBytes(rs, + Message message = (Message) deserializer.convert(lobHandler.getBlobAsBytes(rs, "MESSAGE_BYTES")); return message; } diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParser.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParser.java index bd520b05dc..2cdc8029bc 100644 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParser.java +++ b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParser.java @@ -55,6 +55,8 @@ public class JdbcMessageStoreParser extends AbstractBeanDefinitionParser { } IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "lob-handler"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "serializer"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "deserializer"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "table-prefix", "tablePrefix"); IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "region", "region"); diff --git a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/util/SerializationUtils.java b/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/util/SerializationUtils.java deleted file mode 100644 index c7c1b66bed..0000000000 --- a/spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/util/SerializationUtils.java +++ /dev/null @@ -1,71 +0,0 @@ -/* - * 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/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd index 4ce14d0b29..b122e8fd05 100644 --- a/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd +++ b/spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd @@ -90,6 +90,30 @@ + + + + + + + + + + + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java index a23f36daa5..61518e5f18 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java @@ -24,6 +24,11 @@ import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; import static org.springframework.integration.test.matcher.PayloadAndHeaderMatcher.sameExceptIgnorableHeaders; +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.io.OutputStream; import java.util.Iterator; import java.util.UUID; @@ -33,7 +38,10 @@ import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.commons.serializer.InputStreamingConverter; +import org.springframework.commons.serializer.OutputStreamingConverter; import org.springframework.integration.Message; +import org.springframework.integration.message.GenericMessage; import org.springframework.integration.store.MessageGroup; import org.springframework.integration.store.MessageGroupCallback; import org.springframework.integration.store.MessageGroupStore; @@ -76,6 +84,30 @@ public class JdbcMessageStoreTests { assertNotNull(result.getHeaders().get(JdbcMessageStore.CREATED_DATE_KEY)); } + @Test + @Transactional + public void testSerializer() throws Exception { + // N.B. these serializers are not realistic (just for test purposes) + messageStore.setSerializer(new OutputStreamingConverter>() { + public void convert(Message object, OutputStream outputStream) throws IOException { + outputStream.write(object.getPayload().toString().getBytes()); + outputStream.flush(); + } + }); + messageStore.setDeserializer(new InputStreamingConverter>() { + public Message convert(InputStream inputStream) throws IOException { + BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream)); + return new GenericMessage(reader.readLine()); + } + }); + Message message = MessageBuilder.withPayload("foo").build(); + Message saved = messageStore.addMessage(message); + assertNull(messageStore.getMessage(message.getHeaders().getId())); + Message result = messageStore.getMessage(saved.getHeaders().getId()); + assertNotNull(result); + assertEquals("foo", result.getPayload()); + } + @Test @Transactional public void testAddAndGetWithDifferentRegion() throws Exception { diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java index b48875bad7..e63637ef7c 100644 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java @@ -3,11 +3,19 @@ package org.springframework.integration.jdbc.config; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertTrue; +import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; + import org.junit.After; import org.junit.Test; +import org.springframework.commons.serializer.java.JavaStreamingConverter; import org.springframework.context.support.ClassPathXmlApplicationContext; +import org.springframework.integration.Message; import org.springframework.integration.jdbc.JdbcMessageStore; import org.springframework.integration.store.MessageStore; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.test.util.TestUtils; import org.springframework.jdbc.support.lob.LobHandler; import org.springframework.test.util.ReflectionTestUtils; @@ -29,7 +37,16 @@ public class JdbcMessageStoreParserTests { assertTrue(store instanceof JdbcMessageStore); } - @After + @Test + public void testSimpleMessageStoreWithSerializer() { + setUp("serializerJdbcMessageStore.xml", getClass()); + MessageStore store = context.getBean("messageStore", MessageStore.class); + Object serializer = TestUtils.getPropertyValue(store, "serializer.streamingConverter"); + assertTrue(serializer instanceof EnhancedSerializer); + Object deserializer = TestUtils.getPropertyValue(store, "deserializer.streamingConverter"); + assertTrue(deserializer instanceof EnhancedSerializer); + } + @Test public void testMessageStoreWithAttributes() { setUp("soupedUpJdbcMessageStore.xml", getClass()); @@ -50,4 +67,18 @@ public class JdbcMessageStoreParserTests { context = new ClassPathXmlApplicationContext(name, cls); } + public static class EnhancedSerializer extends JavaStreamingConverter { + @Override + public Object convert(InputStream inputStream) throws IOException { + Message message = (Message) super.convert(inputStream); + return message; + } + + @Override + public void convert(Object object, OutputStream outputStream) throws IOException { + Message message = (Message) object; + super.convert(MessageBuilder.fromMessage(message).setHeader("serializer", "CUSTOM").build(), outputStream); + } + } + } diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/serializerJdbcMessageStore.xml b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/serializerJdbcMessageStore.xml new file mode 100644 index 0000000000..acbed98558 --- /dev/null +++ b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/serializerJdbcMessageStore.xml @@ -0,0 +1,14 @@ + + + + + + + + + + diff --git a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/util/SerializationUtilsTests.java b/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/util/SerializationUtilsTests.java deleted file mode 100644 index ab5872759f..0000000000 --- a/spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/util/SerializationUtilsTests.java +++ /dev/null @@ -1,70 +0,0 @@ -/* - * 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)); - } - -}