1

I am using the default messaging subsystem in Jboss EAP 6.1 I have implemented transactions in my component. In Address setting configuration i have specified the below configuration. Address setting configuration:

 <address-setting match="jms.queue.Src.in">
                        <dead-letter-address>jms.queue.DLQ</dead-letter-address>
                        <expiry-address>jms.queue.ExpiryQueue</expiry-address>
                        <redelivery-delay>5000</redelivery-delay>
                        <max-delivery-attempts>4</max-delivery-attempts>
 </address-setting>

I have specified global transaction in switchyard component,when ever an exception occurs the transaction is rollbacked and after retry the message is placed in dead letter queue.along with that i need to frame an error message and place it in Error Queue for the consumer to pick it.The message is going to error queue but the transaction is not getting commited since there was an exception.

Switchyard XML:

<?xml version="1.0" encoding="UTF-8"?>
<sy:switchyard xmlns:camel="urn:switchyard-component-camel:config:1.1" xmlns:jms="urn:switchyard-component-camel-jms:config:1.1" xmlns:sca="http://docs.oasis-open.org/ns/opencsa/sca/200912" xmlns:sy="urn:switchyard-config:switchyard:1.1" name="JmsToSca" targetNamespace="urn:com.company.esb.prototypes.consumer.async:JmsToSca:1.0">
  <sca:composite name="JmsToSca" targetNamespace="urn:com.company.esb.prototypes.consumer.async:JmsToSca:1.0">
    <sca:component name="RequestJmsToSca">
      <camel:implementation.camel requires="managedTransaction.Global">
        <camel:xml path="route/camelroutesend.xml"/>
      </camel:implementation.camel>
      <sca:service name="RequestInService" requires="propagatesTransaction">
        <sca:interface.java interface="com.company.esb.common.services.StringMessageService"/>
      </sca:service>
      <sca:reference name="RequestOutService" requires="propagatesTransaction">
        <sca:interface.java interface="com.company.esb.common.services.esbMessageService"/>
      </sca:reference>
      <sca:reference name="ResponseInService" requires="propagatesTransaction">
        <sca:interface.java interface="com.company.esb.common.services.esbMessageService"/>
      </sca:reference>
      <sca:reference name="AuditLoggerService" requires="suspendsTransaction">
        <sca:interface.java interface="com.company.esb.common.services.AuditLoggerService"/>
      </sca:reference>
      <sca:reference name="ExceptionHandlerService" requires="propagatesTransaction">
        <sca:interface.java interface="com.company.esb.common.services.ExceptionHandlerService"/>
      </sca:reference>
    </sca:component>
    <sca:service name="RequestInService" promote="RequestJmsToSca/RequestInService">
      <sca:interface.java interface="com.company.esb.common.services.StringMessageService"/>
      <jms:binding.jms name="jms1">
        <jms:contextMapper includes="JMSMessageID.*,JMSXDeliveryCount.*"/>
        <jms:queue>${Consumer.Async.JmsToSca.RequestQueueName}</jms:queue>
        <jms:connectionFactory>${Consumer.Async.JmsToSca.ConnectionFactoryname}</jms:connectionFactory>
        <jms:disableReplyTo>true</jms:disableReplyTo>
        <jms:transacted>true</jms:transacted>
        <jms:transactionManager>#jtaTransactionManager</jms:transactionManager>
      </jms:binding.jms>
    </sca:service>
    <sca:reference name="RequestOutService" multiplicity="0..1" promote="RequestJmsToSca/RequestOutService">
      <sca:interface.java interface="com.company.esb.common.services.esbMessageService"/>
      <sca:binding.sca sy:target="RequestInService" sy:targetNamespace="urn:com.company.esb.prototypes.consumer.async:AsyncScaToJms:1.0" name="sca1"/>
    </sca:reference>
    <sca:reference name="AuditLoggerService" multiplicity="0..1" promote="RequestJmsToSca/AuditLoggerService">
      <sca:interface.java interface="com.company.esb.common.services.AuditLoggerService"/>
      <sca:binding.sca sy:target="AuditLoggerService" sy:targetNamespace="urn:com.company.esb.components:AuditLogger:1.0" name="ScaAudit"/>
    </sca:reference>
    <sca:reference name="ExceptionHandlerService" multiplicity="0..1" promote="RequestJmsToSca/ExceptionHandlerService">
      <sca:interface.java interface="com.company.esb.common.services.ExceptionHandlerService"/>
      <sca:binding.sca sy:target="ExceptionHandlerService" sy:targetNamespace="urn:com.company.esb.components:ExceptionHandler:1.0" name="ScaExceptionHandler"/>
    </sca:reference>
    <sca:component name="ResponseScaToJms">
      <camel:implementation.camel requires="managedTransaction.Global">
        <camel:xml path="route/camelroutereceive.xml"/>
      </camel:implementation.camel>
      <sca:service name="ResponseInService" requires="propagatesTransaction">
        <sca:interface.java interface="com.company.esb.common.services.esbMessageService"/>
      </sca:service>
      <sca:reference name="AuditLoggerService" requires="suspendsTransaction">
        <sca:interface.java interface="com.company.esb.common.services.AuditLoggerService"/>
      </sca:reference>
      <sca:reference name="ExceptionHandlerService" requires="propagatesTransaction">
        <sca:interface.java interface="com.company.esb.common.services.ExceptionHandlerService"/>
      </sca:reference>
      <sca:reference name="ResponseOutService" requires="propagatesTransaction">
        <sca:interface.java interface="com.company.esb.common.services.StringMessageService"/>
      </sca:reference>
    </sca:component>
    <sca:service name="ResponseInService" promote="ResponseScaToJms/ResponseInService">
      <sca:interface.java interface="com.company.esb.common.services.esbMessageService"/>
      <sca:binding.sca name="sca1"/>
    </sca:service>
    <sca:reference name="ResponseOutService" multiplicity="0..1" promote="ResponseScaToJms/ResponseOutService">
      <sca:interface.java interface="com.company.esb.common.services.StringMessageService"/>
      <jms:binding.jms name="jms1">
        <jms:contextMapper excludes="JMSXDeliveryCount.*" includes="JMSCorrelationID.*"/>
        <jms:queue>${Consumer.Async.JmsToSca.ResponseQueueName}</jms:queue>
        <jms:connectionFactory>${Consumer.Async.JmsToSca.ConnectionFactoryname}</jms:connectionFactory>
        <jms:transacted>false</jms:transacted>
      </jms:binding.jms>
    </sca:reference>
  </sca:composite>
  <sy:domain>
    <sy:properties>
      <sy:property name="org.switchyard.handlers.messageTrace.enabled" value="true"/>
      <sy:property name="org.switchyard.propagateExceptionOnInOnly" value="true"/>
    </sy:properties>
  </sy:domain>
