Defining a custome-converter within JdbcMessageStore

I am trying to use the xml namespace support for defining a jdbc message store.

However, the payloads of my messages flowing around are JaxB type, which fail throwing this exception when added to message store:

Code:

Caused by: org.springframework.core.serializer.support.SerializationFailedException: Failed to serialize object using DefaultSerializer; nested exception is java.io.NotSerializableException: com.xxx.commontypes.xxxx at org.springframework.core.serializer.support.SerializingConverter.convert(SerializingConverter.java:65) at org.springframework.integration.jdbc.JdbcMessageStore.addMessage(JdbcMessageStore.java:279)

I believe we should be able to specify a custom converter something like:

Code:

<int-jdbc:message-store id="quoteMessageStore" data-source="dataSource" message-converter="jaxBToStringConverter" />

Thoughts?

It should support reverse also, that is after deserializing the object retrieved from DB, convert it to expected format (JaxB in this case).

I resolved it this way.. Couldn't think of anything else...

1. Wrote custom messagestore extending JdbcMessageStore
2. Wrote custom serializer extending DefaultSerializer and deserializer extending DefaultDeserializer.

I would like to know if this can be done any simpler :

Custom messageStore

Code:

@SuppressWarnings({ "unchecked", "rawtypes" }) public QuoteMesssageStore(QuoteDefaultSerializer serializer, QuoteDefaultDeserializer deserializer, DataSource dataSource ) { super.setSerializer(serializer); super.setDeserializer((Deserializer) deserializer); super.setDataSource(dataSource); logger.info("Setting Constructor based seralizer / deserializer and dataSource"); }

Serializer:

Code:

public class QuoteDefaultSerializer extends DefaultSerializer { private Marshaller marshaller; public void setMarshaller(Marshaller marshaller) { this.marshaller = marshaller; } @Override public void serialize(Object object, OutputStream outputStream) throws IOException { Message<?> message = (Message<?>) object; FndtMsg fndtMsg = (FndtMsg) message.getPayload(); StringResult strResult = new StringResult(); marshaller.marshal(fndtMsg, strResult); Message<?> newMessage = MessageBuilder.withPayload(strResult.toString()).copyHeaders(message.getHeaders()).build(); super.serialize(newMessage, outputStream); } }

Deserializer:

Code:

public class QuoteDefaultDeserializer extends DefaultDeserializer { private Unmarshaller unmarshaller; public void setUnmarshaller(Unmarshaller unmarshaller) { this.unmarshaller = unmarshaller; } @Override public Object deserialize(InputStream inputStream) throws IOException { Message<?> message = (Message<?>) super.deserialize(inputStream); String fndtMessageString = (String) message.getPayload(); FndtMsg fndtMsg = (FndtMsg) unmarshaller.unmarshal(new StringSource(fndtMessageString)); return MessageBuilder.withPayload(fndtMsg).copyHeaders(message.getHeaders()).build(); } }

The JdbcMessageStore does already support the Serializer and Deserializer strategies. So, you should be able to avoid the custom subclass of the JdbcMessageStore and just provide references to your Serializer/Deserializer implementations.

Also, is there any particular reason you are (de)serializing the marshalled string? It's not clear what value this is adding.

If you just store the string itself, instead of subclassing the default implementations, you can just implement the (De)Serializer interfaces.

Thanks for your valuable comments Mark n Gary.

@Mark: You are correct, however, I don't want to support removeMessage( ) from MessageStore interface (and want to throw an unsupported exception), so eventually I have to replicate most of the things from JdbcMessageStore with some customization. (That means, no xml namespace support :S)

@Gary: I am sorry, but I am trying to understand your comments completely. I see the unneccessary serialization of String. So (correct me if I am wrong) what you are suggesting is to customize my jdbcMessageStore implementation something like:

Code:

.... public <T> Message<T> addMessage(final Message<T> message) { ........ ........ final String messageString = serializer.convert(result); ...... ...... jdbcTemplate.update(getQuery(CREATE_MESSAGE), new PreparedStatementSetter() { public void setValues(PreparedStatement ps) throws SQLException { logger.debug("Inserting message with id key=" + messageId); ps.setString(1, messageId); ps.setString(2, region); ps.setTimestamp(3, new Timestamp(createdDate)); lobHandler.getLobCreator().setClobAsString(ps, 4, messageString ); } });

Similarly, while deserializing:

Code:

private class MessageMapper implements RowMapper<Message<?>> { public Message<?> mapRow(ResultSet rs, int rowNum) throws SQLException { Message<?> message = (Message<?>) deserializer.convert(lobHandler.getClobAsString(rs, "MESSAGE_BYTES")); return message; } }

I will need to change the column type from BLOB to CLOB etc...

And, for Serializer / Deserializer, I implement as per needs... Is this what you were suggesting or I am thinking more than I should!

Thanks

Something like that, but I would leave it as a BLOB and use getBytes() on the string (and new String(bytes) during retrieval).

That way, if you later want to handle very large data more efficiently on the retrieval side you could use lobHandler.getBlobAsBinaryStream() and pass the resulting InputStream into your deserializer, wherein you can wrap it in a StreamSource for oxm unmarshalling.

It's a little trickier to provide such optimization on the output side because the Serializer writes to an OutputStream and the LobCreator needs an InputStream. You could use a PipedInputStream but you'd need to run the serializer on a separate thread. Not rocket science, but certainly a little trickier than the retrieval side.