Processing large text file using smooks in wso2esb

Smooks  library supports large file processing which is used in ESB to process file which are in text/xml format.
Text files can be a CSV file or with a space delimiter or a fixed length file. To process large file, smooks write the output stream , which is a single line record, to a temporary file or a jms queue. After writing all files to the temporary location, we need to process all those records/files again. This is how we can avoid large memory growth/out of heap space issue which occurs when we try to process large file as in- memory process.

Following is a sample configuration to process fixed length large text file in wso2esb.

Proxy configuration

To pick and process the file form a file location , we could use VFS transport.

<proxy xmlns="http://ws.apache.org/ns/synapse"
       name="cor_file_pull"
       transports="https,http,vfs"
       statistics="disable"
       trace="disable"
       startOnLoad="true"target>
     <inSequence>
  <property name="DISABLE_SMOOKS_RESULT_PAYLOAD"
                   value="true"
                   scope="default"
                   type="STRING"/>
          <smooks config-key="simple-smook">
            <input type="text"/>
           <output type="xml"/>
         </smooks>
       </inSequence>
      <outSequence/>
   </target>
   <parameter name="transport.vfs.Streaming">true&lt;/parameter>
   <parameter name="transport.PollInterval">10&lt;/parameter>
   <parameter name="transport.vfs.ActionAfterProcess">MOVE&lt;/parameter>
   <parameter name="transport.vfs.MoveAfterProcess">C:/Users/TOSH/Desktop/success</parameter>
   <parameter name="transport.vfs.FileURI">C:/Users/TOSH/Desktop/in&lt;/parameter>
   <parameter name="transport.vfs.MoveAfterFailure">C:/Users/TOSH/Desktop/failure</parameter>
   <parameter name="transport.vfs.Locking">disable&lt;/parameter>
   <parameter name="transport.vfs.FileNamePattern">^(cor).+</parameter>
   <parameter name="transport.vfs.ContentType">text/plain</parameter>
   <parameter name="transport.vfs.ActionAfterFailure">MOVE</parameter>
   <description/>
</proxy>

Smooks configuration

simple-smook localentry

<smooks-resource-list xmlns="http://www.milyn.org/xsd/smooks-1.1.xsd" xmlns:ftl="http://www.milyn.org/xsd/smooks/freemarker-1.1.xsd" xmlns:fl="http://www.milyn.org/xsd/smooks/fixed-length-1.3.xsd" xmlns:file="http://www.milyn.org/xsd/smooks/file-routing-1.1.xsd">
<params>
            <param name="stream.filter.type">SAX </param>
           <param name="default.serialization.on">false </param>
         </params>

<fl:reader fields="RecordId[4],CorpName[60],FileNumber[9]" skipLines="0">
</fl:reader>
      <resource-config selector="record">
         <resource> org.milyn.delivery.DomModelCreator</resource>
     </resource-config>
      <file:outputStream openOnElement="record" resourceName="fileSplitStream">
         <file:fileNamePattern> ${.vars["record"].FileNumber}.xml</file:fileNamePattern>
         <file:destinationDirectoryPattern> C:\Users\TOSH\Desktop\processed</file:destinationDirectoryPattern>
         <file:highWaterMark mark="10000000"> </file:highWaterMark>
      </file:outputStream>
      <ftl:freemarker applyOnElement="record">
         <ftl:template> /repository/resources/smooks/record_as_xml.ftl</ftl:template>
         <ftl:use>
            <ftl:outputTo outputStreamResource="fileSplitStream"> </ftl:outputTo>
         </ftl:use>
      </ftl:freemarker>
   </smooks-resource-list>

Each record will   be written to the specified file location   "C:\Users\TOSH\Desktop\processed" with the name constructed through the parameter "  ${.vars["record"].FileNumber}.xml ".  Here it is a xml file, which will be named with a file number    record in the text file. We have to select a unique parameter to write the processed files else there will be file overwriting issue will be thrown in smooks.
To process large file we have to use 'SAX' filter type. DOM needs more memory and it will lead to out of memory issue.
In this sample, i have used DOM+SAX mixing model to keep the lower footprint of memory.
High  water mark provides the maximum file records need to be written. provide enough number based on the lines in the text file.
The property ;
 <property name="DISABLE_SMOOKS_RESULT_PAYLOAD"   value="true" scope="default"     type="STRING"/>