</sy:switchyard>

Camel Route(camelroutesend.xml):

<?xml version="1.0" encoding="ASCII"?>
<routes xmlns="http://camel.apache.org/schema/spring">
    <route id="ConsumerAsyncJmsToScaRoute" streamCache="true">
        <from uri="switchyard://RequestInService" />
        <transacted ref="transactionPolicy" />
        <doTry>
            <bean ref="MessageGenerator" method="createesbMessage()" />
            <removeHeader headerName="JMSMessageID" />
            <bean ref="esbHelper" method="prepareAuditCall()" />
            <wireTap uri="switchyard://AuditLoggerService" />
            <setProperty propertyName="OriginalesbMessage">
                <simple>${body}</simple>
            </setProperty>

            <!-- Step3- invokes next pattern -->
            <to uri="switchyard://RequestOutService" />
            <doCatch>
                <!-- Step7- captures error in the route and invokes esbErrorHandler -->
                <exception>java.lang.Exception</exception>
                <handled>
                    <constant>true</constant>
                </handled>
                <log message="called Exception block ConsumerAsyncJmsToScaRoute ${body}"/>
                <bean ref="esbHelper" method="prepareErrorHandlerCall(false)" />
                <to uri="switchyard://ExceptionHandlerService" />
                <to uri="switchyard://ResponseInService" />
                <!-- Step8- Prepares response message in case of error -->
            </doCatch>
        </doTry>
    </route>
</routes>

Camel Route(camelroutereceive.xml):

<?xml version="1.0" encoding="ASCII"?>
<routes xmlns="http://camel.apache.org/schema/spring">
    <route id="ConsumerAsyncScaToJmsRoute" streamCache="true">
        <from uri="switchyard://ResponseInService" />
        <transacted ref="transactionPolicy" />
        <doTry>
            <bean ref="esbHelper" method="prepareAuditCall()" />
            <wireTap uri="switchyard://AuditLoggerService" />
            <setProperty propertyName="OriginalesbMessage">
                <simple>${body}</simple>
            </setProperty>

            <to uri="xslt:xslt/create-response.xslt" />

            <bean ref="MessageGenerator" method="setCorrelationID()" />
            <!-- Step3- invokes Provider service -->
            <!--<to uri="switchyard://ResponseOutService" /> -->
            <to
                uri="jms:queue:{{Consumer.Async.JmsToSca.ResponseQueueName}}?connectionFactory={{Consumer.Async.JmsToSca.ConnectionFactoryname}}&amp;transacted=true"
                pattern="InOnly" />

            <doCatch>
                <!-- Step7- captures error in the route and invokes esbErrorHandler -->
                <exception>java.lang.Exception</exception>
                <handled>
                    <constant>true</constant>
                </handled>
                <bean ref="esbHelper" method="prepareErrorHandlerCall(false)" />
                <to uri="switchyard://ExceptionHandlerService" />
                <!-- Step8- Prepares response message in case of error -->
            </doCatch>
        </doTry>
    </route>
</routes>
Ravi
  • 1,247
  • 4
  • 15
  • 35

0 Answers0