Revisiting ESB

Messaging Architecture

High-level Architecture

Adding proxy server behind WSO2 ESB

When the message flow needs to be routed through a proxy, you need to add following parameters to transportSender configuration in axis2.xml.

http.proxyHost - proxy server's host name
http.proxyPort - port number of the proxy  server
http.nonProxyHosts - any host that need to by pass above proxy

Else you can set Java networking properties.

-Dhttp.proxyHost=example.org -Dhttp.proxyPort=5678 -Dhttp.nonProxyHosts=localhost|127.0.0.1|foo.com

Sample:
This scenario illustrates how a routing via proxy and nonproxy happens. I have echo service in an AppServer which is fronted by stockquaote.org HTTP proxy. I also have SimpleStockQuoteService in localhost which is set as a nonProxyHost.

I have my transport sender in axis2.xml configured as below:

<transportSender name="http" class="org.apache.synapse.transport.passthru.PassThroughHttpSender">
 <parameter name="non-blocking" locked="false">true</parameter>
 <parameter name="http.proxyHost" locked="false">stockquote.org</parameter>
 <parameter name="http.proxyPort" locked="false">8080</parameter>
 <parameter name="http.nonProxyHosts" locked="false">localhost</parameter>
</transportSender> 

Proxy rout:

<proxy name="Echo_viaProxy" transports="https http" startOnLoad="true" trace="disable">
  <description/>
  <target>   
   <endpoint>       
    <address uri="http://xx.xx.xx.xxx:9765/services/Echo"/>    
   </endpoint>    
  <outSequence>      
   <send/>    
  </outSequence>  
 </target>
</proxy>