will be available in ESB 4.9.0. If you use lower version, you might need a patch to be applied to your system.

XML template file
record_as_xml.ftl

<#assign rec = .vars["record"]>
<p:insert_acc_data_file xmlns:p="http://ws.wso2.org/dataservice"><p:RecordId>${rec.RecordId}</p:RecordId>
<p:CorpName>${rec.CorpName}&lt;/p:CorpName>
<p:FileNumber>${rec.FileNumber}&lt;/p:FileNumber></p:insert_acc_data_file>

Based on above template the records will be formatted and will be written to the .

Using another vfs proxy, we can pick the entries written to the  'processed' file and do further processing.

Executing FaultSequence for SOAPFaults in WSO2ESB

In WSO2ESB/Synapse, whenever mediation errors occur, user can execute faultsequence to find error details.
The fault sequence can be defined at a sequence level or at proxy service level. Mediation level issues and endpoint errors can be captured at faultsequence. But SOAPFault can not be captured at fault sequence, since ESB/Synapse will treat it as the response from the endpoint.
But from user's pont of view , those are service endpoint failures which need to be handled properly.
To handle the soapfaults, now we can use a property which forces to execute fault sequence whenever soapfault occurs.
We need to set  following property before sending the message to endpoint.;

<property name="FORCE_ERROR_ON_SOAP_FAULT" value="true"/>

Sample conf;

<proxy xmlns="http://ws.apache.org/ns/synapse"
       name="testProxy"
       transports="https,http"
       statistics="disable"
       trace="disable"
       startOnLoad="true">
  <target faultSequence="myfaultSeq">
      <inSequence>
         <property name="FORCE_ERROR_ON_SOAP_FAULT" value="true"/>
      </inSequence>
      <outSequence>
         <send/>
      </outSequence>
      <endpoint>
         <address uri="http://localhost:9764/services/echo/"/>
     </endpoint>
   </target>
   <description/>
</proxy>
           

In the above sample , whenever we get soapfault, the 'myfaultSeq' will be executed.

Related Post: Sending mails for errors

Securevault support for credientials used in mediation configuration

Secure vault tool which  is available in carbon servers, helps to encrypt passwords which are configured in plain text in the configuration files. This tool can be directly used for the passwords which are configured in carbon.xml,  axis2.xml, datasources.xml etc..We can not use secure vault tool directly in the mediation configuration , where we might provide passwords to make secured connection  for any basic auth protected endpoints.

In the older versions of ESB/APIM, we provide basic auth header by setting the "Authorization" transport header,where password was in plain text.

Eg: 

&lt;property name="Authorization" expression="fn:concat('Basic ', base64Encode('<Username>:<Password>'))" scope="transport"/>

In the newer versions (AM 1.6.0 / ESB 4.8.0 onwards), user can store the encrypted password in registry and  the 'alias' can be used to refer the stored passwords.

Eg:

<property name="secPassword" expression="wso2:vault-lookup('securedDS.endpoint.password')"/>

• securedDS.endpoint.password : - Password alias, which is stored in config registry under /repository/components/secure-vault

• wso2:vault-lookup :- Is a custom xpath implementation for synapse to do lookups. To use this custom xpath, user needs to add following synapse property in the synapse.properties file.
  

synapse.xpath.func.extensions=org.wso2.carbon.mediation.security.vault.xpath.SecureVaultLookupXPathFunctionProvider 

In APIManager;

•  If user needs to encrypt the password in the mediation configuration, he has to enable the <EnableSecureVault> property in the api-manager.xml, and needs to run the cipher tool before running the product. 
• Cipher tool is available in the  /bin folder.
• # ciphertool.bat/sh -Dconfigure

• When user enables securevault option and  publishes an API via the publisher UI, created API configuration will have the securevault enabled password protection. User can check the alias configured in the API configuration and can map that with the configuration registry entry. Alias is created with the string merge of  publisherName+apiName+version;

 In ESB, user can use management console to create aliases.

Axiom.soap.SOAPProcessingException: First Element must contain the local name, Envelope.

I faced this issue [1] and struggled a bit to find the root cause of this issue in synapse. The error was not clear enough to find out what was the issue.
The fix was, we need to set  the 'SOAPAction' property before sending out the message.

<property name="SOAPAction" value="urn:select_address " scope="transport"/>

Axis2 dispatches the message based on ;

• Request URI based
• Request URI Operation based
• SOAP Action based
• SOAP MessageBody based etc..

