Persistent channel?

[huima]

Hi,

we are currently looking into Spring integration and how to use it as a base for messaging subsystem in our project.One thing that we would like to implement or see implemented with SI is some kind of persistent channel that makes sure that messages are lot lost if system crashes before message is routed to the endpoint that persists it.

Has anyone yet implemented a persistent channel that we could use, or does anyone have any pointers where to look at first?

Someone linked this: http://www.gaborcselle.com/writings/...ent_queue.html previously here, but I haven't looked at it at all yet.

[Mark Fisher]

The easiest way to provide persistence is to use Spring Integration's JMS channel-adapters in place of any channel where persistence is necessary.

See the reference manual section here:
http://static.springframework.org/sp...rence.html#jms

That way, you can rely on your choice of JMS provider to manage the persistence. We have considered adding a single "jms:channel" element that would essentially combine the inbound and outbound channel adapters.

I would be interested to know if the JMS channel adapter approach suits your use case.

[rganger]

Hi,

If you don't mind me entering the discussion, I have a few thoughts which might be useful to this thread.

There is a related thread here:
http://forum.springsource.org/showthread.php?t=64295

I have recently been working with spring integration using JBoss as the jms provider. My use case requires message persistence on a jms topic between a publisher and multiple subscribers. I need to guarantee delivery of the messages to the subscribers while the publisher app is up and running.

JBoss (on a non-durable jms topic) seems to dump all messages when the last publisher/subscriber disconnects. Because the spring JmsTemplate does not keep its connection open between polls, there are times when (even though the apps are still running) the jms topic has no open connections, and undelivered messages disappear.

A potential answer to the problem is to use durable topics (I don't know if this might bring some other negative issues, such as how to keep the topic cleaned out). Another potential answer would be the ability to create a persistent connection to the jms topic through spring that would stay connected all the time. My current solution to the problem is to run a non-spring client that opens a persistent jms connection to the topic, and does not consume any of the messages.

Does the spring team have any thoughts on the best way to handle this issue?

Thanks,
Bob

[Mark Fisher]

A couple things to consider...

1) if you use Spring Integration's message-driven-channel-adapter, it will be backed by a MessageListener container rather than polling with JmsTemplate each time. This also typically provides better response time (unless you are already polling with a negligible interval and long timeout).

2) Spring provides some ConnectionFactory options: SingleConnectionFactory and CachingConnectionFactory. Their respective JavaDocs provide a lot of detail.

Please post back here if you get a chance to try out these other options.

[qmx]

Only to enrich Mark's comment, I was facing the same problem, that got solved with SingleConnectionFactory.

[rganger]

Thanks for the great information...I'm in the middle of a crunch right now, but when I get a chance I'll try out the different options and report back how they help my use case.

[huima]

That way, you can rely on your choice of JMS provider to manage the persistence. We have considered adding a single "jms:channel" element that would essentially combine the inbound and outbound channel adapters.

I would be interested to know if the JMS channel adapter approach suits your use case.

We'll have to take a look at it. We're using Caucho Resin as the application server, and it's JMS options look promising. We are doing - for now - something really simple, so using JMS feels and sounds like an overkill and adding too much complexity - but it might be just a unwarranted feeling.

Let's see.

[mpilone]

I'm just getting up to speed with Spring Integration, but back to the original question, how hard would it be to implement persistence without JMS?

I currently use JMS now for reliable messaging but it is all within a single application server. I've had a number of problems with the configuration (mostly related to requiring distributed transactions) and I was looking at Spring Integration as a solution. I don't need guaranteed deliver (no retries) and everything is local. However I do need persistence for server shutdown (or crash).

I was originally thinking that this could be done with some kind of datasource backed channel or a channel adapter. The idea would be to use the same data source as the service sending the messages so the TX of the service controls the commit/rollback of the messages to the channel.

Thoughts?

-mike

[rganger]

Hello,

Following up on my earlier post, and based on Mark Fisher's suggestions, I worked up examples of a topic publisher and subscriber that provides for guaranteed delivery of messages from the publisher to the subscriber until the subscriber disconnects. This example works on JBoss Messaging (1.4.0.SP3 on JBoss 4.2.2.GA), and does not require a durable topic. It avoids lost messages that may occur if the publisher disconnects from the JMS server before the subscriber has reconnected.

The JNDI-referenced connection factory is passed to either a Spring SingleConnectionFactory or CachingConnectionFactory bean which in turn is used in the channel adapters. These beans do not disconnect the topic session between sends or polls. I don't know if there is any advantage between the two types of factories.

Additionally, the message-driven-channel-adapter provides for listening on the topic without having to specify a poll interval. This is convenient, but the inbound-channel-adapter with a poll interval also works OK as long as one of the above connection factories are used.

