How to enable RedeliveryDelay for ActiveMQ in Mule XA transactions -


i have simple flow in mule, listens activemq queue , if message queue contains string "fail" exception thrown. without xa transaction, works ok. message redelivered after 5 seconds, next time after seconds more. after 5 redeliveries message moved dead letter queue.

with xa transactions message redelivered 5 times , moved dead letter queue. how can tell mule or activemq wait redeliverydelay?

<?xml version="1.0" encoding="utf-8"?>  <mule xmlns:scripting="http://www.mulesoft.org/schema/mule/scripting"     xmlns:jbossts="http://www.mulesoft.org/schema/mule/jbossts" xmlns:jms="http://www.mulesoft.org/schema/mule/jms"     xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation"     xmlns:spring="http://www.springframework.org/schema/beans" version="ce-3.6.1"     xmlns:xsi="http://www.w3.org/2001/xmlschema-instance"     xsi:schemalocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-current.xsd http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd http://www.mulesoft.org/schema/mule/jms http://www.mulesoft.org/schema/mule/jms/current/mule-jms.xsd http://www.mulesoft.org/schema/mule/jbossts http://www.mulesoft.org/schema/mule/jbossts/current/mule-jbossts.xsd http://www.mulesoft.org/schema/mule/scripting http://www.mulesoft.org/schema/mule/scripting/current/mule-scripting.xsd">      <spring:beans>          <!-- redelivery policy -->         <spring:bean id="redeliverypolicy" class="org.apache.activemq.redeliverypolicy">             <spring:property name="maximumredeliveries" value="5" />             <spring:property name="initialredeliverydelay" value="5000" />             <spring:property name="redeliverydelay" value="5000" />             <spring:property name="useexponentialbackoff" value="true" />             <spring:property name="backoffmultiplier" value="1.5" />         </spring:bean>          <!-- activemq connection factory -->         <spring:bean id="amqxafactory"             class="org.apache.activemq.activemqxaconnectionfactory" lazy-init="true"             name="amqxafactory">             <spring:property name="brokerurl"                 value="tcp://localhost:61616" />             <spring:property name="redeliverypolicy" ref="redeliverypolicy" />         </spring:bean>      </spring:beans>      <jbossts:transaction-manager doc:name="jboss transaction manager">         <property key="com.arjuna.ats.arjuna.coordinator.defaulttimeout"             value="47" /><!-- in seconds -->         <property key="com.arjuna.ats.arjuna.coordinator.txreapertimeout"             value="108000" /><!-- in milliseconds -->     </jbossts:transaction-manager>      <jms:activemq-connector name="active_mqconnectormessages"         specification="1.1" validateconnections="true" doc:name="active mq xa"         persistentdelivery="true" connectionfactory-ref="amqxafactory"         numberofconsumers="1" maxredelivery="5">         <reconnect-forever frequency="10000" />     </jms:activemq-connector>      <flow name="xatestflow">         <jms:inbound-endpoint queue="xa.test" doc:name="jms"             connector-ref="active_mqconnectormessages">             <xa-transaction action="always_begin" />         </jms:inbound-endpoint>         <logger message="message queue: #[payload]" level="info"             doc:name="logger" />         <scripting:component doc:name="fail if payload contains fail">             <scripting:script engine="groovy"><![cdata[if (payload.contains("fail")) {   throw new illegalargumentexception("failed...") } ]]></scripting:script>         </scripting:component>         <logger message="message ready" level="info" doc:name="logger" />     </flow> </mule>

it took time, discovered solution. there 2 types of configuration, server side in activemq , client side.

in client redeliverypolicy changed maximumredeliveries 0. on server created redeliveryplugin in activemq.xml:

    <broker ....>     ......     <plugins>         <redeliveryplugin fallbacktodeadletter="true" sendtodlqifmaxretriesexceeded="true">             <redeliverypolicymap>                 <redeliverypolicymap>                     <redeliverypolicyentries>                         <!-- destination specific policy -->                         <redeliverypolicy                              queue="amm.input.to.router" maximumredeliveries="3"                              redeliverydelay="7000"                             initialredeliverydelay="5000"                             usecollisionavoidance="true"                             />                     </redeliverypolicyentries>                      <!-- fallback policy other destinations -->                     <defaultentry>                         <redeliverypolicy                             maximumredeliveries="4"                             useexponentialbackoff="true"                             initialredeliverydelay="10000"                             redeliverydelay="5000"                             usecollisionavoidance="true"                             backoffmultiplier="1.5"                             maximumredeliverydelay="93600000" />                     </defaultentry>                 </redeliverypolicymap>             </redeliverypolicymap>         </redeliveryplugin>     </plugins>      </broker>

now activemq server takes care of redeliveries, not client.

see http://activemq.apache.org/message-redelivery-and-dlq-handling.html


Comments

Post a Comment

Popular posts from this blog

searchKeyword not working in AngularJS filter -

i writing app using angularjs on front end. want search through table word/field; 1 search box everything. according article here: http://hello-angularjs.appspot.com/searchtable can use filter: searchkeyword this. this code, , not working expected. <label>search: <input ng-model="searchkeyword"></label> <table ng-repeat="post in posts | orderby: sort | filter: searchkeyword"> <tr> <td> {{index(post)}} </td> <td> {{post.title}} </td> <td> {{post.author}} </td> <td> {{post.content}} </td> <td> {{post.date | date: "d/m/yyyy"}} </td> </tr> </table> what should do? thank you try this: controller $scope.posts = [ {title:'title one', author:'author one', content: 'content one'}, {title:'title two', author:'author two', content: 'cont...
Read more

sequelize.js - Sequelize: sort by enum cases -

i have model ( article ) field (the name type ) of type enum('source', 'translated') , return 1 article ordered type field. source articles should returned before translated articles. like: article.findone(order: [ { type: [ 'source', 'translated' ] } ]); unfortunately there appears no built-in solution sorting enums. instead have sort by case , couldn't find documentation on how sequelize. what conventional approach here? sorting case hard in sequelize, sort field (the second answer in linked post) should possible: order: sequelize.fn('field', sequelize.col('type'), 'source', 'translated') it's supported mysql far know though
Read more

user interface - how to replace an ongoing process of image capture from another process call over the same ImageLabel in python's GUI TKinter -

i trying use following code create interactive gui changes color-space(hsv, rbg, grayscale etc) on event of button click. display opencv video in tkinter using multiprocessing being new python having problems multiprocessing , attempts on making gui change it's color space on button clicks hangs entire system. on it's implementation highly appreciated. thankyou. below code: import cv2 pil import image,imagetk import tkinter tk import numpy np multiprocessing import process , queue def quit_it(root,process): root.destroy() process.terminate() def black_andwhite(root,process): process.terminate p=process(target=capture_image, args=(5,queue, )) p.start() root.after(0, func=lambda: update_all(root, image_label, queue)) def update_image(image_label, queue): frame = queue.get() = image.fromarray(frame) b = imagetk.photoimage(image=a) image_label.configure(image=b) image_label._image_cache = b root.update() def update_all(root...
Read more