addEventListener

addEventListener(handler, [eventType], [condition], [times="all"], [at], [wait], [within], [exceedTime], [name])

Defines or dynamically adds event listeners within a monitor.

handler is a callback function to be invoked when the event listener triggers.

• If handler is triggered by each matched event (with eventType specified), it is a unary function that takes the matched event as its parameter.

• If handler is triggered by a scheduled time interval (without eventType), it is a function without any parameters.

def action1(stockVal){
    // insert attributes of event stockVal to a shared table
    insert into objByName("sharedUSPrices") values([stockVal.price, stockVal.qty])
}

You can add listeners dynamically in a handler. For example:

def action1(stockVal){
    insert into objByName("sharedUSPrices") values([stockVal.price, stockVal.qty])
    // add a listener
    addEventListener(handler=action2, eventType="Stock", condition=<self.price > Stock.price>)
}

eventType (optional) is a string scalar indicating the event to be monitored. If set to "any", the handler will be triggered for any event that matches the specified condition.

condition (optional) is metacode of Boolean expression to identify the event or pattern of events that you want to match. For example, when eventType is "Stock", condition can be <Stock.price > 10 and Stock.qty < 10>.

times (optional) is "all" or an integer that specifies the lifetime of an event listener by setting a limit on the number of times the handlercan be triggered. The default value is "all", which means the listener remains active within the CEP engine, continuously monitoring for events and invoking the handler until the engine is explicitly dropped.

at (optional) is a tuple of length 6 that specifies the scheduled time for triggering the handler. The tuple takes the following form: (seconds, minutes, hours, days_of_the_week, days_of_the_month, months). For example, at = (0, 5, , , , ) means the handler will be triggered at the 5th minute of every hour.

wait (optional) is of DURATION type, indicating the interval to trigger the handler. For example, wait = 60s means handler is triggered every 60 seconds.

within (optional) is a DURATION scalar that specifies the time interval during which the handler is triggered by matched events. When within is specified, timesmust be set to 1. For example, when within=60s,

• If one or more matching events occur within a 60-second interval, the handler is invoked for the first matching event and the event listener will be removed once the handler is triggered.

• If no matching events occur within this interval, the event listener is automatically removed after 60s.

exceedTime (optional) is a DURATION scalar that specifies the time interval during which the handler is triggered by the absence of matched events. If exceedTime is specified, times must be set to 1. For example, exceedTime=60s,

• If no matching events occur within a 60-second interval, the handler is invoked after 60s. The event listener will be removed once the handler is triggered.

• If one or more matching events occur within this interval, the listener is automatically removed.

name (optional) is a STRING scalar indicating the unique identifier of the listener. The default naming rules are as follows:

• If condition is specified, the default name is the string representation of the condition expression.
• If condition is not specified but eventType is specified, the default name iseventType.
• If both condition and eventType are not specified, the default name is "timer".
• If the generated name duplicates with an existing listener, a numeric suffix will be appended automatically to ensure uniqueness.

When defining a listener, do not mix parameters for different trigger conditions. For example, you should not specify an eventType and a wait time together.

Returns an EventListener instance.

class trades{
    trader :: STRING
    market :: STRING
    code :: STRING
    price :: DOUBLE
    qty :: INT
    eventTime :: TIMESTAMP
    def trades(t, m, c, p, q){
        trader = t
        market = m
        code = c
        price = p
        qty = q
        eventTime = now()
    }
}

class mainMonitor:CEPMonitor{
    tradesTable :: ANY
    isBusy :: BOOL
    def mainMonitor(){
        tradesTable = array(ANY, 0)
        isBusy = false
    }

    def updateTrades(event)

    def updateTrades2(event)

    def unOnload(){
        undef('traderDV', SHARED)
    }

    def onload(){
        addEventListener(updateTrades, `trades, , "all",,,,,"trades1")
        addEventListener(updateTrades2, `trades, , "all",,,,,"trades2")
        traderDV = streamTable(array(STRING, 0) as trader, array(STRING, 0) as market, array(SYMBOL, 0) as code, array(DOUBLE, 0) as price, array(INT, 0) as qty, array(INT, 0) as tradeCount, array(BOOL, 0) as busy, array(DATE, 0) as orderDate, array(TIMESTAMP, 0) as updateTime)
        share(traderDV, 'traderDV')
        createDataViewEngine('traderDV', objByName('traderDV'), `trader, `updateTime, true)
    }

    def updateTrades(event) {
        tradesTable.append!([event.trader, event.market, event.code, event.price, event.qty])
        getDataViewEngine('traderDV').append!(table(event.trader as trader, string() as market, string() as code, 0.0 as price, 0 as qty, 0 as tradeCount, false as busy, date(event.eventTime) as orderDate))
        updateDataViewItems('traderDV', event.trader, ['market', 'code', 'price', 'qty', 'tradeCount'], [event.market, event.code, event.price, event.qty, tradesTable.size()])
    }

    def updateTrades2(event) {
        tradesTable.append!([event.trader, event.market, event.code, event.price, event.qty])
        getDataViewEngine('traderDV').append!(table(event.trader as trader, string() as market, string() as code, 0.0 as price, 0 as qty, 0 as tradeCount, false as busy, date(event.eventTime) as orderDate))
        updateDataViewItems('traderDV', event.trader, ['market', 'code', 'price', 'qty', 'tradeCount'], [event.market, event.code, event.price, event.qty, tradesTable.size()])
    }
}

dummy = table(array(TIMESTAMP, 0) as eventTime, array(STRING, 0) as eventType, array(BLOB, 0) as blobs)
engineCep = createCEPEngine('cep1', <mainMonitor()>, dummy, [trades], 1, 'eventTime', 10000)
trade1 = trades('t1', 'sz', 's001', 11.0, 10)
go
appendEvent(engineCep, trade1)

monitors = getCEPEngineMonitor('cep1',"cep1","mainMonitor")

listeners = monitors.getEventListener()
print(listeners)
// Terminate listener
listeners['trades1'].terminate()
print(listeners)

Related functions: getEventListener, terminate