If system could not find any of the option to dispatch the request to correct endpoint, axis2 will throw an error states "operation not found/ EPR not found error"  .
But here, the error stack is not descriptive  to find the actual missing configuration.

[1] ERROR - RelayUtils Error while building Passthrough stream
org.apache.axiom.soap.SOAPProcessingException: First Element must contain the local name, Envelope , but found faultstring
        at org.apache.axiom.soap.impl.builder.StAXSOAPModelBuilder.constructNode(StAXSOAPModelBuilder.java:305)
        at org.apache.axiom.soap.impl.builder.StAXSOAPModelBuilder.createOMElement(StAXSOAPModelBuilder.java:252)
        at org.apache.axiom.soap.impl.builder.StAXSOAPModelBuilder.createNextOMElement(StAXSOAPModelBuilder.java:234)
        at org.apache.axiom.om.impl.builder.StAXOMBuilder.next(StAXOMBuilder.java:249)
        at org.apache.axiom.soap.impl.builder.StAXSOAPModelBuilder.getSOAPEnvelope(StAXSOAPModelBuilder.java:204)
        at org.apache.axiom.soap.impl.builder.StAXSOAPModelBuilder.(StAXSOAPModelBuilder.java:154)
        at org.apache.axiom.om.impl.AbstractOMMetaFactory.createStAXSOAPModelBuilder(AbstractOMMetaFactory.java:73)
        at org.apache.axiom.om.impl.AbstractOMMetaFactory.createSOAPModelBuilder(AbstractOMMetaFactory.java:79)
        at org.apache.axiom.om.OMXMLBuilderFactory.createSOAPModelBuilder(OMXMLBuilderFactory.java:196)
        at org.apache.axis2.builder.SOAPBuilder.processDocument(SOAPBuilder.java:55)
        at org.apache.synapse.transport.passthru.util.DeferredMessageBuilder.getDocument(DeferredMessageBuilder.java:118)
        at org.apache.synapse.transport.passthru.util.RelayUtils.builldMessage(RelayUtils.java:107)
        at org.apache.synapse.transport.passthru.util.RelayUtils.buildMessage(RelayUtils.java:82)
        at org.apache.synapse.mediators.AbstractListMediator.mediate(AbstractListMediator.java:68)

Sending mails using mail transport in wso2esb

Scenario:

If any error occurs in a message mediation flow, Administrator has to send mail to manager.

Axis configuration

To send mails, we use mail transport. Enable the 'MailTransportSender' to send emails in specific scenarios.

@axis2.xml;

<transportSender name="mailto" class="org.apache.axis2.transport.mail.MailTransportSender">
        <parameter name="mail.smtp.host">smtp.gmail.com</parameter>
       <parameter name="mail.smtp.port">587</parameter>
       <parameter name="mail.smtp.starttls.enable">true</parameter>
       <parameter name="mail.smtp.auth">true</parameter>
       <parameter name="mail.smtp.user">admin@abc.com</parameter>
        <parameter name="mail.smtp.password">XXXXX</parameter>
       <parameter name="mail.smtp.from">Admin</parameter>
  </transportSender>

Proxy configuration

 <proxy xmlns="http://ws.apache.org/ns/synapse"
       name="FileProxy"
       transports="https,http,vfs"
       statistics="disable"
       trace="disable"
       startOnLoad="true">
    <target faultSequence="sendErrMailSequence">

       <inSequence>
.................
</inSEquence>
</proxy>

To get  a mail content in HTML format, we need to add HTML message formatter in axis2.xml
<messageFormatter contentType="text/html" class="org.apache.axis2.transport.http.ApplicationXMLFormatter"/>


SendErrMailSequence  configuration

<sequence xmlns="http://ws.apache.org/ns/synapse" name="sendErrMailSequence">
   <property name="messageType" value="text/html" scope="axis2"/>
   <property name="ContentType" value="text/html" scope="axis2"/>
  <property name="Subject" value=" Error occured" scope="transport"/>
   <property name="OUT_ONLY" value="true"/>
 <payloadFactory media-type="xml">
            <format>
               <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
                  <soapenv:Body>
                     <ErrorMessage xmlns=""> Please check endpoint errors</ErrorMessage>
                  </soapenv:Body>
               </soapenv:Envelope>
           </format>
         </payloadFactory>
   <send>
      <endpoint name="MailEpr">
         <address uri="mailto:manager@abc.com"/>
      </endpoint>
   </send>

