action

action action_name([arglist])returns retType{
do_something>;
}
 
action notifyUser(){
log "Event sequence detected.";
}

aggregate

aggregate [bounded|unbounded] aggregateName (
[arglist]) returns retType {
aggregateBody }
 
aggregate bounded wstddev( decimal x, decimal w )
returns decimal {
    do something}

all

all event_template
 
on all StockTick(*,*):newTick processTick();

all event_template_with_no_other_operators
 
stream<Tick> ticks := all Tick(symbol=”APMA”);

and

on event_template and event_template action;
 
on A() and B() executeAction();

if

if ordinary_exp and ordinary_exp then block;
 
if x and y then {myBlock;}

as

import "plug-in-library" as identifier;
 
import "MyPlugin" as foo;

event1() key field1 duration;
event2() key field2 as field1 duration;
 
CarNum() key road within 1 hour;
Accident() key roadName as road within 1 hour;

at

at(minutes, hours, days_of_month, months,
days_of_week [,seconds])
 
on all at(5, 9, *, *, *) success;

log string [at log_level];
 
log "Your message here" at INFO;

between

between (identifier1 identifier2 ...)
 
between (a b)

boolean

boolean identifier;
 
boolean marketOpen;

bounded

break

break;

by

catch

chunk

chunk identifier;
 
chunk complexProductInfo;

completed

on all completed event_expression action;
 
on all completed A(f < 10.0) {}

constant

constant type name := literal;
 
constant float GOLDEN := 1.61803398874;

context

context(string name)
context(string name, boolean receivesInput)
 
context c:=context("test");

continue

continue;

currentTime

log currentTime.toString();
send TestEvent(currentTime) to "output";

decimal

decimal identifier;
 
decimal exactValue;
exactValue := 1.2345d;

day
days

within integer day | days
 
within 3 days

dictionary

dictionary <key_type, data_type> identifier;
 
dictionary <integer, string> myOrders;

die

die;
 
on NewStock (chosenStock.name,
chosenStock.owner) die;

else

emit

emit event;
 
emit newEvent;

emit...to

emit event to channel;
 
emit newEvent to "com.apamax.pricechanges";

enqueue

enqueue event;
 
enqueue newEvent;

enqueue ...to

enqueue event_expr to context_expr;
enqueue event_expr to sequence<context_expr>;
 
enqueue tick to c;

event

event event_type {
   [ [wildcard] field_type field_name; |
   constant field_type field_name := literal; |
   action_definition ... ]
}
 
event StockTick {
   string name;
   float price;
}

every

every batchPeriodExpr
 
from v in values
within 3.0 every 3.0
select v

every batchSizeExpr
 
from v in values
retain 3 every 3
select v

find every event:coassignment
select_or_having_clause
 
find every A:a
select avg(a.x):aax { }

false

find

find
   [every]
   query_event_pattern
   [where_clause]
   [within_clause]
   [without_clause]
   [select_clause]
   [having_clause] {
   block
   }
 
find
   Withdrawal:w1 -> Withdrawal:w2
   where w2.country != w1.country {
      log "Suspicious withdrawal: "
   
   }

float

float identifier;
 
float squareRoot;

for

forStatement ::= for counter
in sequenceblock;
 
for i in s { 
print i.toString();
}

from

from itemIdentifier in streamExpr
[windowDefinition]
 
from t in ticks retain 3

from itemIdentifier in streamExpr
[windowDefinition]
from itemIdentifier in streamExpr
[windowDefinition]
 
from x in letters retain 2
from y in numbers retain 2
select P(x,y)

[listener :=] from streamExpr :
identifier statement
 
float p;
from t in all Tick(symbol="APMA")
   select t.price : p {
   print "'APMA' price is: " +
p.toString();
}

group by

group by groupByExpr [, groupByExpr]...
 
from t in ticks
within 60.0 group by t.symbol
select mean(t.price)

having

from t in all Temperature()
within 60.0
having count() > 10
select mean(t.value)

having boolean_projection_expr
 