When you send a request to above SimpleStckQ_viaProxy, the request will be direct to stockquaote.org which will route to BE (http://localhost:9000/services/SimpleStockQuoteService).

nonProxy:

<proxy name="StockQuote_direct"
       transports="https http"
       startOnLoad="true"
       trace="disable">
   <description/>
   <target>
      <endpoint>
         <address uri="http://localhost:9000/services/SimpleStockQuoteService"/>
      </endpoint>
      <outSequence>
         <send/>
      </outSequence>
   </target>
 </proxy>

When you send a request to above StockQuote_direct, the request will be directly served by SimpleStockService in localhost.


known issue and fix:
https://wso2.org/jira/browse/ESBJAVA-3165 is fixed in 4.9.0 M4.

Using operations scope to hold my values while iterating

Iterator mediator breaks a message from the given xpath pattern and produces smaller messages. If you need to collect an attribute value throughout the iteration, how do you do it?

With lot of help from IsuruU, I managed to workout this solution. In here I am Iterating through this message [a], and breaking it to sub messages using this [b] pattern. I need to collect the values of '//m0:symbol' in a property and send the processed values to client and values of failed messages to failure sequence.

 [a]  - Request message

  <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
            <soapenv:Body>
         <m0:getQuote xmlns:m0="http://services.samples">
            <m0:request>
               <m0:symbol>IBM</m0:symbol>
            </m0:request>
            <m0:request>
               <m0:symbol>WSO2</m0:symbol>
            </m0:request>
            <m0:request>
               <m0:symbol>AAA</m0:symbol>
            </m0:request>
            <m0:request>
               <m0:symbol>SUN</m0:symbol>
            </m0:request>
         </m0:getQuote>
      </soapenv:Body>
   </soapenv:Envelope> 

[b] xpath expression

//m0:getQuote/m0:request" 

Here is my insquence:

    <inSequence>
         <iterate xmlns:m0="http://services.samples"
                  continueParent="true"
                  preservePayload="true"
                  attachPath="//m0:getQuote"
                  expression="//m0:getQuote/m0:request"
                  sequential="true">
            <target>
               <sequence>
                  <property name="PID"
                            expression="fn:concat(get-property('operation','PID'),//m0:symbol,' ')"
                            scope="operation"
                            type="STRING"/>
                  <store messageStore="pid_store"/>
               </sequence>
            </target>
         </iterate>
         <log level="custom">
            <property name="Processed_PIDs" expression="get-property('operation','PID')"/>
         </log>
         <payloadFactory media-type="xml">
            <format>
               <ax21:getQuoteResponse xmlns:ax21="http://services.samples/xsd">
                  <ax21:pid>$1</ax21:pid>
               </ax21:getQuoteResponse>
            </format>
            <args>
               <arg xmlns:ns="http://org.apache.synapse/xsd"
                    evaluator="xml"
                    expression="get-property('operation','PID')"/>
            </args>
         </payloadFactory>
         <respond/>
      </inSequence>

Let me explain above; With 'expression="//m0:getQuote/m0:request"' the request message will be split into different messages as I described earlier. Since my scenario was to collect a given value from each of the split message and send them to appropriate path as a single message, I have used continueParent="true" and sequential="true". By this I am making sequential processing instead of default parallel processing behaviour of iterator.

Then as a target sequence within iterator to mediate the split message, I have opened a property mediator. Using this, I am collecting the value of //m0:symbol and storing it in a variable (property name) 'PID'.

The scope of the PID property was set to scope=operations to preserve the property within iterated message flow.

Later, as per the initial requirement the message is sent to a message store. A log is printed on property for ease of tracking. Then I prepared a payload to send the message with PID as an attribute.

Fault sequence was done like this to capture faulty messages and pass their PIDs.

<faultSequence>
         <log level="full">
            <property name="MESSAGE"
                      value="--------Executing default &#34;fault&#34; sequence--------"/>
            <property name="ERROR_CODE" expression="get-property('ERROR_CODE')"/>
            <property name="ERROR_MESSAGE" expression="get-property('ERROR_MESSAGE')"/>
         </log>
         <property xmlns:m0="http://services.samples"
                   name="PID"
                   expression="fn:substring(fn:concat(get-property('operation','PID'),//m0:symbol,' '),1,(fn:string-length(get-property('operation','PID'))-1))"
                   scope="operation"
                   type="STRING"/>
         <log level="custom">
            <property name="Failed_PIDs" expression="get-property('operation','PID')"/>
         </log>
         <payloadFactory media-type="xml">
            <format>
               <ax21:getQuoteResponse xmlns:ax21="http://services.samples/xsd">
                  <ax21:pid>$1</ax21:pid>
               </ax21:getQuoteResponse>
            </format>
            <args>
               <arg xmlns:ns="http://org.apache.synapse/xsd"
                    evaluator="xml"
                    expression="get-property('operation','PID')"/>
            </args>
         </payloadFactory>
         <respond/>
      </faultSequence>

The complete proxy configuration can be found here. TCPMon outputs are attached below for further clarity.

useOriginalwsdl

I learnt a practical example on using 'useOriginalwsdl' parameters in an ESB proxy service which has a wsdl associated to it.  Thanks Charitha - as this was by listening to how Charitha approached and handled an ESB issue.

Firstly, there is a proxy service which is associating a wsdl and we were to access an operation in it via soap-ui. This particular wsdl-operation has a soap header specified. Issue being addressed is that the soap header is not appearing in the soap-request that was generated by soap-ui.

What we were doing to generate this request was generating ?wsdl of the proxy and using it in soap-ui.

When analyzing this auto generated wsdl (?wsdl), we found out that some of the parameters in the original wsdl are not appearing in it. This is because at the auto generation process ESB recreates the wsdl treating it as part of the proxy.

But for us to avoid this and have the wsdl syntax as its original we could use the parameter 'useOriginalwsdl'. After this correction, when the ?wsdl was used in soap-ui project we were able to see the correct request.

So when creating a proxy with a wsdl associated, if we need to make sure original wsdl is available for message invocation, we need to set useOriginalwsdl=true.
 

 Just now I read about enablePublishWSDLSafeMode  parameter in ESb proxies from one of Prabath's blog posts. Will write about it after trying the scenario.

Activity monitoring using WSO2 BAM Server

In this scenario we will be monitoring requests and responses passed through a proxy service in WSO2 ESB. The proxy service is calling an in/out operation in an Axis2Service in WSO2 WSAS.


Lets make the setup.

A) Deploy the axis2 service in WSO2 App Server.

