WSO2 Enterprise Integrator is shipped with a separate message broker profile (WSO2 MB). In this Post I will be using message broker profile in EI (6.3.0).
1) Setting up the message broker profile
1.1) Copy the following JAR files from the <EI_HOME>/wso2/broker/client-lib/ directory to the <EI_HOME>/lib/ directory.
andes-client-3.2.13.jar
geronimo-jms_1.1_spec-1.1.0.wso2v1.jar
org.wso2.securevault-1.0.0-wso2v2.jar
1.2) Open the <EI_HOME>/conf/jndi.properties file and add the following line after the queue.MyQueue = example.MyQueue line:
queue.JMSMS=JMSMS
1.3) Open axis2.xml in <EI_HOME>/confconf/axis2/axis2.xml and uncomment configure to JMS transport support with WSO2 EI Broker Profile
There you will find transportReceiver and transportSender.
1.4) Add transport.jms.SessionTransacted value for each transportReceiver
<parameter name="transport.jms.SessionTransacted">true</parameter>
Eg:
<parameter name="myQueueConnectionFactory" locked="false">
<parameter name="java.naming.factory.initial" locked="false">org.wso2.andes.jndi.PropertiesFileInitialContextFactory</parameter>
<parameter name="java.naming.provider.url" locked="false">conf/jndi.properties</parameter>
<parameter name="transport.jms.ConnectionFactoryJNDIName" locked="false">QueueConnectionFactory</parameter>
<parameter name="transport.jms.ConnectionFactoryType" locked="false">queue</parameter>
<parameter name="transport.jms.SessionTransacted">true</parameter>
</parameter>
1.5) Start EI and Broker profiles
2) Implementation
Use case : There is API it will called WSO2 MB Queue and pass to the End point when EP is on Error it need to pass Message to dead letter channel.
2.1) First we will put message to Queue in WSO2 MB with API
<?xml version="1.0" encoding="UTF-8"?>
<api context="/dl-test" name="dl-test" xmlns="http://ws.apache.org/ns/synapse">
<resource methods="POST">
<inSequence>
<log level="custom">
<property name="property_name" value="DL-test API is called"/>
</log>
<property xmlns="http://ws.apache.org/ns/synapse" name="HEADER" value="VALUE" scope="transport" type="STRING"/>
<property description="xml" name="ContentType" scope="axis2" type="STRING" value="application/xml"/>
<property name="messageType" scope="axis2" type="STRING" value="application/xml"/>
<property name="OUT_ONLY" scope="default" type="STRING" value="true"/>
<property name="FORCE_SC_ACCEPTED" scope="axis2" type="STRING" value="true"/>
<send>
<endpoint>
<address uri="jms:/jmsms?transport.jms.ConnectionFactoryJNDIName=QueueConnectionFactory&java.naming.factory.initial=org.wso2.andes.jndi.PropertiesFileInitialContextFactory&java.naming.provider.url=conf/jndi.properties&transport.jms.DestinationType=queue"/>
</endpoint>
</send>
</inSequence>
<outSequence/>
<faultSequence>
<log level="custom">
<property name="property_name" value="faultSequence - test API is hitted"/>
</log>
</faultSequence>
</resource>
</api>
2.2) Then Create Inbound End point for this queue
<?xml version="1.0" encoding="UTF-8"?>
<inboundEndpoint name="abc-inbound-ep" onError="c2b-1-integration-v1-common-fault-sequence" protocol="jms" sequence="test-seq" suspend="false" xmlns="http://ws.apache.org/ns/synapse">
<parameters>
<parameter name="interval">1000</parameter>
<parameter name="sequential">true</parameter>
<parameter name="coordination">true</parameter>
<parameter name="transport.jms.Destination">jmsms</parameter>
<parameter name="transport.jms.CacheLevel">3</parameter>
<parameter name="transport.jms.ConnectionFactoryJNDIName">QueueConnectionFactory</parameter>
<parameter name="java.naming.factory.initial">org.wso2.andes.jndi.PropertiesFileInitialContextFactory</parameter>
<parameter name="java.naming.provider.url">conf/jndi.properties</parameter>
<parameter name="transport.jms.SessionAcknowledgement">AUTO_ACKNOWLEDGE</parameter>
<parameter name="transport.jms.SessionTransacted">true</parameter>
<parameter name="transport.jms.SubscriptionDurable">false</parameter>
<parameter name="transport.jms.ConnectionFactoryType">queue</parameter>
<parameter name="transport.jms.SharedSubscription">false</parameter>
</parameters>
</inboundEndpoint>
2.3) Here is sample Sequence to used in Inbound.
<?xml version="1.0" encoding="UTF-8"?>
<sequence name="test-seq" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
<log level="custom">
<property name="property_name" value="test-seq ABC Queue is called"/>
<property expression="$trp:HEADER" name="property_name"/>
</log>
<log level="full"/>
</sequence>
2.4) Then Improve Sequence to send value (XML) to End point.
If the Endpoint have issue message should retry and move to the dead letter channel.
The Dead Letter Channel (DLC) is a sub-set of a queue, specifically designed to persist messages that are typically marked for deletion, providing you with a choice on whether to delete, retrieve or reroute the messages from the DLC.
<?xml version="1.0" encoding="UTF-8"?>
<sequence name="test-seq" onError="test-fault-sequence" trace="disable" xmlns="http://ws.apache.org/ns/synapse">
<log level="custom">
<property name="property_name" value="test-seq ABC Queue is hitted"/>
<property expression="$trp:HEADER" name="property_name"/>
</log>
<log level="full"/>
<log level="custom">
<property name="Message" value="test-proxy is hitted"/>
</log>
<property name="OUT_ONLY" scope="default" type="STRING" value="true"/>
<call blocking="true">
<endpoint key="dl-test-ep"/>
</call>
</sequence>
2.4.2) Adding new sequence for on Error that make message to DLC.
<?xml version="1.0" encoding="UTF-8"?>
<sequence name="test-fault-sequence" trace="disable"
xmlns="http://ws.apache.org/ns/synapse">
<log level="full">
<property name="MESSAGE" value="Executing default "fault" sequence" />
<property expression="get-property('ERROR_CODE')" name="ERROR_CODE" />
<property expression="get-property('ERROR_MESSAGE')" name="ERROR_MESSAGE" />
</log>
<property name="SET_ROLLBACK_ONLY" scope="axis2" type="STRING"
value="true" />
<log level="custom">
<property name="Transaction Action" value="Rollbacked" />
</log>
</sequence>
OR you can have proxy listen for the Queue (Inbound is much better)
<?xml version="1.0" encoding="UTF-8"?>
<proxy name="test-proxy" startOnLoad="true" transports="http https jms" xmlns="http://ws.apache.org/ns/synapse">
<target>
<inSequence>
<log level="custom">
<property value="test-proxy is hitted" name="Message"/>
</log>
<property name="OUT_ONLY" scope="default" type="STRING" value="true"/>
<call blocking="true">
<endpoint key="dl-test-ep"/>
</call>
</inSequence>
<outSequence/>
<faultSequence>
<log level="full">
<property name="MESSAGE" value="Executing default "fault" sequence"/>
<property expression="get-property('ERROR_CODE')" name="ERROR_CODE"/>
<property expression="get-property('ERROR_MESSAGE')" name="ERROR_MESSAGE"/>
</log>
<property name="SET_ROLLBACK_ONLY" scope="axis2" type="STRING" value="true"/>
<log level="custom">
<property name="Transaction Action" value="Rollbacked"/>
</log>
</faultSequence>
</target>
<parameter name="transport.jms.ContentType">
<rules>
<jmsProperty>contentType</jmsProperty>
<default>application/xml</default>
</rules>
</parameter>
</proxy>
3) Test the use case
Do small change for end point for testing
<?xml version="1.0" encoding="UTF-8"?>
<endpoint name="dl-test-ep" xmlns="http://ws.apache.org/ns/synapse">
<http method="get" uri-template="http://bacde.com/175"/>
</endpoint>
Wrong URI
Then you will see the retrying it for 10 times.
If you go the MB Dead Letter Channel, you will find the message.
Home -> Manage -> Dead Letter Channel -> List
You can restore your message or delete any message or reroute.
Once you restore you will see it worked (after fixing EP URL)
View comments