Extending Siddhi 3.0

Siddhi[1] 3.0.0 is released now with lots of new features and improved performance. Some of the main features supported by siddhi are,

• Filter 
• Join 
• Aggregation 
• Stream handlers (e.g:- WIndows) 
• Pattern processing 
• Sequence processing 
• Event Tables

In addition to above, Siddhi facilitates writing various types of extensions via its plugable architecture. Siddhi is shipped with in-built set of extensions including,

• math
• str
• geo
• r
• regex
• time
• nlp
• pmml 
• ml

Not only these extension we can write our own 

1. aggregators
2. functions
3. windows
4. stream functions 
5. stream processors 

easily which can be used with siddhi queries. It's simply a matter of extending a siddhi interface and adding the mapping file to map the extension class with its function name and namespace.

Check my next blog posts for detailed info on writing extensions for each of the five types.

• http://sachinij.blogspot.com/2015/10/writing-custom-function-in-siddhi-30.html

You can read more about extensions from [2] and [3].

[1]https://github.com/wso2/siddhi

[2]https://docs.wso2.com/display/CEP400/SiddhiQL+Guide+3.0#SiddhiQLGuide3.0-SiddhiExtensions
[3]https://docs.wso2.com/display/CEP400/Writing+Extensions+to+Siddhi

Using sonar to analyze code quality of maven projects

• Download sonar

• Start SonarQube: In my case I started sonaQube by running ./sonar.sh start from the directory sonarqube-4.4/bin/linux-x86-64

• Go to your project directory and run the command mvn sonar:sonar

• You can view the results from http://localhost:9000/

WSO2 ESB - Enrich Mediator

The Enrich Mediator can process a message based on a given source configuration and then perform the specified action on the message by using the target configuration. It is often used to do slight modifications to the payload.

This post explains some scenarios where Enrich mediator can be used.

Original payload

<orders>
<order>
<price>50.00</price>
<quantity>500</quantity>
<symbol>IBM</symbol>
<comment>REF 10053</comment>
</order>
<order>
<price>18.00</price>
<quantity>500</quantity>
<symbol>MSFT</symbol>
<comment>ref 20088398289</comment>
</order>
</orders>

Required payload-1

<orders>
<order>
<price>50.00</price>
<quantity>500</quantity>
<symbol>IBM</symbol>
<comment>REF 10053</comment>
</order>
<order>
<price>18.00</price>
<quantity>500</quantity>
<symbol>MSFT</symbol>
<comment>ref 20088398289</comment>
</order>
</orders>
<orderID>2</OrderID>

mediator

<property name="orderID" scope="default" description="orderID">
            <orderID xmlns="">2</orderID>
 </property>
  <enrich>
            <source clone="true" xpath="$ctx:orderID"/>
            <target action="sibling" xpath="//orders"/>
  </enrich>

Define a property <orderID>2</orderID>
Then add the property as a sibling of <orders>

Required payload-2

<orders>
<order>
<price>50.00</price>
<quantity>500</quantity>
<symbol>IBM</symbol>
<comment>REF 10053</comment>
</order>
<order>
<price>18.00</price>
<quantity>500</quantity>
<symbol>MSFT</symbol>
<comment>ref 20088398289</comment>
</order>
<orderID>2</OrderID>
</orders>

mediator

<property name="orderID" scope="default" description="orderID">
            <orderID xmlns="">2</orderID>
 </property>
   <enrich>
            <source clone="true" xpath="$ctx:orderID"/>
            <target action="child" xpath="//orders"/>
         </enrich>

Define a property <orderID>2</orderID>
Then add the property as a child of <orders>


Required payload-3

<orders>
<order>
<price>50.00</price>
<quantity>500</quantity>
<symbol>IBM</symbol>
<comment>REF 10053</comment>
<volume>100<volume>
</order>
<order>
<price>18.00</price>
<quantity>500</quantity>
<symbol>MSFT</symbol>
<comment>ref 20088398289</comment>
</order>
<orderID>2</OrderID>
</orders>

mediator

<property name="VOLUME" value="100"/>
 <enrich>
           <source type="inline" clone="true">
                   <volume xmlns=""/>
            </source>
            <target action="child" xpath="//orders/order"/>
 </enrich>
 <enrich>
            <source type="property" clone="true" property="VOLUME"/>
            <target xpath="//orders/order/volume"/>
 </enrich>

Using first enrich mediator we add the element <volume>  as a child of <order>
Second enrich mediator adds the value of “VOLUME” property to the newly created element


Required payload-4

<orders>
<order>
<price>50.00</price>
<quantity>500</quantity>
<symbol>IBM</symbol>
<comment>REF 10053</comment>
<volume>100<volume>
</order>
<order>
<price>18.00</price>
<quantity>500</quantity>
<symbol>MSFT</symbol>
<volume>100</volume>
<comment>ref 20088398289</comment>
</order>
<orderID>2</OrderID>
</orders>

mediator

 <property name="VOLUME" value="100"/>
 <enrich>
            <source type="inline" clone="true">
               <volume xmlns=""/>
            </source>
            <target action="child" xpath="//orders/order[symbol/text() = 'MSFT']"/>
  </enrich>
  <enrich>
            <source type="property" clone="true" property="VOLUME"/>
            <target xpath="//orders/order[symbol/text() = 'MSFT']/volume"/>
  </enrich>


