How to Integrate WSO2 ESB with WSO2 MB as a Message Broker

This post will describe how to integrate WSO2 ESB with WSO2 Message Broker. This will provide the configuration required for ESB to connect to the Message Broker.

I am using the WSO2 EI 6.1.1 since It has both Integrator and Broker product reside as the profiles.

First we need to start the message broker. Message broker profile can be found under EI_HOME/wso2/broker directory.

1) Start the the message broker by executing the broker.sh in ESB_HOME/bin
     sh broker.sh
As an alternative, you can start the broker by executing the wso2server.sh under EI_HOME/wso2/broker/bin directory. Both will do the same thing that message broker starts.


Now Message broker is up and running. We need to configure the Integrator(ESB) with the broker details.

2) Add the below configuration in jndi.properties file under the EI_HOME/conf directory if they are not exist. The configuration has the host name and the ports information where broker is running.

 connectionfactory.QueueConnectionFactory = amqp://admin:admin@clientID/carbon?brokerlist='tcp://localhost:5675'  
 connectionfactory.TopicConnectionFactory = amqp://admin:admin@clientID/carbon?brokerlist='tcp://localhost:5675'  

 

3) Enable the JMS transport listener and sender configuration in axis2.xml of the Integrator. It can be found on EI_HOME/conf/axis2/axis2.xml
  by default It is commented out. So It has to be enabled as bellow.
  The java.naming.provider.url define the conf/jndi.properties and according to the configuration
  in jndi.properties file we have configured above, Integrator will connect with the broker.

 <transportReceiver name="jms" class="org.apache.axis2.transport.jms.JMSListener">  
     <parameter name="myTopicConnectionFactory" 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">TopicConnectionFactory</parameter>  
       <parameter name="transport.jms.ConnectionFactoryType" locked="false">topic</parameter>  
     </parameter>  
     <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>  
     <parameter name="default" 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>  
   </transportReceiver>  

  <transportSender name="jms" class="org.apache.axis2.transport.jms.JMSSender"/>  

Now All the requied configuration has been done in Integrator. You can start the integrator.

4) Start the Integrator.
      go to the EI_HOME/bin and run the integrator.sh
      sh integrator.sh

The integrator is ready to consume message and publish message with WSO2 MB. You can deploy jms services such as JMS proxy service or Inbound Endpoint in ESB and They will consume messages from Message Broker.

Durable Subscription in WSO2 ESB with WSO2 MB

In this post, We are going to implement a jms service in WSO2 ESB which has the durable subscription with WSO2 MB Topic. The Durable Subscriber avoids missing messages while subscriber not listening for topic. Once the subscriber appears online, The messages will reached to the subscriber.

It is always better to implement an inbound endpoint in WSO2 ESB when the subscription is durable instead of JMS proxy service. Since durable subscription creates a subscription with Message broker with the given subcription name, another service can not create a subscription with the same name. If a JMS proxy service is created with the durable subscription name, It will work for single instance. But will cause the issue in clustered environment because of It gets deployed in all instances of the ESB cluster.

Inbound endpoint has the coordination support in the cluster. So that It can only running in a single instance at a given time. So there will be no issue with durable subscription too. Once a node which runs inbound endpoint went down, It automatically schedule to run another node in a cluster. 

For this sample demonstration, The WSO2 EI 6.1.1 used since It has the Integrator(ESB) and Broker(MB) as the profile. So using one product we can implement above scenario.

1) First we need to start a WSO2 Message Broker.

    go to the EI_HOME/bin and run the broker.sh

    sh broker.sh

Then It will start the broker profile

You need to configure the Integrator to connect with broker. Please follow the link How to Integrate WSO2 ESB with WSO2 MB for configuring integrator if you have not configured integrator.

2) Start the Integrator.
      go to the EI_HOME/bin and run the integrator.sh
      sh integrator.sh

Now we have configured Message broker and Integrator. Both servers are up and running. Need to deploy the inbound endpoint on the integrator to consume the message from MB topic.