find every ATMWithdrawal:w
   having last(w.amount) > THRESHOLD
      * avg(w.amount)
   select last(w.transactionId):tid {
   send SuspiciousTransaction(tid) to
      SuspiciousTxHandler;
   }

hour
hours

within integer hour | hours
 
within 5 hour

if

ifStatement ::=
if booleanExpression then block
| if booleanExpression
then block else block
| if booleanExpression
then block else  ifStatement block
::= {
statementList
}
 
if floatVariable > 5.0 then {
integerVariable := 1;
} else if floatVariable < -5.0 then {
integerVariable := -1;
} else {
integerVariable := 0;
}

import

import "plug-in_name" as identifier;
 
import "complex_plugin" as complex;

in

on event_name(event_field in [range])
 
on all A(m in [0:10])

inputs

inputs {
event_type(event_filter)
key query_key
[within_clause]
[retain_clause]
[with_unique_clause]
[time_from_clause wait_clause
[or_clause]] ;
 
[ event_type(event_filter)
key query_key
[within_clause]
[retain_clause]
[with_unique_clause]
[time_from_clause wait_clause
[or_clause]] ; ]...
}

inputs {
Transaction() key
source as txSource,
dest as txDest
within PERIOD;
Acknowledgement() key
dest as txSource,
source as txDest
within PERIOD
}

integer

integer identifier;
 
integer count;

join

join itemIdentifier in streamExpr
[windowDefinition]
on joinKeyExpr1 equals joinKeyExpr2
 
from r in priceRequest
join p in prices
partition by p.symbol retain 1
on r.symbol equals p.symbol
select p.price

join(sequence<string> s)
 
sequence<string> s :=
["Something", "Completely", "Different"];   
print ", ".join(s);

key

key field_name [as field_name2] duration

inputs {
Withdrawal() key cardNumber within (period);
}

largest

location

location(15.23, 24.234, 19.1232, 28.873)

log

log string [at log_level];
 
log "Your message here" at INFO;

millisecond
milliseconds
msec

within integer millisecond | milliseconds | msec
 
within 100 msec

minute
minutes
min

within integer minute | minutes | min
 
within 3 min

monitor

monitor monitor_name {
...
}
 
monitor SimpleShareSearch {
...
}

monitor.subscribe("channel_name");
monitor.unsubscribe("channel_name");
 