1. Download WSO2 App Server-3.2.1 distribution from here and extract it.

In this setup we are going to have 3 WSO2 servers running at the same time. Therefore, we need to change the http/https ports of each server to make the servers start in different ports. To do this go to the CARBON_HOME/repository/conf and update the http/https ports in mgt-transports.xml and axis2.xml files.

2. Start WSO2 App Server distribution by running the startup scripts in WSAS_HOME/bin/.
$ sh wso2server.sh in Ubuntu and
$ wso2server.bat in Windows.

3. Once the server is started, Sign-in to admin console and go to Services >Add >Axis2Service and upload this service.

Now we have WSO2 WSAS waiting with a service in it at http://192.168.1.4:9764/wsas/services/Axis2Service?wsdl. Let's setup WSO2 ESB.



B) Configure WSO2 ESB

1. Download a WSO2 ESB-3.0.1 distribution from here and extract it. Change its ports and start the server as instructed above.

2. Once the server is started, Sign-in to admin console and go to Manage > Service Bus > Source View. Paste the following configuration into the source view.

<proxy name="DemoProxy1" transports="https http" startOnLoad="true" trace="disable" statistics="enable">
        <target>
            <endpoint name="endpoint_act">
                <address uri="http://192.168.1.4:9764/wsas/services/Axis2Service/" statistics="enable"/>
            </endpoint>
            <inSequence statistics="enable">
                <log>
                    <property name="IN" value="*********IN****************"/>
                </log>
            </inSequence>
        </target>
        <publishWSDL uri="http://192.168.1.4:9764/wsas/services/Axis2Service?wsdl"/>
    </proxy>

NOTE: You may need to chnage the endpoint epr according to your network settings.


C) Add publisher jars to ESB.

We need to enable Activity Publishing in ESB. Since ESB doesn't have activity publishing feature in it by default, we need to add the relevant jars manually.
So download following org.wso2.carbon.bam.data.publisher.activity.mediation-3.0.1.jar & org.wso2.carbon.bam.data.publisher.activity.mediation.ui-3.0.1.jar and copy to ESB_HOME/repository/components/dropins/ folder. This will add 'Activity Publishing' menu to ESB admin console under Configure menu.

Also we need to update the org.wso2.carbon.bam.data.publisher.activity.mediation-3.0.1.jar in ESB_HOME/repository/components/plugins with this patched version.

Lastly we need to enable message tracing in ESB. Open ESB_HOME/repository/conf/carbon.xml and add this;

<MediationStat>
    <MessageTracing>enabled</MessageTracing>
</MediationStat>

In the next step access ESB UI and enable message tracing. Sign-in to ESB admin console, navigate to 'Activity Publishing. In the 'Activity Publisher Configuration' page enable message tracing, go to Activity Publishing and enable eventing.
Message Threshold will be set to 2 by default. This is the minimum number of messages you need to have in the que before being read by a subscriber.
Enable 'Message Dumping' and 'Message Lookup'. Xpath expression is the value from which your message will be filtered for the subscriber.

After configurations on ESB side is done, we need to configure WSO2 BAM to monitor messages from and to ESB.



D) Configure BAM


1. Download WSO2 BAM-1.2.0 distribution from here and extract it. Change its ports and start the server as instructed above.

NOTE: BAM is configured to use H2 database by default. If you want to connect to oracle, mySql or an MS SQL db you need to create the database by running a script and update driverClassName,url, username, password in datasources.properties file which resides in BAM_HOME/repository/conf. Details of configuring to other DBMSs are explained here.

2. Now start WSO2 BAM server and sign-in to the admin console.

3. Navigate to Configure > Monitored Servers > Add Servers.

4. Add the server information of your ESB server, which we are going to monitor. Since we are monitoring mediation, our data collection method should be 'eventing' and type of data thats traced will be 'message'. So the coonfiguration is:

