Notifications

Notifications in BAM can be done in two ways:-

• CEP based Notifications
• Hive Script based Notifications

CEP based Notifications

CEP based notifications are implemented using a CEP flow to define an input source of a message by creating an event stream, then a CEP execution plan can be implemented using Siddhi if necessary, and the result can be written off to a output event adapters such as Email and SMS, via a message formatter. For more information on how to process events using CEP, go to CEP Configuration Overview. 

Hive Script based Notifications

Hive script based notifications are used when a notification needs to be sent out directly from a Hive script execution. This is achieved by inserting data records to a well known Cassandra column family from the Hive script, where, a background task will be polling this specific column family to check if there are any new records, and if so, it will load those records, and dispatch them to a stream mentioned by each of the records, and at the end, it will delete the processed records from the column family.

The name of the Cassandra column family is "bam_notification_messages", inside the Cassandra data source with the name "WSO2BAM_UTIL_DATASOURCE". For each of the row inserted for "bam_notification_messages" column family, it should contain a column with the name "streamId", which identifies the target stream id (i.e. "stream_name:version") of the stream this record should be written to. The columns other than with the name "streamId" basically is considered payload properties of the target stream, where the column name is the property name and the column value is the property value.

The schedule task used to poll the notifications column family will send the data to the mentioned streams, where those stream can contain a CEP flow, which can define a notification target for the messages to be delivered.

Below contains a sample Hive script snippet, which demonstrates the Hive based notifications.

CREATE EXTERNAL TABLE IF NOT EXISTS BAMNotifications 
    (id STRING, streamId STRING, country STRING,  requestCount STRING) 
    STORED BY 'org.apache.hadoop.hive.cassandra.CassandraStorageHandler'
    WITH SERDEPROPERTIES ( 
    "wso2.carbon.datasource.name" = "WSO2BAM_UTIL_DATASOURCE" ,
    "cassandra.cf.name" = "bam_notification_messages" , 
    "cassandra.columns.mapping" = ":key,streamId,country,requestCount" );    
    
insert into table BAMNotifications select countryCode, "country_notifications:1.0.0", country, requestCount from CountrySummary where requestCount > 600;

The above snippet can be used with the HTTPD log analyzing sample Hive script.

The Hive script based notifications functionality can be switched off in a specific BAM node by setting the Java system property "disable.notification.task" (i.e. sh wso2server.sh -Ddisable.notification.task).