action start_trade()
{
// Subscribe to two channels:
monitor.subscribe(“SOW_Ticks");
monitor.subscribe(“IBM_Ticks");
}

new

new typeName;
 
b := new Foo();

not

not event_template
 
on A() and not B() executeAction();

if not ordinary_exp then block;
 
if not x then myBlock;

on

on [all] event_expression action;
 
on NewsItem("ACME",*) findStockChange();

onBegin
Recovery

action onBeginRecovery() { }
 
action onBeginRecovery() {  
if (timeFormatPlugin.getTime() -
currentTime > (60.0 * 60.0 * 2)
then
{ longDowntime:=true;
... // do something if
// downTime was long
}
}

onConclude
Recovery

action onConcludeRecovery() { }
 
action onConcludeRecovery() {    
initiateListener(); // go back
//to normal
}

ondie

action ondie() { }
 
action ondie() {    
log "sub-monitor terminating for "
+ myId;
route InternalError("Foo");
}

onload

action onload(){ ... }
 
action onload() {   
on all StockTick(*,*):newTick {
processTick();   
}
}

onunload

action onunload() { };
 
action onunload() {   
route LastMonitorTerminating();
}

optional

or

on event_template or event_template action;
 
on A() or B() executeAction();

if ordinary_exp or ordinary_exp then block;
 
if x or y then myBlock;

package

package identifier;
 
package com.apamax.orders;

parameters

parameters {
data_type parameter_name;
[ data_type parameter_name; ]...
}
 
parameters {
integer threshold;
float period;
}

partition
by

partition by partitionByExpr
[, partitionByExpr]...
 
from t in all Tick()
partition by t.symbol
retain 10 with unique t.price
select t.price

persistent

persistent monitor string
 
persistent monitor ManageOrders

print

print string;
 
print "Your message here.";

query

query name {
[ parameters { parameters_block } ]
inputs { inputs_block }
find pattern block
[ action_definition ... ]
}
 
query FraudulentWithdrawalDetection2 {
inputs {
Withdrawal() key userId retain 3;
}
find Withdrawal:w1 -> Withdrawal:w2
where w1.city != w2.city {
log "Suspicious withdrawal: "
+ w2.toString() at INFO;
}
}

retain

retain windowSizeExpr
 
inputs {
Withdrawal() key userId retain 3;
}
 
from v in values
retain 10
select mean(v)

retain all

retain all
 
from v in values
retain all
select mean(v)

return

returns typeToReturn
return retValue
 
action complexAction(
integer i, float f) returns string {   
// do something  
return "Hello";
}

returns

route

route event();
 
route StockTick();

rstream

select [rstream] selectExpr
 
from i in inputs
retain 2
select rstream i;

second
seconds
sec

within integer second | seconds | sec
 
within 3 sec

select

select [rstream] selectExpr
 
from v in values
retain 10
select mean(v);

select projection_expr:identifier
 
find every ATMWithdrawal:w
having last(w.amount) > THRESHOLD * avg(w.amount)
select last(w.transactionId):tid {
send SuspiciousTransaction(tid) to
SuspiciousTxHandler;
}

send...to

send event_expr to channel;
send event_expr to context;
send event_expr to sequence<channel>;
 
send tick to "ticks-SOW";

sequence

sequence<data_type> identifier;
 
sequence<float> myPrices;

smallest

spawn

spawn action([parameter_list]);
 
action onload() {   
spawn forward("a", "channelA");   
spawn forward("b", "channelB");
}

spawn...to

spawn action([arg_list]) to context_expr;
 
spawn doCalc(cal) to context(“Calculation”);

static

stream

stream<type> name;
 
stream<decimal> prices;

streamsource

string

string identifier;
 
string message;

then

throw

to

true

try

try block1
catch(Exception variable) block2

try {
return float.parse(prices[fxPair]);
} catch(Exception e) {
return 1.0;
}

unbounded

unique

unmatched

on all unmatched event_expression[:coassignment]
action;
 
on all unmatched Tick():tick processTick();

using

using packageName.{aggregateName|eventName};
 
using com.apamax.custom.myAggregateFunction;

wait

wait(float)
 
on A() -> wait(10.0) -> C() success;

wait(value):identifier

find wait(1 minute):previous
-> DoorOpened:d without Unlock:u

where

where booleanExpr
 
from t in ticks
retain 100 where t.price*t.volume>threshold
select mean(t.price)

where booleanExpr
 
find LoggedIn:lc -> OneTimePass:otp
where lc.user = otp.user
within 30.0 {
emit AccessGranted(lc.user);
}

while

whileStatement ::= while booleanExpression
block
 
while integerVariable > 10 {   
integerVariable := integerVariable – 1;   
on StockTick(
"ACME", integerVariable) doAction(); }

wildcard

wildcard param_type param_name;
 
event StockTick {   
string name;   
float price;   
wildcard string exchange;
}

with unique

with unique keyExpr
 
from p in pairs
retain 3 with unique p.letter
select sum(p.number)

within

within(float)
 
on A() -> B() within(30.0) notifyUser();

within windowDurationExpr
 
from v in values
within 20.0
select mean(v);

within durationExpression
[ between ( identifer1 identifier2 ... ) ]
 
find Withdrawal:w1 -> Withdrawal:w2
where w2.country != w1.country
within 1 hour {
log "Suspicious withdrawal: "
+ w2.toString() at INFO;
}

without

without typeId : coassignmentId
[ where boolean_expression ]
[ between ( identifier1 identifier2... )]
 
find OuterDoorOpened:od -> InnerDoorOpened:id
where od.user = id.user
without SecurityCodeEntered:sce
where od.user = sce.user {
emit Alert("Intruder "+id.user);
}

xor

xor event_template
 
on A() xor B() notifyUser();

if ordinary_exp xor ordinary_exp then block;
 
if x xor y then myBlock;

#

#identifier
 
print f.#integer.toString();