Sample 0107 - Using RDBMS event tables

Introduction

This sample demonstrates how to set up an execution plan to filter out credit card transactions that makes use of an in-memory event table to identify blacklisted transactions. This sample uses the Event Simulator for inputs and the logger publisher for logging the outputs to the DAS console.

The execution plan used in this sample are as follows:

@from(eventtable = 'rdbms' , datasource.name = 'WSO2_CARBON_DB' , table.name = 'CEPSample0107CardUserTable')
define table CardUserTable (name string, cardNum string, blacklisted bool) ;  

Given above is the table definition. It does the following:

• Defines a table named CardUserTable with the given attributes. 
• The annotation @from is used to link the table to an RDBMS event table named CEPSample0107CardUserTable and the data source used to access the table as WSO2_CARBON_DB (this is the datasource of the default H2 database that is shipped with WSO2 DAS. If needed you can define a separate data source pointing to some other DB such as MySQL and use it here.)  

from CardUserStream
select * 
insert into CardUserTable;

The first query,

• Processes the events received through the CardUserStream.
• Selects all the attributes under the select clause, from each event received. 
• Inserts it to the CardUserTable.

from BlackListStream
select cardNo as cardNum, true as blacklisted  
update CardUserTable
	on cardNum == CardUserTable.cardNum;

The second query,

• Processes the events received through the BlackListStream.
• Selects cardNo and renames it as cardNum, introduces a new attribute named blacklisted with the value true under the select clause, for each event received. 
• Updates the CardUserTable with the condition cardNum == CardUserTable.cardNum. Here the blacklisted attribute in the table is updated with the new value.

from PurchaseStream#window.length(1) as p join CardUserTable as c
	on  p.cardNo == c.cardNum and c.blacklisted == false
select p.cardNo as cardNo, c.name as name, p.price as price
insert into WhiteListPurchaseStream ;

The third query,

• Defines a length window that keeps 1 event of the input stream PurchaseStream.
• Joins it with the CardUserTable with the condition p.cardNo == c.cardNum and c.blacklisted == false. In this condition, the events with blacklisted == true in the table gets filtered out and then the remaining events are joined based on the card number.
• Emits those events as output events through the TransformedRoomTempStream.

from DeleteAllUsers 
delete CardUserTable
	on deleteAll == true;

The last query is used to clean up the table from an external trigger event through DeleteAllUsers stream,

• It processes the events received through the DeleteAllUsers.
• Checks for the condition deleteAll == true and if its true, deletes all the records in the CardUserTable. 

Prerequisites

Set up the prerequisites required for all samples.

Building the sample

Start the WSO2 DAS server with the sample configuration numbered 0107. For instructions, see Starting sample CEP configurations.

This sample configuration points the default Axis2 repo to <DAS_HOME>/samples/cep/artifacts/0107 (by default, the Axis2 repo is <DAS_HOME>/repository/deployment/server).

Executing the sample

1. Log into the DAS Management Console using the following URL and admin/admin credentials:  https://<DAS_HOST>:<DAS_PORT>/carbon.

2. Click Tools, and then click Event Simulator. You view four files listed under the Send multiple Events section that contains some sample data as follows. 

3. Click the corresponding Play button of the userEvents.csv file which contains sample data that is used to fill the in-memory CardUserTable. This starts sending events to fill the table.

4. Click the corresponding Play button of the blackListUserEvents.csv file which contains sample data that is used to mark user entries in the CardUserTable as blacklisted. This starts sending blacklisted events and marks some table entries as blacklisted.
5. Click the corresponding Play button of the purchaseEvents.csv file which contains credit card transactions data. This sends the transaction data. 

6. Click the corresponding Play button of the deleteUserEvents.csv file to clean up the tables.
   The output logs can be viewed in the CLI in which you run the products as shown below.