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

INT-1202: add serializer strategy to JDBC message store

Browse Source
This commit is contained in:
David Syer
2010-09-01 15:11:58 +00:00
parent d71472bba1
commit b88644b7db
9 changed files with 144 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-integration-jdbc/pom.xml
View File

@@ -32,6 +32,10 @@
<groupId>org.springframework.integration</groupId>
<artifactId>spring-integration-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.commons</groupId>
<artifactId>spring-commons-serializer</artifactId>
</dependency>
<!-- test-scoped dependencies -->
<dependency>
<groupId>cglib</groupId>

40
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/JdbcMessageStore.java
View File

@@ -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<? super Message<?>> serializer) {
this.serializer = new SerializingConverter((OutputStreamingConverter<Object>) serializer);
}
/**
* A converter for deserializing byte arrays to messages.
*
* @param deserializer the deserializer to set
*/
@SuppressWarnings("unchecked")
public void setDeserializer(InputStreamingConverter<? super Message<?>> deserializer) {
this.deserializer = new DeserializingConverter((InputStreamingConverter<Object>) deserializer);
}
/**
* Check mandatory properties (data source and incrementer).
@@ -228,7 +260,7 @@ public class JdbcMessageStore extends AbstractMessageGroupStore implements Messa
Message<T> 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<Message<?>> {
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;
}

2
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParser.java
View File

@@ -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");

71
spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/util/SerializationUtils.java
View File

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

24
spring-integration-jdbc/src/main/resources/org/springframework/integration/jdbc/config/spring-integration-jdbc-2.0.xsd
View File

@@ -90,6 +90,30 @@
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="serializer" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Reference to an OutputStreamingConverter . Defaults to using java.io native serialization.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.commons.serializer.OutputStreamingConverter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
<xsd:attribute name="deserializer" type="xsd:string">
<xsd:annotation>
<xsd:documentation><![CDATA[
Reference to an InputStreamingConverter.
]]></xsd:documentation>
<xsd:appinfo>
<tool:annotation kind="ref">
<tool:expected-type type="org.springframework.commons.serializer.InputStreamingConverter" />
</tool:annotation>
</xsd:appinfo>
</xsd:annotation>
</xsd:attribute>
</xsd:extension>
</xsd:complexContent>
</xsd:complexType>

32
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/JdbcMessageStoreTests.java
View File

@@ -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<Message<?>>() {
public void convert(Message<?> object, OutputStream outputStream) throws IOException {
outputStream.write(object.getPayload().toString().getBytes());
outputStream.flush();
}
});
messageStore.setDeserializer(new InputStreamingConverter<Message<?>>() {
public Message<?> convert(InputStream inputStream) throws IOException {
BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
return new GenericMessage<String>(reader.readLine());
}
});
Message<String> message = MessageBuilder.withPayload("foo").build();
Message<String> 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 {

33
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/JdbcMessageStoreParserTests.java
View File

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

14
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/config/serializerJdbcMessageStore.xml Normal file
View File

@@ -0,0 +1,14 @@
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
xsi:schemaLocation="http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
http://www.springframework.org/schema/integration/jdbc http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc.xsd
http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd">
<jdbc:embedded-database id="dataSource" type="HSQL" />
<bean id="serializer" class="org.springframework.integration.jdbc.config.JdbcMessageStoreParserTests$EnhancedSerializer" />
<int-jdbc:message-store id="messageStore" data-source="dataSource" deserializer="serializer" serializer="serializer"/>
</beans>

70
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/util/SerializationUtilsTests.java
View File

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