Below is the inbound configuration. Once It deployed, It will creates a subscription with MB and started to consume messages from the topic defined with transport.jms.Destination.
The inbound endpoint will inject the messages once It pull from topic to to the sequence LogMsgSeq and If there is any error while mediating the message, It will inject to LogErrorSeq.

 <?xml version="1.0" encoding="UTF-8"?>  
 <inboundEndpoint name="DurableTopicInboundListener" onError="LogErrorSeq"   
 protocol="jms" sequence="LogMsgSeq" 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">TestInboundTopicDurable</parameter>  
     <parameter name="transport.jms.CacheLevel">3</parameter>  
     <parameter name="transport.jms.ConnectionFactoryJNDIName">TopicConnectionFactory</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">false</parameter>  
     <parameter name="transport.jms.SubscriptionDurable">true</parameter>  
     <parameter name="transport.jms.ConnectionFactoryType">topic</parameter>  
     <parameter name="transport.jms.DurableSubscriberClientID">test_inbound_sbscrb</parameter>  
     <parameter name="transport.jms.ContentType">application/json</parameter>  
     <parameter name="transport.jms.DurableSubscriberName">test_inbound_sbscrb</parameter>  
     <parameter name="transport.jms.SharedSubscription">false</parameter>  
   </parameters>  
 </inboundEndpoint>  

 <sequence name="LogMsgSeq" trace="disable" xmlns="http://ws.apache.org/ns/synapse">  
   <log description="msg" level="full"/>  
 </sequence>  

 <sequence name="LogErrorSeq" trace="disable" xmlns="http://ws.apache.org/ns/synapse">  
   <log description="Error" level="custom">  
     <property expression="$ctx:ERROR_MESSAGE" name="Error Message"/>  
   </log>  
 </sequence>  

In above configuration, transport.jms.DurableSubscriberName and transport.jms.DurableSubscriberClientID is important when creating durable subscription with MB. Without the parameter it wont be a durable subscription. And please make sure not to add any inbound configuration or proxy service with the same DurableSubscriberName and ID. Those parameter should be unique for services to service.

Now you can test the inbound endpoint by publishing message to the topic. If you log into the management console of MB, you can see a topic named TestInboundTopicDurable. Publish a message to that topic. Then Integrator terminal will log the message content by pulling the message from the topic. Then stop the Integration and publish messages to topic. Again star the Integrator and you can see Integrator consumes thos messages as well even It had been stopped at the time of messages was published.

Java Keytool Comands

Java Keytool is useful tool for managing the certificates. It allows users to manage their own public/private key pairs and certificates. Below commands will be used when working on keytool.

Create a JKS file with a private certificate.

 keytool -genkey -alias nuwanw -keyalg RSA -keysize 1024 -keypass mypassoword  
 -keystore sample.jks -storepass mypassword  

List the alias of a given keystore

 keytool -list -v -keystore sample.jks -storepass mypassword  

Export public certificate of the service alies nuwanw

 keytool -export -alias nuwanw -keystore sample.jks -storepass mypassword   
 -file publicCert.pem  

Import public certificate into a Trust Store.

 keytool -import -alias nuwanw -file serviceCert.pem -keystore sample-trust-store.jks   
 -storepass mypassword  

Convert XML payload to JSON payload in a WSO2 ESB Class Mediator

Bellow class mediator can access the XML payload and convert it to JSON message and set It as a payload.

package org.wso2.demo;

import org.apache.axiom.om.OMElement;
import org.apache.synapse.MessageContext;
import org.apache.synapse.commons.json.JsonUtil;
import org.apache.synapse.core.axis2.Axis2MessageContext;
import org.apache.synapse.mediators.AbstractMediator;
import org.json.JSONObject;

public class ConvertXmlToJSON  extends AbstractMediator {

    public boolean mediate(MessageContext context) {
        try {
            org.apache.axis2.context.MessageContext axis2MessageContext = ((Axis2MessageContext) context)
                    .getAxis2MessageContext();
            OMElement xmlResponse = context.getEnvelope().getBody().getFirstElement();
            xmlResponse.build();
            JSONObject jsonObject = new JSONObject(JsonUtil.toJsonString(xmlResponse).toString());

            //setting the payload as the message payload
            JsonUtil.getNewJsonPayload(axis2MessageContext, jsonObject.toString(), true, true);
        } catch (Exception e) {
            handleException("Error while converting the message", e, context);
        }
        return true;
    }
}

Working with JSON Message in WSO2 ESB Class Mediator

                 How to access the JSON payload from the class mediator

WSO2 ESB provides inbuilt mediators for mediating the messages. However when It comes to custom requirement, Class mediators are very useful. Using a java class mediator, you can do any custom modification to the message. Here this sample will demonstrate how to access the JSON message payload within class mediator and set a new JSON message as the payload.

package org.wso2.demo;

import org.apache.synapse.MessageContext;
import org.apache.synapse.commons.json.JsonUtil;
import org.apache.synapse.core.axis2.Axis2MessageContext;
import org.apache.synapse.mediators.AbstractMediator;
import org.json.JSONObject;

/**
 * This mediator will replace the element fName to name
 * {"fName" : "Nuwan"} --> {"name" : "Nuwan"}
 */