Instead of comparing 'MSFT' twice,  we can use a property to store the value 'MSFT' and then use it for comparisons as below.
 <property name="VOLUME" value="100"/>
         <property name="SYMBOL" value="MSFT"/>
         <enrich>
            <source type="inline" clone="true">
               <volume xmlns=""/>
            </source>
            <target action="child"
                    xpath="//orders/order[symbol/text() = get-property('SYMBOL')]"/>
         </enrich>
         <enrich>
            <source type="property" clone="true" property="VOLUME"/>
            <target xpath="//orders/order[symbol/text() = get-property('SYMBOL')]/volume"/>
         </enrich>

Writing Siddhi Extensions

WSO2 CEP is a lightweight, easy-to-use, open source Complex Event Processing server. Back-end runtime engine of WSO2 CEP is Siddhi. More info on siddhi can be found at[1].

This post explains how to write extensions to siddhi.

An extension allows you to add and use your own custom logic within siddhi queries. Latest released siddhi version supports writing extension to Windows, Transformers, OutputAttributeProcessors and Functions.

I will explain how to write a custom OutputAttributeProcessor with an example.

Writing custom output aggregator

We'll write an output aggregator which will give the unique count of a given attribute. (Siddhi supports count as an inbuillt output aggregator , we'll improve it to get a unique count)

To write an output aggregator,

• write a class implementing org.wso2.siddhi.core.query.selector.attribute.factory.OutputAttributeAggregatorFactory 

• write a class extending org.wso2.siddhi.core.query.selector.attribute.handler.OutputAttributeAggregator. 

Step 1:

We have to give a namespace and a function name to each extension we write. This have to be included in the class which implements OutputAttributeAggregatorFactory. When implemeting this class, we'll have to override the method createAttributeAggregator. Her what we have to do is return a new instance of the class we created by extending the class OutputAttributeAggregator. (Refer to step 2)

Let's give namespace as “test” and function as “getUniqueCount” for our custom aggregator. Below is our class,

import org.wso2.siddhi.core.query.selector.attribute.factory.OutputAttributeAggregatorFactory; 
import org.wso2.siddhi.core.query.selector.attribute.handler.OutputAttributeAggregator; 
import org.wso2.siddhi.query.api.definition.Attribute; 
import org.wso2.siddhi.query.api.extension.annotation.SiddhiExtension; 


    @SiddhiExtension(namespace = "test", function = "getUniqueCount") 
    public class UniqueCountAggregatorFactory implements OutputAttributeAggregatorFactory {
 
        @Override 
        public OutputAttributeAggregator createAttributeAggregator(Attribute.Type[] types) { 
            return new UniqueCountAggregatorInt(); 
        } 
    }

Step 2:

Class which extends OutputAttributeAggregator will have the actual logic. When writing this calls we have to implement several methods.

• GetReturnType() : return type of the aggregator. Can be Int, String, Long, Bool, Double, Float 
• processAdd(Object obj) 
• processRemove(Object obj) 
• newInstance() 
• destroy() : here we can do the cleanup tasks clode connection if we have cretaed any, etc 

import org.wso2.siddhi.core.query.selector.attribute.handler.OutputAttributeAggregator; 
import org.wso2.siddhi.query.api.definition.Attribute; 

import java.util.ArrayList; 
import java.util.List; 


public class UniqueCountAggregatorInt implements OutputAttributeAggregator { 
    private static final long serialVersionUID = 1358997438272544590L; 
    private int uniqueCount = 0; 
    private List<Object> valuesCounted = new ArrayList<Object>(); 

    @Override 
    public Attribute.Type getReturnType() { 
        return Attribute.Type.INT; 
    } 


    @Override 
    public Object processAdd(Object obj) { 
        if (!valuesCounted.contains(obj)) { 
            valuesCounted.add(obj); 
            uniqueCount++; 
        } 
        return uniqueCount; 
    } 


    @Override 
    public Object processRemove(Object obj) { 
        valuesCounted.remove(obj); 
        uniqueCount--; 

        return uniqueCount; 
    } 


    @Override 
    public OutputAttributeAggregator newInstance() { 
        return new UniqueCountAggregatorInt(); 
    } 


    @Override 
    public void destroy() { 
    } 
}

This is all you have to do to create a custom output aggregator. You can refer [2] for more details.
We can use this extension on siddhi query as follows,

         from StockExchangeStream#window.time(1 hour)
         select symbol,test:getUniqueCount(price) as uniquePriceCount
         group by symbol
         insert into outputStream;

If you want to use the implemented output aggregator extension in WSO2 CEP you haveto follow below steps,

• compile the created two classes 
• add the jar files to CEP_HOME/repository/components/lib 
• add the fully-qualified class name of class which impelemtnts OutputAttributeAggregatorFactory in a new line to the siddhi.extension file which can be found at <CEP_HOME>/repository/conf/siddhi 

[1]http://srinathsview.blogspot.com/2011/12/siddhi-second-look-at-complex-event.html
[2]https://docs.wso2.com/display/CEP310/Writing+Extensions+to+Siddhi