WSO2 ESB - Sample Class Mediator

In WSO2 ESB / EI when we want to write our own logic of mediation, class mediators are very helpful.

A class mediator can be written by extending AbstractMediator

Following is a sample class mediator which simply logs a sting message (Doesn't actually do any mediation)

This source can be used as a template to start a class mediator quickly

public class SampleClassMediator extends AbstractMediator {

    private static final Log log = LogFactory.getLog(SampleClassMediator.class);

    public boolean mediate(MessageContext messageContext) {
        log.info("This is a sample class mediator");
        return true;
    }
}

Full project can be found in  https://github.com/bsenduran/sample-class-mediator/tree/v1.0

Setting up WSO2 EI Cluster

This post explains on setting up a WSO2 Enterprise Integrator (WSO2 EI)'s 2 node non worker manager cluster in 5 simple steps.

1. Setting up database

2. Configuring datasource, reqistry and user-mgt

3. Enabling clusting in axis2.xml

4. Configuring Node 2

5. Configure a common deployment directory

I will be using WSO2 EI 6.1.1 and MySQL 5.7 as the database.

1. Setting up database

Creating the database and tables

create database WSO2_EI_611_REG_DB;
use WSO2_EI_611_REG_DB;
source /path-to-<EI-HOME>/dbscripts/mysql5.7.sql;

create database WSO2_EI_611_UM_DB;
use WSO2_EI_611_UM_DB;
source /path-to-<EI-HOME>/dbscripts/mysql5.7.sql;

Adding the MySQL driver

Add the MySQL driver jar  to <EI-HOME>/lib directory

2. Configuring datasource, reqistry and user-mgt

master-datasources.xml

Update the <EI-HOME>/conf/datasources/master-datasources.xml with following data source configuration (You may need to modify the username / password / URL

    <datasource>
        <name>WSO2_SHARED_REG_DB</name>
        <description>The datasource is used for shared config and governance registry</description>
        <jndiConfig>
            <name>jdbc/WSO2SharedDB</name>
        </jndiConfig>
        <definition type="RDBMS">
            <configuration>
                <url>jdbc:mysql://localhost:3306/WSO2_EI_611_REG_DB</url>
                <username>user</username>
                <password>password</password>
                <driverClassName>com.mysql.jdbc.Driver</driverClassName>
                <maxActive>50</maxActive>
                <maxWait>60000</maxWait>
                <testOnBorrow>true</testOnBorrow>
                <validationQuery>SELECT 1</validationQuery>
                <validationInterval>30000</validationInterval>
            </configuration>
        </definition>
    </datasource>

    <datasource>
        <name>WSO2_UM_DB</name>
        <description>The datasource is used for user management</description>
        <jndiConfig>
            <name>jdbc/WSO2UmDB</name>
        </jndiConfig>
        <definition type="RDBMS">
            <configuration>
                <url>jdbc:mysql://localhost:3306/WSO2_EI_611_UM_DB</url>
                <username>user</username>
                <password>password</password>
                <driverClassName>com.mysql.jdbc.Driver</driverClassName>
                <maxActive>50</maxActive>
                <maxWait>60000</maxWait>
                <testOnBorrow>true</testOnBorrow>
                <validationQuery>SELECT 1</validationQuery>
                <validationInterval>30000</validationInterval>
            </configuration>
        </definition>
    </datasource>

registry.xml

Update the <EI-HOME>/conf/registry.xml with following

(Please keep the existing dbConfig as it is and add the following as new)

<dbConfig name="sharedregistry">
    <dataSource>jdbc/WSO2SharedDB</dataSource>
</dbConfig>
<remoteInstance url="https://localhost:9443/registry">
    <id>instanceid</id>
    <dbConfig>sharedregistry</dbConfig>
    <readOnly>false</readOnly>
    <enableCache>true</enableCache>
    <registryRoot>/</registryRoot>
</remoteInstance>
<mount path="/_system/config" overwrite="true">
    <instanceId>instanceid</instanceId>
    <targetPath>/_system/esbnodes</targetPath>
</mount>
<mount path="/_system/governance" overwrite="true">
    <instanceId>instanceid</instanceId>
    <targetPath>/_system/governance</targetPath>
</mount>

user-mgt.xml

Update the <EI-HOME>/conf/user-mgt.xml with following. Remove the existing dataSource property and add the following property to UserManger/Realm/Configuration

<Property name="dataSource">jdbc/WSO2UmDB</Property>

3. Enabling clusting in axis2.xml

By editing the following in <EI-HOME>/conf/axis2/axis2.xml enable the nonWokerManger cluster

<clustering class="org.wso2.carbon.core.clustering.hazelcast.HazelcastClusteringAgent" enable="true">

<parameter name="clusteringPattern">nonWorkerManager</parameter>

<parameter name="localMemberHost">ei.wso2.com</parameter>

ei.wso2.com has to be added to /etc/hosts. Giving the local IP address (127.0.0.1) as the localMemberHost is not recommended.

Updating the member list

<members>
    <member>
        <hostName>ei.wso2.com</hostName>
        <port>4200</port>
    </member>
</members> 

In this setup both the nodes are in the same machine, that's why the member's hostname and the localMemberHost are same. If you are setting up the cluster in a different machine please add the correct host name.

4. Configuring Node 2

Take a copy of the configured EI611 pack (can be name the copy as node2), And configure the following

Configure member list

Update the local member port (4200 is what we configured in the previous node(Node 1)'s, known member list)

<parameter name="localMemberPort">4200</parameter>

Update the member list as follows
(4100 is previous node(Node 1)'s, local port number)

<members>
    <member>
        <hostName>ei.wso2.com</hostName>
        <port>4100</port>
    </member>
</members>

Configure the port offset

If the cluster is in a same machine, then the port offset must be configured inorder to avoid the port binding conflicts
Edit the <EI-HOME>/conf/carbon.xml as follows (eg: 2 as offset) <Offset>2</Offset>

5. Configure a common deployment directory

Since we do not recommend the deployment synchronizer. Here we are using a file share to share the artifacts between the 2 nodes. For the other available options kindly have a look at https://docs.wso2.com/display/EI611/Clustering+the+ESB+Profile, Deploying artifacts across the nodes First remove the existing deployment directory from node2 and create a soft link to the node1's deployment directory

node2:repository$> rm -r deployment/

node2:repository$> ln -s <path-to-node1>/reposository/deployment deployment

While loop in WSO2 ESB

How to invoke an endpoint many number of times.
There are two situations

1. The number of invocation is defined. i.e we know how many time we are going to invoke the endpoint.
   In such a situation, we can construct a mock payload with that number of elements and by iterating it we can invoke the endpoint
2. The number of invocation is not defined. i.e we don't know how many time we need to invoke the endpoint. i.e the response of previous invocation determine the number of calls

This post gives a sample ESB configuration to invoke an endpoint any number of time. Like a while loop (until a condition is satisfied)

IMPORTANT: I do NOT recommend the following configuration in a production environment. If you come across a similar situation, I recommend to revisit your use case and come up with the better way.

Anyway the following sample is really fun and shows you how powerful WSO2 ESB is. 

The idea behind creating a loop is, I am using filter mediator and based on a condition I decide whether to continue the flow or terminate. When I say continuing the flow, I will be invoking the endpoint using send mediator and dispatching the response to the same sequence which does the filter so there will be loop until it satisfies the terminating condition.

In this sample I will be using a simple data base to be updated. But I will be querying only one entry at a time and will be looping until all the entries are updated.

Setting up the sample

I am using WSO2 Enterprise Integrator 6.1.1. and MySQL as my database.

Creating the database and table

CREATE DATABASE SimpleSchool;
USE SimpleSchool;

CREATE TABLE `Student` (`Id` int(11) DEFAULT NULL, `Name` varchar(200) DEFAULT NULL, `State` varchar(4) DEFAULT NULL); 

Data Service Configuration

Remember to add the mysql driver jar into EI_HOME/lib

SimpSchool.dbs

<data name="SimpSchool" transports="http https local">
   <config enableOData="false" id="SimplSchool">
      <property name="driverClassName">com.mysql.jdbc.Driver</property>
      <property name="url">jdbc:mysql://localhost:3306/SimpleSchool</property>
      <property name="username">USERNAME</property>
      <property name="password">PASSWORD</property>
   </config>
   <query id="insertStudent" useConfig="SimplSchool">
      <sql>insert into Student (Id, Name, State) values (? ,? ,?);</sql>
      <param name="Id" sqlType="STRING"/>
      <param name="Name" sqlType="STRING"/>
      <param name="State" sqlType="STRING"/>
   </query>
   <query id="getCount" useConfig="SimplSchool">
      <sql>select count(*) as count from Student where State = 'New';</sql>
      <result element="Result" rowName="">
         <element column="count" name="count" xsdType="string"/>
      </result>
   </query>
   <query id="UpdateState" returnUpdatedRowCount="true" useConfig="SimplSchool">
      <sql>update Student set state='Done' where Id=?;</sql>
      <result element="UpdatedRowCount" rowName="" useColumnNumbers="true">
         <element column="1" name="Value" xsdType="integer"/>
      </result>
      <param name="Id" sqlType="STRING"/>
   </query>
   <query id="selectStudent" useConfig="SimplSchool">
      <sql>select Id, Name from Student where state='New' limit 1;</sql>
      <result element="Entries" rowName="Entry">
         <element column="Id" name="Id" xsdType="string"/>
         <element column="Name" name="Name" xsdType="string"/>
      </result>
   </query>
   <resource method="POST" path="insert">
      <call-query href="insertStudent">
         <with-param name="Id" query-param="Id"/>
         <with-param name="Name" query-param="Name"/>
         <with-param name="State" query-param="State"/>
      </call-query>
   </resource>
   <resource method="POST" path="getCount">
      <call-query href="getCount"/>
   </resource>
   <resource method="POST" path="select">
      <call-query href="selectStudent"/>
   </resource>
   <resource method="POST" path="update">
      <call-query href="UpdateState">
         <with-param name="Id" query-param="Id"/>
      </call-query>
   </resource>
</data>

ESB Configuration

I will be using an API to initiate the call, the API will invoke the loop_getCountAndCheck sequence which queries the database and retrieve the count, the result is being filtered and if the count is zero we call the loop_done sequence. Else we call the loop_update_logic sequence. Inside the loop_update_logic we update the data base and let the response to be processed inside the loop_getCountAndCheck sequence.

loop_getCountAndCheck 

<?xml version="1.0" encoding="UTF-8"?>
<sequence name="loop_getCountAndCheck" xmlns="http://ws.apache.org/ns/synapse">
    <payloadFactory media-type="xml">
        <format>
            <dat:_postgetcount xmlns:dat="http://ws.wso2.org/dataservice"/>
        </format>
    </payloadFactory>
    <call>
        <endpoint>
            <http method="POST" uri-template="http://localhost:8280/services/SimpSchool/getCount"/>
        </endpoint>
    </call>
    <log level="custom">
        <property expression="//n1:Result/n1:count"
            name="called get count. Remaining are:"
            xmlns:n1="http://ws.wso2.org/dataservice" xmlns:ns="http://org.apache.synapse/xsd"/>
    </log>
    <filter regex="0" source="//n1:count"
        xmlns:n1="http://ws.wso2.org/dataservice" xmlns:ns="http://org.apache.synapse/xsd">
        <then>
            <log level="custom">
                <property name="remaining are 0" value="calling done sequence"/>
            </log>
            <sequence key="loop_done"/>
        </then>
        <else>
            <log level="custom">
                <property name="remaining are not zero" value="calling update_logic sequence"/>
            </log>
            <sequence key="loop_update_logic"/>
        </else>
    </filter>
</sequence>

loop_done sequence

<?xml version="1.0" encoding="UTF-8"?>
<sequence name="loop_done" xmlns="http://ws.apache.org/ns/synapse">
    <log level="custom">
        <property name="this is done sequence" value="Responding to client"/>
    </log>
    <payloadFactory media-type="xml">
        <format>
            <done xmlns="">updating all</done>
        </format>
    </payloadFactory>
    <respond/>
</sequence>

loop_update_logic

<?xml version="1.0" encoding="UTF-8"?>
<sequence name="loop_update_logic" xmlns="http://ws.apache.org/ns/synapse">
    <log level="custom">
        <property name="this is logic sequence" value="selecting entries"/>
    </log>
    <payloadFactory media-type="xml">
        <format>
            <dat:_postselect xmlns:dat="http://ws.wso2.org/dataservice"/>
        </format>
    </payloadFactory>
    <call>
        <endpoint>
            <http method="POST" uri-template="http://localhost:8280/services/SimpSchool/select"/>
        </endpoint>
    </call>
    <log level="custom">
        <property name="data queried" value="for updating"/>
    </log>
    <payloadFactory media-type="xml">
        <format>
            <dat:_postupdate xmlns:dat="http://ws.wso2.org/dataservice">
                <dat:Id>$1</dat:Id>
            </dat:_postupdate>
        </format>
        <args>
            <arg evaluator="xml" expression="//n1:Id" literal="false"
                xmlns:n1="http://ws.wso2.org/dataservice" xmlns:ns="http://org.apache.synapse/xsd"/>
        </args>
    </payloadFactory>
    <log level="custom">
        <property name="data constructed" value="for updating"/>
    </log>
    <send receive="loop_getCountAndCheck">
        <endpoint>
            <http method="POST" uri-template="http://localhost:8280/services/SimpSchool/update"/>
        </endpoint>
    </send>
</sequence>

API configuration

<api xmlns="http://ws.apache.org/ns/synapse" name="UpdateInLoop" context="/loop">
   <resource methods="GET">
      <inSequence>
         <sequence key="loop_getCountAndCheck"/>
      </inSequence>
   </resource>
</api>

Sample data to fill the Student table

INSERT INTO Student (Id, Name, State) values (1, 'AAAA' , 'New'), (2, 'BBBB', 'New'), (3, 'CCCC' , 'New'), (4, 'DDDD' , 'New');

Sample curl request

curl -v http://localhost:8280/loop

Console output on a happy path

[2017-08-07 16:58:52,474] [EI-Core]  INFO - LogMediator called get count. Remaining are: = 4
[2017-08-07 16:58:52,474] [EI-Core]  INFO - LogMediator remaining are not zero = calling update_logic sequence
[2017-08-07 16:58:52,475] [EI-Core]  INFO - LogMediator this is logic sequence = selecting entries
[2017-08-07 16:58:52,488] [EI-Core]  INFO - LogMediator data queried = for updating
[2017-08-07 16:58:52,489] [EI-Core]  INFO - LogMediator data constructed = for updating
[2017-08-07 16:58:52,517] [EI-Core]  INFO - LogMediator called get count. Remaining are: = 3
[2017-08-07 16:58:52,518] [EI-Core]  INFO - LogMediator remaining are not zero = calling update_logic sequence
[2017-08-07 16:58:52,518] [EI-Core]  INFO - LogMediator this is logic sequence = selecting entries
[2017-08-07 16:58:52,524] [EI-Core]  INFO - LogMediator data queried = for updating
[2017-08-07 16:58:52,524] [EI-Core]  INFO - LogMediator data constructed = for updating
[2017-08-07 16:58:52,553] [EI-Core]  INFO - LogMediator called get count. Remaining are: = 2
[2017-08-07 16:58:52,553] [EI-Core]  INFO - LogMediator remaining are not zero = calling update_logic sequence
[2017-08-07 16:58:52,553] [EI-Core]  INFO - LogMediator this is logic sequence = selecting entries
[2017-08-07 16:58:52,559] [EI-Core]  INFO - LogMediator data queried = for updating
[2017-08-07 16:58:52,559] [EI-Core]  INFO - LogMediator data constructed = for updating
[2017-08-07 16:58:52,586] [EI-Core]  INFO - LogMediator called get count. Remaining are: = 1
[2017-08-07 16:58:52,586] [EI-Core]  INFO - LogMediator remaining are not zero = calling update_logic sequence
[2017-08-07 16:58:52,587] [EI-Core]  INFO - LogMediator this is logic sequence = selecting entries
[2017-08-07 16:58:52,597] [EI-Core]  INFO - LogMediator data queried = for updating
[2017-08-07 16:58:52,598] [EI-Core]  INFO - LogMediator data constructed = for updating
[2017-08-07 16:58:52,619] [EI-Core]  INFO - LogMediator called get count. Remaining are: = 0
[2017-08-07 16:58:52,620] [EI-Core]  INFO - LogMediator remaining are 0 = calling done sequence
[2017-08-07 16:58:52,620] [EI-Core]  INFO - LogMediator this is done sequence = Responding to client

The whole flow is working on the same message context. I haven't done any error handling here. We can introduce a property and by incrementing it in each iteration, we can further control the loop.

Cheers !

WSO2 ESB - Powerful Capabilities

WSO2 Enterprise Service Bus is a lightweight, high performance, near-zero latency product, providing comprehensive support for several different technologies like SOAP, WS* and REST as well as domain-specific solutions and protocols like SAP, FIX and HL7. It goes above and beyond by being 100% compliant with enterprise integration patterns. It also has 160+ ready-made, easy-to-use connectors to seamlessly integrate between cloud service providers. WSO2 Enterprise Service Bus is 100% configuration driven, which means no code needs to be written. Its capabilities can be extended too with the many extension points to plug into.

In the IT world it is vital to communicate among the heterogeneous systems. WSO2 ESB helps you integrate services and applications in an easy, efficient and productive manner.
WSO2 ESB, a 100% open source enterprise service bus help us transforming data seamlessly across different formats and transports.

WSO2 Enterprise Service Bus is the main integration engine of WSO2 Enterprise Integrator

Following are some of the Powerful capabilities of WSO2 ESB

• Service mediation
• Help achieve separation of concerns with respect to business logic design and messaging
• Shield services from message formats and transport protocols
• Offload quality of service aspects such as security, reliability, caching from business logic
  
• Message routing
• Route, filter, enrich and re-sequence messages in a content aware manner or content unaware manner (without regard to the content) and using rules
  
• Data transformation
• Transform data across varying formats and media types to match data acceptance criteria of various services and applications
  
• Data transportation
• Support for various transport protocols based on data formats and data storage and destination characteristics including HTTP, HTTPS, JMS, VFS
  
• Service hosting
• It is feasible with WSO2 ESB to host services, however, this could become an anti pattern if used in combination with service mediation and service hosting when considering layered deployment for separation of concerns between mediation and hosting

To Learn more on What is WSO2 ESB please check the article written by Samisa Abeysinghe  (Chief Engineering and Delivery Officer at WSO2)  

http://wso2.com/library/articles/2017/07/what-is-wso2-esb/

WSO2 ESB Iterate mediator sample

Following is a small API using WSO2 ESB iterate mediator.

<api xmlns="http://ws.apache.org/ns/synapse" name="simpleIterator" context="/simpleIterator">
   <resource methods="POST">
      <inSequence>
         <property name="root_element">
            <responses/>
         </property>
         <iterate expression="//symbols/symbol" sequential="true">
            <target>
               <sequence>
                  <property name="messageType" value="application/xml" scope="axis2"/>
                  <send>
                     <endpoint>
                        <http method="POST" uri-template="http://localhost:8080/sample-responses/simpleQuote"/>
                     </endpoint>
                  </send>
               </sequence>
            </target>
         </iterate>
      </inSequence>
      <outSequence>
         <aggregate>
            <completeCondition>
               <messageCount min="-1" max="-1"/>
            </completeCondition>
            <onComplete expression="//response" enclosingElementProperty="root_element">
               <log level="full"/>
               <respond/>
            </onComplete>
         </aggregate>
      </outSequence>
   </resource>
</api>
                        

sample (curl) request :

curl -d @symbols.xml -H "Content-Type: application/xml"  http://localhost:8280/simpleIterator -v 

symbols.xml

<symbols>
   <symbol>WSO2</symbol>
   <symbol>MSFT</symbol>
   <symbol>IBM</symbol>
   <symbol>ADBE</symbol>
   <symbol>AAPL</symbol>
   <symbol>ORCL</symbol>
   <symbol>RHT</symbol>
   <symbol>FB</symbol>
   <symbol>TWTR</symbol>
   <symbol>LNKD</symbol>
</symbols>

Sample backend can be found here.

Useful XPath expressions

The following WSO2 ESB proxy service creates a sample XML payload and logs some useful XPath expressions

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51

<?xml version="1.0" encoding="UTF-8"?> <proxy xmlns="http://ws.apache.org/ns/synapse" name="xpathSample" transports="https,http" statistics="disable" trace="disable" startOnLoad="true"> <target> <inSequence> <payloadFactory media-type="xml"> <format> <students xmlns=""> <student student-id="001"> <name first="tom" last="hanks" /> <age>30</age> <scores> <subject id="sub1">70</subject> <subject id="sub2">80</subject> </scores> </student> <student student-id="002"> <name first="tom" last="cruise" /> <age>25</age> <scores> <subject id="sub1">60</subject> <subject id="sub2">75</subject> </scores> </student> <student student-id="003"> <name first="jessica" last="chastain" /> <age>24</age> <scores> <subject id="sub1">75</subject> <subject id="sub2">70</subject> </scores> </student> </students> </format> <args /> </payloadFactory> <log level="custom" separator="&#xA;"> <property name="student 001's first name" expression="//student[@student-id='001']/name/@first" /> <property name="student 001's last name" expression="//student[@student-id='001']/name/@last" /> <property name="student 001's age" expression="//student[@student-id='001']/age" /> <property name="student 001's sub1 score" expression="//student[@student-id='001']/scores/subject[@id='sub1']/text()" /> <property name="student 001's sub2 score" expression="//student[@student-id='001']/scores/subject[@id='sub2']/text()" /> <property name="sub2 score for students whose first name is 'tom'" expression="//student[name[@first='tom']]/scores/subject[@id='sub2']" /> <property name="sub2 score for students whose first name is 'tom' [with XPath2.0 function]" expression="string-join(//student[name[@first='tom']]/scores/subject[@id='sub2']/text(), ', ')" /> <property name="students' last name whose age is lesser than '30'" expression="string-join(//student[age/text() &lt; 30]/name/@last, ', ')" /> <property name="students' last name whose age is greater than '25' OR sub1 score is greater than or equal to '70'" expression="string-join(//student[age/text() /text() &gt; 25 or scores/subject[@id='sub1'] &gt;= 70]/name/@last, ', ')" /> </log> <respond /> </inSequence> </target> <description /> </proxy>

once the proxy is invoked using tryit or soapUI you can see the following logs in the terminal

student 001's first name = tom
student 001's last name = hanks
student 001's age = 30
student 001's sub1 score = 70
student 001's sub2 score = 80
sub2 score for students whose first name is 'tom' = 8075
sub2 score for students whose first name is 'tom' [with XPath2.0 function] = 80, 75
students' last name whose age is lesser than '30' = cruise, chastain
students' last name whose age is greater than '25' OR sub1 score is greater than or equal to 70 = hanks, chastain

Please note that, in Line 43, 44 and 45 I am using stirng-join(..) function to concat the results with comma. This function is only supported in XPath 2.0 version. But WSO2 ESB supports XPath 1.0 by default. To enable XPath 2.0 support in WSO2 ESB following line should be uncommented in CARON_HOME/repository/conf/synapse.properties

synapse.xpath.dom.failover.enabled=true