</sequence>

XML to Text transformation using smooks mediator

Smooks mediator can be used to apply transformations. Inputs can be provided as text/xml and smooks transformation resource has to be passed as an external resource.
Here is the sample smooks mediator configuration and smooks configuration which can be used to read an xml element from the input and parse it to a preferred output xml format.

Smooks meditaor config;

<smooks config-key="smooks">
               <input type="xml" expression="//csv/arg0"/>
               <output type="text"/>
            </smooks>

Sample request;

<soapenv:Envelope xmlns:soapenv=http://schemas.xmlsoap.org/soap/envelope/ >
 <soapenv:Body>
  <csv>
   <arg>1000,20130926,UAE</arg0>
  </csv>
 </soapenv:Body>
</soapenv:Envelope>

Smooks configuration;

<smooks-resource-list xmlns="http://www.milyn.org/xsd/smooks-1.0.xsd">
   <resource-config selector="org.xml.sax.driver">
    <resource>org.milyn.csv.CSVParser</resource>
    <param name="fields" type="string-list">RecordID,FileNumber,CorporationName</param>
  </resource-config>
</smooks-resource-list>

Output;

<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
<soapenv:Body>
<ax:text xmlns:ax="http://ws.apache.org/commons/ns/payload">
<csv-set>
<csv-record number="1">
<RecordID>1000</RecordID>
<FileNumber>00238374</FileNumber>
<CorporationName>UAE</CorporationName>
</csv-record>
</csv-set>
</ax:text>
</soapenv:Body>
</soapenv:Envelope>

Accessing various information of a message in a mediation flow in WSO2ESB

Property mediator helps us to access various information of a message ,which passes through ESB. Within a sequence , we could use property mediator to get info about a message.
We can assign a value to a property as a string value or using xpath expression. Two actions are defined for property mediator.

• Set
• Remove

eg:

 <property name="MESSAGE" value="Executing in sequence"/>
 <property xmlns:m="http://sample.com"   name="test"   expression="//m:getQuote/m:symbol"
                   scope="default"  type="STRING"/>

We can use supported xpath functions,such as mathematical/logical operations and other predefined functions can be used to assign values in the property mediator.

eg: 

<property xmlns:ns="http://org.apache.synapse/xsd"   name="testProp"
                expression="fn:concat('hello','world')"   scope="default"/>

We can use  built in get-property() xpath function to read properties,which are available in the synapse messagecontext.
Eg: If a property "SYSTEM_TIME" available in the synapse message context; 

<property xmlns:ns="http://org.apache.synapse/xsd"   name="systemtime"
                expression="get-property('SYATEM_TIME')"   scope="default"/>

We can use property mediator to set values in different scopes, which can be accessed in those specific scope. There are 4 scopes,which are;

• synapse/default : This property will be available in both insequence and outsequence.
• axis2 : This property will be available in the particular sequence. can be retrieved from Axsi2Messagecontext.
• transport:  This scope is used to set a property in the transport header.
• axis2-client: Like "axis2" scope, but can be  retrieved form the Axis2Messagecontext.options.

eg:

<property name="myProp" value="abc" scope="default" type="STRING"/>
<property name="myProp" value="abc" scope="transport" type="STRING"/>
<property name="myProp" value="abc" scope="axis2" type="STRING"/> 
<property name="myProp" value="abc" scope="axis2-client" type="STRING"/>

When we retrieve properties from a specific scope;
eg:

 <property xmlns:ns="http://org.apache.synapse/xsd"    name="remotehost"
                    expression="get-property('axis2', 'REMOTE_HOST')"/>

There are some predefined xpath properties , which we can use directly to retrieve some common properties.

• $axis2: Used to get the property at the axis2 scope.

 eg :
 <property name="MessageType is" expression="$axis2:messageType"/>

• $trp: Used to get transport header properties.

eg:
 <property name="axis2Prop" expression="$trp:Content-Type"/>

• $ctx : Used to get synapse scope properties;

eg:
 If we have a property named as "myProp" in synapse scope, which can be retrieved in the following manner;
 <property name="testProp" expression="$ctx:myProp"/>

• $url : Used to retrieve parameters defined in the url

  eg:
For a request url; http:// localhost:8280/getSimpleQuote?symbol=IBM