Server URL : https://<server IP>:<server port>
Data Collection method : Eventing
Type of data : Messagega
User Name : user name for ESB
Password : Password for ESB


After this you will have an entry in the 'Monitored Servers' with the information you gave above.

E) Monitor the messages

1. Using a client send messages to the proxy service we created above. i am using following java client.

/*
*Copyright (c) 2005-2010, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
*WSO2 Inc. licenses this file to you under the Apache License,
*Version 2.0 (the "License"); you may not use this file except
*in compliance with the License.
*You may obtain a copy of the License at
*
*http://www.apache.org/licenses/LICENSE-2.0
*
*Unless required by applicable law or agreed to in writing,
*software distributed under the License is distributed on an
*"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
*KIND, either express or implied.  See the License for the
*specific language governing permissions and limitations
*under the License.
*/

package org.wso2.carbon.test;

import org.apache.axiom.om.OMAbstractFactory;
import org.apache.axiom.om.OMElement;
import org.apache.axiom.om.OMFactory;
import org.apache.axiom.om.OMNamespace;
import org.apache.axis2.AxisFault;
import org.apache.axis2.Constants;
import org.apache.axis2.addressing.EndpointReference;
import org.apache.axis2.client.Options;
import org.apache.axis2.client.ServiceClient;
import org.apache.axis2.transport.http.HTTPConstants;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;

import javax.xml.namespace.QName;

public class EsbDemoProxy {

    private static EndpointReference targetEPR = new EndpointReference(
            "http://10.100.1.120:8280/bam/services/DemoProxy1");

    public static OMElement echoPayload(String x) {
        OMFactory fac = OMAbstractFactory.getOMFactory();
        OMNamespace omNs = fac.createOMNamespace("http://service.carbon.wso2.org", "example");
        OMElement method = fac.createOMElement("echoInt", omNs);
        OMElement value = fac.createOMElement("x", omNs);
        value.addChild(fac.createOMText(value, x));
        method.addChild(value);
        return method;
    }



    public static void main(String[] args) throws AxisFault {

        // creates a new connection manager and a http client object
        MultiThreadedHttpConnectionManager httpConnectionManager = new MultiThreadedHttpConnectionManager();
        HttpClient httpClient = new HttpClient(httpConnectionManager);
        for (int a = 0; a < 1; a++) {
            for (int i = 0; i < 1; i++) {
                ServiceClient sender = new ServiceClient();
                try {
                    OMElement payload1 = EsbDemoProxy.echoPayload("55");
                    Options options = new Options();
                    options.setTo(targetEPR);
                    options.setTransportInProtocol(Constants.TRANSPORT_HTTP);

                    OMFactory omFactory = OMAbstractFactory.getOMFactory();
                    OMElement omElement = omFactory.createOMElement(new QName("http://wso2.org/ns/2010/10/bam", "BAMEvent", "ns"), null);
                    omElement.addAttribute("activityID", "AAAAIsiasd-sdswodi-2329", null);

                    OMElement bampropertyElement = omFactory.createOMElement(new QName("http://wso2.org/ns/2010/10/bam", "Property", "ns"), null);
                    bampropertyElement.addAttribute("name", "ESB2_a", null);
                    bampropertyElement.addAttribute("value", "A", null);

                    omElement.addChild(bampropertyElement);
                    omElement.addChild(omElement);

                    sender.addHeader(omElement);

                    sender.setOptions(options);
                    OMElement result1 = sender.sendReceive(payload1);
                    String response1 = result1.getFirstElement().getText();
                    System.out.println(response1);

                } catch (Exception e) { // (XMLStreamException e) {
                    System.out.println(e.toString());
                }
                finally {
                    if (sender != null) {
                        try {
                            sender.cleanupTransport();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

                        try {
                            sender.cleanup();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }

                httpConnectionManager.closeIdleConnections(0);
                httpConnectionManager.shutdown();
            }
        }
    }
}

2. After sending few messages you will be able to observe the messages from the BAM Admin Console > Main Dashboard > Actvity Tab