Using Control bus to stop message-driven-channel-adapter that uses transactional session

Question

My requirement is to use transactional session with message-driven-channel-adapter (JmsMessageDrivenEndpoint). I am able to setup the configuration buy using sessionTransacted = true for DefaultMessageListenerContainer.

Work flow: receive a message -> call the service activator -> service activator calls dao class

On successful commit to database a commit() is called by spring framework and on any runtime exception a rollback() is called by spring framework. Which works just fine. When a rollback happens JMS Broker sends the message back again to my application.

For a specific type of exception in dao I want to add a message header (i.e redelivery time) so that JMS Broker will not send the message again right away. How can I do it?

For another specific type of exception in dao I want to use control bus to stop the end point (message-driven-channel-adapter) and before stopping it rollback the previous transaction. How can I do it?

Any one can help me out?

Answer 1

There is no wonder, how to use Control Bus for start/stop endpoints:

<int:control-bus input-channel="controlChannel"/>

<int-jms:message-driven-channel-adapter id="jmsInboundEndpoint"/>

<int:transformer input-channel="stopImsInboundEndpointChannel"
           outbound-channel="controlChannel"
           expression="'@jmsInboundEndpoint.stop()'"/>

Or you can send to the controlChannel the same command string from any place of your code.

But it doesn't matter that the last transaction will be rollbacked. It depends on your 'unit of work' (in other words - the behaviour of your service).

However you can, at the same time when you send 'stop command', mark current transaction for rollback:

TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();

Another your question about 'adding some message header' is abnormal for Messaging at all. If you change the message it will be a new one and you can't rollback message to the queue with some new info.

Of course, you can do it anyway and have new message. But you should resend it, not rollback. So, you should commit transaction anyway and send that new message somewhere (or to the same queue), but it will be new message as for Broker as well for your application. And one more time: for this case you have to commit transaction.

Not sure that it is very clear and I go right way in my asnwer, but hope it helps you a bit.

Answer 2

You cannot modify the message (add a header) before rollback. You could, of course, requeue it as a new message after catching the exception. Some brokers (e.g. ActiveMQ) provide a back-off retry policy after a rollback. That might be a better solution if your broker supports it.

You can use the control bus to stop the container, but you will probably have to do it asynchronously (invoke the stop on another thread, e.g. by using an ExecutorChannel on the control bus). Otherwise, depending on your environment you might have problems with the stop waiting for the container thread to exit, so you shouldn't execute the stop on the container thread itself.

Best thing to do is experiment.

Answer 3

Thanks Gary and Artem. The solution is working. I am using the below configuration:

<jms:message-driven-channel-adapter id="jmsMessageDrivenChannelAdapter" connection-factory="connectionFactory"
        destination="destination" transaction-manager="jmsTransactionManager" channel="serviceChannel" error-channel="ultimateErrorChannel" />
<si:service-activator input-channel="ultimateErrorChannel" output-channel="controlChannel">
    <bean class="play.spring.integration.TestErrorHandler">
        <property name="adapterNeedToStop" value="jmsMessageDrivenChannelAdapter" />
        <property name="exceptionWhenNeedToStop" value="play.spring.integration.ShutdownException" />
    </bean>
</si:service-activator>
<si:channel id="controlChannel">
    <si:dispatcher task-executor="controlBusExecutor" />
</si:channel>
<task:executor id='controlBusExecutor' pool-size='10' queue-capacity='50' />
<si:control-bus input-channel="controlChannel" />

Now my question is if I want to stop multiple inbound adapters how can I send a single message to control-bus for all these adapters?

I am going to study SpEL. Would appreciate if someone already know it.

Thanks