<property name="symbol" expression="$url:symbol"/>
The above will retrieve the symbol value; ie:IBM

Accessing registry resources from class mediator in wso2esb

In the mediation process, users might need to access the registry resources, which are stored in the governance/config registries in wso2esb.
In the normal sequence flow, we could point the resources as registry keys in the mediators. But if we want to access them in a class mediator, where we may need to do custom process on the resource, we could retrieve the "WSO2Registry" instance from the  "SynapseConfiguration" and from the registry instance we could access different resources which are stored in the different registries.

Likewise if we have a 'LocalEntry' defined in the synapse configuration we could retrieve it from the "LocalRegistry". LocalEntries  are elements, used to declare registry entries that are local to the ESB instance. They are static resources and has high priority than a resource with the same name  hosted in embedded wso2registry.

Here is a sample class mediator, which can be used to retrieve a LocalEntry and resources hosted in governance/config registry.

import org.apache.synapse.MessageContext;
import org.apache.synapse.config.Entry;
import org.apache.synapse.mediators.AbstractMediator;
import org.apache.synapse.registry.Registry;

public class CustomRegistryRetriveMediator extends AbstractMediator {
   
    public boolean mediate(MessageContext synapseMsgContext) {

        // if localentry "testLocalentry" defined in synapse econfig
        Entry localEntryObj = (Entry) synapseMsgContext.getConfiguration()
                .getLocalRegistry().get("testLocalentry");
        System.out.println(" value: " + localEntryObj.getValue());
        System.out.println(" Filename : " + localEntryObj.getFileName());

        // goverance/config registry
        Registry regInstance = synapseMsgContext.getConfiguration()
                .getRegistry();                       
        Object obj=regInstance.getResource(new Entry("gov:/test1.txt"),null);       
        Object obj2=regInstance.getResource(new Entry("conf:/test2.txt"),null);   
       
        return true;
    }
}

Writing a custom class mediator in wso2esb

In this post we will look at how to write a very simple class mediator for wso2esb. All what you have to do is, just extend the AbstractMediator class (org.apache.synapse.mediators.AbstractMediator) or implement the Mediator interface (org.apache.synapse.Mediator). AbstractMediator contains the implementation for some methods , so it would be easy to use that class.
Say, in a scenario, when exception occurs  at ESB, we want to extract the error details , and need to do further processing. In the following  sample,  error details are extracted and printed as sysout..

package org.wso2.carbon.custommediator
import org.apache.synapse.MessageContext;
import org.apache.synapse.core.axis2.Axis2MessageContext;
import org.apache.synapse.mediators.AbstractMediator;

/**
* This custom mediator extracts error detail from synapse messagecontext.
* SynapseConfiguration would be;
*<insequence>
*   <class name="org.wso2.carbon.custommediator.CustomErrorDataMediator">
*   </class>
*</insequence>
**/

public class CustomErrorDataMediator extends AbstractMediator {

public boolean mediate(MessageContext synapseMsgContext) {
  String errorMesssage = (String) synapseMsgContext.getProperty("ERROR_MESSAGE");
  Exception exception = (Exception)  synapseMsgContext.              getProperty("ERROR_EXCEPTION");
  String errorCode = (String) synapseMsgContext.getProperty("ERROR_CODE");
  String errorDetail = (String) synapseMsgContext.getProperty("ERROR_DETAIL");

  System.out.println("Error Message is "+errorMesssage +"ErrorCode "+errorCode);

  return true;
 }
}

To build above code (*.jar as output), use following pom.xml file..

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">


    <modelVersion>4.0.0</modelVersion>
    <groupId> org.wso2.carbon.custommediator</groupId>
    <artifactId> org.wso2.carbon.custommediator</artifactId>
    <version>1.0</version>
    <packaging>jar</packaging>
 
    <dependencies>
        <dependency>
            <groupId>org.apache.synapse</groupId>
            <artifactId>synapse-core</artifactId>
            <version>1.4.0-wso2v2</version>       
        </dependency>      
    </dependencies>

    <build>
        <plugins>  
          <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>2.0</version>
                 <configuration>
                     <source>1.5</source>
                     <target>1.5</target>
                 </configuration>
             </plugin>
        </plugins>
    </build>

</project>

Keep the *.jar in ESB/repository/component/lib folder and place the class mediator in your faultsequence.Whenever fault sequence get invoke, you will see the sysout at your server console.