How Event stream Works in wso2cep3.0.0

Question

I am working with wso2cep3.0.0, My input source and out put source is JMs.I written my Input event adapter and output event adapter like this

Input adapter

is

<?xml version="1.0" encoding="UTF-8"?>
<inputEventAdaptor name="jmsProxy" statistics="disable" trace="enable"
  type="jms" xmlns="http://wso2.org/carbon/eventadaptormanager">
  <property name="java.naming.provider.url">tcp://localhost:61616</property>
  <property name="transport.jms.SubscriptionDurable">false</property>
  <property name="transport.jms.UserName">admin</property>
  <property name="java.naming.factory.initial">org.apache.activemq.jndi.ActiveMQInitialContextFactory</property>
  <property name="transport.jms.Password">admin</property>
  <property name="transport.jms.ConnectionFactoryJNDIName">QueueConnectionFactory</property>
  <property name="transport.jms.DestinationType">queue</property>
</inputEventAdaptor>

and my

Output event adapter

is

<?xml version="1.0" encoding="UTF-8"?>
<outputEventAdaptor name="OUTJmsProxy" statistics="disable" trace="disable"
  type="jms" xmlns="http://wso2.org/carbon/eventadaptormanager">
  <property name="java.naming.security.principal">admin</property>
  <property name="java.naming.provider.url">tcp://localhost:61616</property>
  <property name="java.naming.security.credentials">admin</property>
  <property name="java.naming.factory.initial">org.apache.activemq.jndi.ActiveMQInitialContextFactory</property>
  <property name="transport.jms.ConnectionFactoryJNDIName">QueueConnectionFactory</property>
  <property name="transport.jms.DestinationType">queue</property>
</outputEventAdaptor>

and my input message in jmsproxy jms queue is like this

<soapenv:Body xmlns:soapenv="http://www.w3.org/2003/05/soap-envelope">
   <uuid>cc253480-95b3-418e-b282-7e87f885c99e</uuid>
   <Remarks>t4</Remarks>
   <ReadingsLiteTaildtos>
      <ReadingsLiteTaildto>
         <FinalValue>70</FinalValue>
         <InputText>Chiller Feeder Current R - Ph</InputText>
         <InputValue>0.0</InputValue>
         <ParameterId>-2499999974</ParameterId>
         <SlNo>1</SlNo>
      </ReadingsLiteTaildto>
      <ReadingsLiteTaildto>
         <FinalValue>70</FinalValue>
         <InputText>Chiller Feeder Current Y - Ph</InputText>
         <InputValue>0.0</InputValue>
         <ParameterId>-2499999973</ParameterId>
         <SlNo>2</SlNo>
      </ReadingsLiteTaildto>
      <ReadingsLiteTaildto>
         <FinalValue>70</FinalValue>
         <InputText>Chiller Feeder Current B - Ph</InputText>
         <InputValue>0.0</InputValue>
         <ParameterId>-2499999972</ParameterId>
         <SlNo>3</SlNo>
      </ReadingsLiteTaildto>
      <ReadingsLiteTaildto>
         <FinalValue>70</FinalValue>
         <InputText>Chiller Energy Meter Reading</InputText>
         <InputValue>0.0</InputValue>
         <ParameterId>-2499999971</ParameterId>
         <SlNo>4</SlNo>
      </ReadingsLiteTaildto>
   </ReadingsLiteTaildtos>
   <ReadingDateTime>1381757157596</ReadingDateTime>
   <PartyBranchId>-2500000000</PartyBranchId>
   <ParametersetId>-2499999974</ParametersetId>
   <AssetId>-2499999995</AssetId>
   <TaskId>811291126760647</TaskId>
   <WorkOUId>-1</WorkOUId>
   <activityid>-2500000000</activityid>
   <userid>-2499999993</userid>
   <entrymode>0</entrymode>
   <DeviceId>-1</DeviceId>
</soapenv:Body>

i wish to raise an event when final value cross the max value like more than 100 so how would i write Stream and

ExecutionPlan

In stream-manger-config.xml file consist 3 section

1.metaData 2.Correlation Data 3.Payload Data

so above message how would i define which data is under which section one more we should define input payload and out payload as well in same stream config file else we need to define separate

Is cep help for this usecase or not

Thanx in Advance.

Answer 1

Yes, this is a typical usecase for CEP.

You can use an 'event builder' similar to following.

<?xml version="1.0" encoding="UTF-8"?>
<eventBuilder name="ReadingsDtoBuilder" statistics="disable"
    trace="disable" xmlns="http://wso2.org/carbon/eventbuilder">
<from eventAdaptorName="jmsEventReceiver" eventAdaptorType="jms">
    <property name="transport.jms.Destination">ReadingsQueue</property>
</from>
<mapping customMapping="disable"
    parentXpath="//ReadingsLiteTaildtos" type="xml">
    <property>
        <from xpath="//ReadingsLiteTaildto/ParameterId"/>
        <to name="meta_parameterId" type="string"/>
    </property>
    <property>
        <from xpath="//ReadingsLiteTaildto/Slno"/>
        <to name="meta_slno" type="string"/>
    </property>
    <property>
        <from xpath="//ReadingsLiteTaildto/FinalValue"/>
        <to name="finalValue" type="int"/>
    </property>
    <property>
        <from xpath="//ReadingsLiteTaildto/InputText"/>
        <to name="inputText" type="string"/>
    </property>
    <property>
        <from xpath="//ReadingsLiteTaildto/InputValue"/>
        <to name="inputValue" type="double"/>
    </property>
</mapping>
<to streamName="org.sample.readings.dto.stream" version="1.0.0"/>
</eventBuilder>

The execution plan can be as follows.

<?xml version="1.0" encoding="UTF-8"?>
<executionPlan name="ReadingsAnalyzer" statistics="disable"
  trace="disable" xmlns="http://wso2.org/carbon/eventprocessor">
  <description>This execution plan analyzes readings and triggers notifications based on     threshold.</description>
  <siddhiConfiguration>
    <property name="siddhi.enable.distributed.processing">false</property>
    <property name="siddhi.persistence.snapshot.time.interval.minutes">0</property>
  </siddhiConfiguration>
  <importedStreams>
    <stream as="readings" name="org.sample.readings.dto.stream" version="1.0.0"/>
  </importedStreams>
  <queryExpressions><![CDATA[from readings[finalValue > 100]
select *
insert into notificationStream;]]></queryExpressions>
  <exportedStreams>
    <stream name="notificationStream" valueOf="notificationStream" version="1.0.0"/>
  </exportedStreams>
</executionPlan>

You can define streams inside stream-manager-config.xml similar to the following.

<streamDefinition name="org.sample.readings.dto.stream" version="1.0.0">
<metaData>
        <property name="parameterId" type="STRING"/>
        <property name="slno" type="STRING"/>
</metaData>
    <payloadData>
        <property name="finalValue" type="INT"/>
        <property name="inputText" type="STRING"/>
        <property name="inputValue" type="DOUBLE"/>
    </payloadData>
</streamDefinition>
<streamDefinition name="notificationStream" version="1.0.0">
<metaData>
        <property name="parameterId" type="STRING"/>
        <property name="slno" type="STRING"/>
</metaData>
    <payloadData>
        <property name="finalValue" type="INT"/>
        <property name="inputText" type="STRING"/>
        <property name="inputValue" type="DOUBLE"/>
    </payloadData>
</streamDefinition>