Below are examples of the spring beans:

Publisher:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:int="http://www.springframework.org/schema/integration"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:beans="http://www.springframework.org/schema/beans"
	xmlns:file="http://www.springframework.org/schema/integration/file"
	xmlns:jms="http://www.springframework.org/schema/integration/jms"
	xmlns:stream="http://www.springframework.org/schema/integration/stream"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
		org/springframework/beans/factory/xml/spring-beans-2.5.xsd
		http://www.springframework.org/schema/integration
		org/springframework/integration/config/xml/spring-integration-1.0.xsd
		http://www.springframework.org/schema/integration/jms
		org/springframework/integration/jms/config/spring-integration-jms-1.0.xsd
		http://www.springframework.org/schema/integration/file
		org/springframework/integration/file/config/spring-integration-file-1.0.xsd
		http://www.springframework.org/schema/integration/stream
		org/springframework/integration/stream/config/spring-integration-stream-1.0.xsd">
	
    <bean id="localConnectionFactory" class="org.springframework.jndi.JndiObjectFactoryBean">
        <property name="jndiName" value="java:/ConnectionFactory"/>
    </bean>

    <bean id="localTestDestination" class="org.springframework.jndi.JndiObjectFactoryBean">
        <property name="jndiName" value="/topic/testTopic2"/>
    </bean>

	<int:channel id="testChannel"/>

	<jms:outbound-channel-adapter
		id="testOutboundAdapter"
		connection-factory="localConnectionFactory" 
		destination="localTestDestination" 	
		channel="testChannel"/>

	
</beans>

Subscriber:

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:int="http://www.springframework.org/schema/integration"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xmlns:beans="http://www.springframework.org/schema/beans"
	xmlns:file="http://www.springframework.org/schema/integration/file"
	xmlns:jms="http://www.springframework.org/schema/integration/jms"
	xmlns:stream="http://www.springframework.org/schema/integration/stream"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
		org/springframework/beans/factory/xml/spring-beans-2.5.xsd
		http://www.springframework.org/schema/integration
		org/springframework/integration/config/xml/spring-integration-1.0.xsd
		http://www.springframework.org/schema/integration/jms
		org/springframework/integration/jms/config/spring-integration-jms-1.0.xsd
		http://www.springframework.org/schema/integration/file
		org/springframework/integration/file/config/spring-integration-file-1.0.xsd
		http://www.springframework.org/schema/integration/stream
		org/springframework/integration/stream/config/spring-integration-stream-1.0.xsd">
	
	<bean class="org.springframework.beans.factory.config.PropertyPlaceholderConfigurer"/>

    <bean id="localConnectionFactory" class="org.springframework.jndi.JndiObjectFactoryBean">
        <property name="jndiName" value="java:/ConnectionFactory"/>
    </bean>

<!--	<bean id="persistentConnectionFactory" class="org.springframework.jms.connection.SingleConnectionFactory">-->
<!--		<property name="targetConnectionFactory" ref="localConnectionFactory"/>-->
<!--	</bean>-->

	<bean id="persistentConnectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
		<property name="targetConnectionFactory" ref="localConnectionFactory"/>
	</bean>

    <bean id="localTestDestination" class="org.springframework.jndi.JndiObjectFactoryBean">
        <property name="jndiName" value="/topic/testTopic2"/>
    </bean>

	<int:publish-subscribe-channel id="testChannel"/>

<!--	<jms:inbound-channel-adapter-->
<!--		id="testInboundAdapter"-->
<!--		connection-factory="persistentConnectionFactory" -->
<!--		destination="localTestDestination" 	-->
<!--		channel="testChannel">-->
<!--		<int:poller>-->
<!--			<int:interval-trigger interval="500"/>-->
<!--		</int:poller>-->
<!--	</jms:inbound-channel-adapter>-->

	<jms:message-driven-channel-adapter
		id="testInboundAdapter"
		connection-factory="persistentConnectionFactory" 
		destination="localTestDestination" 	
		channel="testChannel"/>

	<bean id="testConsumer" class="org.springframework.integration.endpoint.EventDrivenConsumer">
		<constructor-arg ref="testChannel"/>
		<constructor-arg ref="testMessageHandler"/>
	</bean>

	<bean id="testMessageHandler" class="org.cubrc.examples.jms.TestMessageHandler"/>
	
</beans>

Thanks for the many good suggestions,

Bob

[Mark Fisher]

Thank you Bob. I would be interested if this experience has triggered any ideas for anything that could be simplified in Spring Integration (e.g. we have considered simply adding 'jms:channel' and 'jms: publish-subscribe-channel'). (intentional space to avoid a smiley)

Regards,
Mark