public class HandleJSONPayload extends AbstractMediator {

    public boolean mediate(MessageContext context) {
        try {
            org.apache.axis2.context.MessageContext axis2MessageContext = ((Axis2MessageContext) context)
                    .getAxis2MessageContext();
            JSONObject jsonPayload = new JSONObject(JsonUtil.jsonPayloadToString(axis2MessageContext));

            String name = jsonPayload.getString("fName");
            jsonPayload.remove("fName");
            jsonPayload.put("name", name);

            //setting the payload as the message payload
            JsonUtil.getNewJsonPayload(axis2MessageContext, jsonPayload.toString(), true, true);
        } catch (Exception e) {
            handleException("Error while mediating the message", e, context);
        }
        return true;
    }
}

     

You need to add a additional dependency to access the JSON classes.

 <dependency>
            <groupId>org.json</groupId>
            <artifactId>json</artifactId>
            <version>20090211</version>
 </dependency>

Benefits of WSO2 ESB

WSO2 Enterprise Service Bus (ESB) is a lightweight, high performance, comprehensive and open source middleware application that enables communication among various heterogeneous systems and business applications. Instead of having to make each of your applications communicate directly with each other in all their various formats, each application simply communicates with the ESB, which handles transforming and routing the messages to their appropriate destinations.

 

Integrating various business systems by creating service and API is just one step away from a xml configuration with WSO2 ESB. You can try out by downloading WSO2 ESB and make your systems integrated.

Some of key benefits of having WSO2 ESB.

• Enables communication among various heterogeneous applications
• Uses event-driven and standards-based messaging engine
• Performs variety of Enterprise Integration Patterns (EIPs)
• Tooling support for artifact(Service and API) development
• Message tracing and Analytic capabilities.
• Mediation debugger capability. 
• Open Standards (WSDL, SOAP, JMS, HTTP, FTP…)
• Supports various protocols(HTTP,  HTTPS, JMS, FTP, VFS, ..etc) and Protocol Switch
• Supports message format transformations
• Supports message routing 
• Supports message enrichment
• Connectors(A ready made tool used to connect to public web APIs) that allow ESB to interact with third party services such as Twitter, Salesforce, Google Docs, JIRA

 Learn more about WSO2 ESB from below
 http://wso2.com/library/articles/2017/07/what-is-wso2-esb

Secure vault support in WSO2 ESB inbound parameters

In ESB 5.0.0, wso2:vault-lookup('alias') expression can be used in all inbound endpoint parameters. even in registry resource text, valut-lookup() expression can be used and set resource as the parameter value.

Parameter value  need to be encrypted and store with a alias and retrieve the actual value in your ESB  inbound configuration as below.

Before adding alias, You need to run the cipher tool in ESB. Go to bin directory and issue
"sh ciphertool.sh -Dconfigure"
Then go to the management console and click on "Manage Password" and create a alias my.username and my.password. here you can store the encrypted value for given values.

Add a registry resource(conf:/userInfo/password) which contains  {wso2:vault-lookup('my.password')} as the text contains.

Then you can use the alias in your inbound parameters with lookup fuction.
 Ex: wso2:vault-lookup('my.password')

<inboundEndpoint xmlns="http://ws.apache.org/ns/synapse"  
          name="activemq"  
          sequence="seq1"  
          onError="fault"  
          protocol="jms"  
          suspend="false">  
   <parameters>  
    <parameter name="interval">10</parameter>  
    <parameter name="sequential">true</parameter>  
    <parameter name="coordination">true</parameter>  
    <parameter name="java.naming.factory.initial">org.apache.activemq.jndi.ActiveMQInitialContextFactory</parameter>  
    <parameter name="java.naming.provider.url">tcp://localhost:61616</parameter>  
    <parameter name="transport.jms.ConnectionFactoryJNDIName">QueueConnectionFactory</parameter>  
    <parameter name="transport.jms.ConnectionFactoryType">queue</parameter>  
    <parameter name="transport.jms.Destination">order</parameter>  
    <parameter name="transport.jms.SessionTransacted">false</parameter>  
    <parameter name="transport.jms.SessionAcknowledgement">AUTO_ACKNOWLEDGE</parameter>  
    <parameter name="transport.jms.CacheLevel">3</parameter>  
    <parameter name="transport.jms.UserName">{wso2:vault-lookup('my.username')}</parameter>  
    <parameter name="transport.jms.Password" key="conf:/userInfo/password"/>  
    <parameter name="transport.jms.SubscriptionDurable">false</parameter>  
    <parameter name="transport.jms.SharedSubscription">false</parameter>  
   </parameters>  
 </inboundEndpoint>