When starting with Apama EPL, you might find yourself frequently referring to the documentation to understand the syntax—for instance, how to create a hashmap, store data locally, or manage information within an object, similar to using a POJO model in Java.
In this article, I aim to provide a collection of straightforward examples that illustrate these concepts. The goal is to present the logic clearly and concisely, making it easy for you to directly apply these samples when needed.
If you come from the Java world like I do, there are certain familiar patterns and constructs you might be curious about replicating in EPL. These include:
- Creating a POJO-like class to hold information
- Working with hashmaps
- Using arrays
- Implementing a hashmap with
Stringas the key andObjectas the value - Creating a hashmap with
Stringas the key and arrays as the value - exchanging information between EPLs
- do something at startup
- keep local cache up to date with the platform
- query for managed objects
- Extract/parse information from managed object
- using try catch block
In this article, I’ll explore how to achieve these common Java patterns in EPL.
When I first started working with EPL, it took me a while to wrap my head around the fact that nothing is sequential. Everything revolves around events, and you need to think in terms of event-driven logic. Additionally, you must ensure that information is stored until all the relevant events are received. Once that’s done, you can proceed with your calculations or processing.
Just as Java has classes, Apama EPL has monitors. Similar to how a class in Java has a constructor, a monitor in EPL begins with an onload() function, which serves as the starting point for execution.
Consider the code below:
// imports go here
// sample event definition like a POJO in java
event Device {
string lowFillLevelTimestamp;
string highFillLevelTimestamp;
float maxMass;
float lowFillLevelValue;
float highFillLevelValue;
float timePeriod;
}
event DeviceFound {
integer reqId;
string deviceId;
string serialNum;
string extId;
boolean isInUse;
}
// java equivalent to class
monitor DeviceChangeDetector {
// Hashmap <string, DeviceFound[]> a hashmap of string and array of objects
//request-id ==> Event
dictionary<string, sequence<DeviceFound>> childDeviceEvents := new dictionary<string, sequence<DeviceFound>>;
// constant definition e.g. some fragment name in your managed object
constant string TRANSMISSION_PERIOD_FRAGMENT := "transmissionTimePeriod";
constant string LOW_FILL_LEVEL_FRAGMENT := "lowFillLevel";
// hashmap of string and object
dictionary<string,Device> devices := {};
/** Initialize the application */
action onload() {
// Application starts here
log "Starting EPL ..." at INFO;
}
}
In the code above we see a couple of events we have defined before our monitor definition. We can use them as model classes like we do in java to store and pass data around. We then use those same events in the hashmaps we defined which in EPL are dictionaries. The array equivalent in EPL is sequence which we have also shown above.
I won’t go into too much detail about Apama, but to summarize briefly: to retrieve data, you first subscribe to a channel, then create and send your query, listen for the response, and finally store or process the data you receive.
It’s important to always subscribe to the channel before sending your query. Otherwise, there’s a risk of missing the response sent by the Apama engine before you’ve started listening. Additionally, there is typically an end event that indicates you’ve received the last event of interest.
Sample code to use query language and get result. Notice we add the query param to the findManagedObject request.
action init()
{
log "Search for devices..." at INFO;
FindManagedObject findManagedObject := new FindManagedObject;
findManagedObject.reqId := Util.generateReqId();
findManagedObject.params.add("query", "$filter=( has(c8y_IsDevice) and someCustomFragment eq 'myCustomDeviceType' )");
/** Subscribe to FindManagedObjectResponse.SUBSCRIBE_CHANNEL to
* listen for responses.
*/
monitor.subscribe(FindManagedObjectResponse.SUBSCRIBE_CHANNEL);
on all FindManagedObjectResponse(reqId=findManagedObject.reqId) as resp
and not FindManagedObjectResponseAck(reqId=findManagedObject.reqId) {
log "Device found with id: " + resp.managedObject.id at INFO;
// do something with the device you found
ProcessDevice(resp.managedObject);
}
/** Once the request has completed, stop the listener.
Listen for request completed acknowledgement. */
on FindManagedObjectResponseAck(reqId=findManagedObject.reqId) {
log "Find devices request completed " +
findManagedObject.reqId.toString() at INFO;
listenForChanges();
}
send findManagedObject to FindManagedObject.SEND_CHANNEL;
}
In the sample below, we demonstrate how to process the managed object response obtained from your query. Suppose we want to process a device, meaning we need to extract various fields from the device and store them locally in a dictionary for future reference.
This example illustrates that we can handle nested properties within the managed object. Additionally, the fields can hold values of different types, such as strings or decimals, showcasing EPL’s flexibility in managing diverse data structures.
Here is a sample nested json that we read in the code below:
"lowFillLevel": {
"value": 0,
"timestamp": "2024-12-11 13:00:00.446"
},
Consider
action initializeDevice(ManagedObject mo) {
try {
// we use the Device event definition from our first snippet
Device device := new Device;
if mo.params.hasKey(LOW_FILL_LEVEL_FRAGMENT){
// CHECK FOR TIMESTAMP
if(mo.params[LOW_FILL_LEVEL_FRAGMENT].hasEntry("timestamp")) {
string timestamp:= AnyExtractor(mo.params[LOW_FILL_LEVEL_FRAGMENT]).getString("timestamp");
device.lowFillLevelTimestamp := timestamp;
}
if(mo.params[LOW_FILL_LEVEL_FRAGMENT].hasEntry("value")) {
optional<float> lowLevelVal := getFloatValue(mo,LOW_FILL_LEVEL_FRAGMENT, "value");
ifpresent lowLevelVal {
device.lowFillLevelValue := lowLevelVal;
} else {
log "unable to read lowFillLevel.value" at WARN;
}
}
}else {
log "device detected without low fill level fragment..." at WARN;
}
optional<float> mMass:= readMaxMassValue(mo);
ifpresent mMass {
device.maxMass := mMass;
}
// local device in cache
devices[mo.id] := device;
float timePeriod := readTimePeriod(mo);
devices[mo.id].timePeriod := timePeriod;
} catch (Exception e) {
log "something went wrong processing the device: " + mo.id at ERROR;
log e.getMessage() at ERROR;
}
}
action getFloatValue(ManagedObject mo, string fragment, string entry) returns optional<float> {
optional<float> response := new optional<float>;
any val:= mo.params[fragment].getEntry(entry);
switch (val as myvalue)
{
case float: {
response := <float> val;
return response;
}
case integer: {
integer int := <integer> val;
response := int.toFloat();
return response;
}
default: {
return response;
}
}
}
In the examples above, so far we have queried the managed objects of interest, parsed them and read them into memory. Now you might be interested in keeping the local cache up to date, meaning whenever something changes in the platform, you would want to update it locally as well.
You can also apply some filtration e.g. comparing some fragment to a value. In the example below we use a custom fragment DEVICE_TYPE_FRAGMENT and compare it to DEVICE_TYPE. This can of course be any other fragment of your interest too.
action listenForChanges()
{
log "Subscribe for device update/create events..." at INFO;
monitor.subscribe(ManagedObject.SUBSCRIBE_CHANNEL);
on all ManagedObject() as mo
{
// if its an update for a device
if mo.isUpdate() and mo.params.hasKey(DEVICE_TYPE_FRAGMENT) and (<string> mo.params[DEVICE_TYPE_FRAGMENT]) = DEVICE_TYPE
{
log "# device update event received : " + mo.toString() at INFO;
processUpdate(mo);
} else if mo.isCreate() {
if mo.params.hasKey(DEVICE_TYPE_FRAGMENT) and (<string> mo.params[DEVICE_TYPE_FRAGMENT]) = DEVICE_TYPE
{
log "# device create event received : " + mo.toString() at INFO;
// send it to the channel to be intercepted by some other EPL where you are listening to the channel
// send mo to "NEW_DEVICE_CHANNEL";
initializeDevice(mo);
}
}
}
// when a device is deleted, check if it exists in our map, if yes, remove
on all ManagedObjectDeleted() as managedObjectDeleted {
// if a device is deleted, remove from map
if(devices.hasKey(managedObjectDeleted.id)){
devices.remove(managedObjectDeleted.id);
}
}
}
Child devices can be easily handled too. In EPL you simply have an array of ids unlike the java/json representation which has the self link inside.
action processChildAdditions(ManagedObject mo) {
log "Log child devices ..." at INFO;
if(mo.childDeviceIds.size() > 0) {
// loop over all child devices in the mo
string childDeviceId;
for childDeviceId in mo.childDeviceIds {
log "found child device with id:" + childDeviceId at INFO;
}
}
}
Below you find an example of converting an array to a CSV string
action sequenceToCsv(sequence<string> ids) returns string {
string csv := "";
string id;
boolean first := true;
for id in ids {
if(first) {
csv:= id;
first := false;
} else {
csv := csv + "," + id;
}
}
return csv;
}
Lets consider another scenario. In apama, events are sent on channels. If you go through the documentation you will see there are multiple channels and you can subscribe to receive
events from these channels. Similarly you can define your own channels aswell. There is no explicit definition of a channel, rather you can simply subscribe to a channel and then
it would be available:
monitor.subscribe("CHILD_SENSOR_FOUND_CHANNEL");
Now finally we consider something relatively complex. First consider that we have a parent device which has a custom fragment called CONNECTED_SENSORS_FRAGMENT. This holds an array of of objects of the following format:
"connectedSensors": [
{
"serialNumber": "1A02OQOUEJO",
"externalId": "70B3D597B000126F",
"isInUse": true,
"childDeviceId": "72100997902"
}, ...
Suppose child devices can be dynamically added or removed from a parent device. In such cases, the connectedSensors fragment in the parent device’s managed object needs to be updated to reflect these changes. To achieve this, we need additional details about the child devices—specifically their serial numbers and external IDs.
Let’s assume we have already identified the child device IDs for devices that are new and not yet included in the connectedSensors fragment. These IDs serve as the input to the starting point of the action in the provided code sample.
There are two main steps in this process:
- Fetch Child Device Details: Retrieve the managed object for each child device from Cumulocity to access its serial number.
- Fetch External ID: Query the external ID for each child device.
Once these details are obtained, the new information is stored locally. Finally, the parent device’s managed object is updated to include the new child devices in the connectedSensors fragment, ensuring the data remains accurate and up to date.
Here are some bullet points which explain what we do in the code below.
DeviceFoundTerminate: An event signaling that all child devices for a given request ID have been processed.childDeviceEvents: A dictionary mapping request IDs to sequences ofDeviceFoundevents, storing information about discovered child devices.processChildSensorsAndUpdate: This action processes child devices of a parent managed object, identifies devices of interest- A unique ID is generated (
globalChildReqId) to track child device processing. This is logged for reference - Subscribes to the
CHILD_SENSOR_FOUND_CHANNELto handleDeviceFoundevents. - Filters devices based on specific criteria, such as the value of a fragment (
FRAGMENT_TO_CHECK). - Action:
initiateExternalIdFetchForFoundDevices-
This action retrieves and updates external IDs for child devices.
-
Iterates over the devices stored in
childDeviceEvents. -
Calls
getAndUpdateExternalIdto fetch and update external IDs. -
On completion:
- Updates the parent device’s connected sensors fragment (
W2A_CONNECTED_SENSORS_FRAGMENT). - Marks the scaling state (
W2A_ISSCALED) asfalse. - Sends an updated managed object to the Cumulocity platform.
- Updates the parent device’s connected sensors fragment (
-
Before you dive in, remember that the Apama EPL code is not really sequential. Technically the starting point of the code execution in the first action is the last line which is annotated by the comment INITIATE
event DeviceFoundTerminate {
integer reqId;
}
// request-id ==> an array of deviceFound events
dictionary<string, sequence<DeviceFound>> childDeviceEvents := new dictionary<string, sequence<DeviceFound>>;
/* fetch the child device
// check if the fragment matches
// if yes then keep it for later use
// from the collected new devices above
// get existing CONNECTED_SENSORS_FRAGMENT add new item to it
// also set isScaled := false
// for every device get the external id
// send update event with updated fragment */
action processChildSensorsAndUpdate(ManagedObject parentMo, sequence<string> childDeviceIds) {
integer globalChildReqId := com.apama.cumulocity.Util.generateReqId();
log "Generated global request id: " + globalChildReqId.toString() +" to fetch/update child device information for parent device:" + parentMo.id;
monitor.subscribe("CHILD_SENSOR_FOUND_CHANNEL");
on all DeviceFound(reqId=globalChildReqId ) as childDeviceFound
and not DeviceFoundTerminate(reqId=globalChildReqId) {
// whenever we get a child device event, we add it to the array
if(childDeviceEvents.hasKey(childDeviceFound.reqId.toString())) {
childDeviceEvents[childDeviceFound.reqId.toString()].append(childDeviceFound);
}else {
childDeviceEvents[childDeviceFound.reqId.toString()] := [childDeviceFound];
}
}
on DeviceFoundTerminate(reqId = globalChildReqId) as requestCompleted {
// once all child devices for a given parent device are found we end up in this section.
initiateExternalIdFetchForFoundDevices(globalChildReqId.toString(), parentMo);
}
// query each child device to get the complete mo
monitor.subscribe(com.apama.cumulocity.FindManagedObjectResponse.SUBSCRIBE_CHANNEL);
// no loop required since we use the ids field in the params below
com.apama.cumulocity.FindManagedObject request := new com.apama.cumulocity.FindManagedObject;
integer reqId := com.apama.cumulocity.Util.generateReqId();
request.reqId := reqId;
log "Request id generated to fetch all new child device managed objects: " + reqId.toString() at INFO;
request.params.add("ids", sequenceToCsv(childDeviceIds));
// Listen for responses based on reqId
on all com.apama.cumulocity.FindManagedObjectResponse(reqId=reqId) as response
and not com.apama.cumulocity.FindManagedObjectResponseAck(reqId=reqId) {
// check type for the response to check if the device is of the type of our interest
if(response.managedObject.params.hasKey("FRAGMENT_TO_CHECK") and <string> response.managedObject.params["FRAGMENT_TO_CHECK"] = "SOME_VALUE") {
log "Child device detected of type: " + FRAGMENT_TO_CHECK;
DeviceFound deviceFound := new DeviceFound;
deviceFound.reqId := globalChildReqId;
deviceFound.deviceId := response.managedObject.id;
deviceFound.isInUse := true;
if(response.managedObject.params.hasKey("c8y_Hardware") and response.managedObject.params["c8y_Hardware"].hasEntry("serialNumber")) {
deviceFound.serialNum := response.managedObject.params["c8y_Hardware"].getEntry("serialNumber").valueToString();
log deviceFound.serialNum at INFO;
}else {
log "Device serial number could not be found" at WARN;
}
send deviceFound to "CHILD_SENSOR_FOUND_CHANNEL";
}
}
// Listen for com.apama.cumulocity.FindManagedObjectResponseAck,
// this indicates that all responses have been received
on com.apama.cumulocity.FindManagedObjectResponseAck(reqId=reqId) as requestCompleted
{
log "All get child device requests completed for global request id:" + globalChildReqId.toString();
send DeviceFoundTerminate(globalChildReqId ) to "CHILD_SENSOR_FOUND_CHANNEL";
//monitor.unsubscribe(com.apama.cumulocity.FindManagedObjectResponse.SUBSCRIBE_CHANNEL);
}
// Send request to find available managed objects
//INITIATE
send request to com.apama.cumulocity.FindManagedObject.SEND_CHANNEL;
}
action initiateExternalIdFetchForFoundDevices (string globalRequestId, ManagedObject parentMo) {
if(childDeviceEvents.hasKey(globalRequestId)) {
sequence<DeviceFound> gatheredEvents := childDeviceEvents[globalRequestId];
DeviceFound item;
integer reqId := 0;
for item in gatheredEvents {
reqId := com.apama.cumulocity.Util.generateReqId();
getAndUpdateExternalId(globalRequestId, reqId, item.deviceId,"c8y_ExternalId_type");
}
if(reqId != 0) {
// here you are listening for the last req id generated in the above loop assuming that all previous would also have been answered
on GenericResponseComplete(reqId= reqId)
{
log "All external id fetch request completed for request id: " + reqId.toString() at INFO;
try {
string childDeviceId;
// an array to send an update of the conencted sensors to the sensor EPL
sequence<string> connectedSensorIdArray := new sequence<string>;
// read the connected sensors frgament from the mo
sequence<dictionary<string, any>> connectedSensors := new sequence<dictionary<string, any>>;
if (parentMo.params.hasKey(CONNECTED_SENSORS_FRAGMENT)) {
// if it has a connected sensor fragment, read it
sequence<any> connectedSensorArray :=<sequence<any>> parentMo.params[CONNECTED_SENSORS_FRAGMENT];
any item;
for item in connectedSensorArray {
dictionary<any,any> temp := <dictionary<any,any>> item;
string id:= AnyExtractor(temp).getString("childDeviceId");
boolean found := false;
for childDeviceId in parentMo.childDeviceIds {
if (id = childDeviceId) {
found := true;
}
}
// only add if you found it!
if(found){
dictionary<string, any> arrayItem := new dictionary<string, any>;
arrayItem["serialNumber"] := AnyExtractor(temp).getString("serialNumber");
arrayItem["isInUse"] := AnyExtractor(temp).getBoolean("isInUse");
arrayItem["childDeviceId"] := AnyExtractor(temp).getString("childDeviceId");
connectedSensorIdArray.append(AnyExtractor(temp).getString("childDeviceId"));
arrayItem["externalId"] := AnyExtractor(temp).getString("externalId");
connectedSensors.append(arrayItem);
}
}
}
DeviceFound temp;
// loop over the new found devices and append them to the existing connected sensor fragment
for temp in gatheredEvents {
//TODO
dictionary<string, any> childSensor := new dictionary <string, any>;
childSensor["childDeviceId"] := temp.deviceId;
connectedSensorIdArray.append(temp.deviceId);
if(temp.serialNum.length() > 0 ){
childSensor["serialNumber"] := temp.serialNum;
}
childSensor["isInUse"] := true;
if(temp.extId.length() > 0) {
childSensor["externalId"] := temp.extId;
}else {
log "No external id found! this may cause issues! Investigate logs to figure out why value was not populated." at WARN;
}
// add them to the conencted sensors array
connectedSensors.append(childSensor);
}
// send update to the platfrom with the updated connected snesor fragment
ManagedObject mo := new ManagedObject;
mo.id := parentMo.id;
mo.params[CONNECTED_SENSORS_FRAGMENT] := connectedSensors;
// mark scaling to false
mo.params[ISSCALED] := false;
sendIsScaledFalseEvent(parentMo.id);
log "New connected sensors found of type ! Updating connected sensors fragment... Mark device as sclaed:= false";
send mo to ManagedObject.SEND_CHANNEL;
log "Removing child process request with id: " + globalRequestId + " from local map. Processing complete..." at INFO;
childDeviceEvents.remove(globalRequestId);
sendConnectedSensorUpdateEvent(parentMo.id, connectedSensorIdArray);
} catch (Exception e ) {
log "failed to update managed object for request :" + globalRequestId at WARN;
log e.getMessage() at WARN;
childDeviceEvents.remove(globalRequestId);
}
}
}
} else {
log "No item found in map for requestId: " + globalRequestId at WARN;
}
}
action getAndUpdateExternalId(string globalReqId, integer reqId, string id, string type) {
log "Get external id for device with id:= " + id at INFO;
string path := "/identity/globalIds/" + id + "/externalIds";
GenericRequest request := new GenericRequest;
request.reqId := reqId;
request.method := "GET";
request.isPaging := true;
request.path := path;
monitor.subscribe(GenericResponse.SUBSCRIBE_CHANNEL);
on all GenericResponse(reqId=request.reqId) as response
and not GenericResponseComplete(reqId=request.reqId)
{
// multiple entries for ext id result in multiple event responses
try {
dictionary<any, any> data := <dictionary<any, any>> response.getBody();
if(data.hasKey("externalId") and data.hasKey("type") and AnyExtractor(data).getString("type") = type) {
string extId:= data["externalId"].valueToString();
log "looking for array entry to update external id" at INFO;
sequence<DeviceFound> gatheredEvents := childDeviceEvents[globalReqId];
DeviceFound item;
for item in gatheredEvents {
if(item.deviceId = id) {
log "updating external id for device: " + id at INFO;
item.extId := extId;
}
}
}else {
log "Not a matching external id for device" + id + " and type: " + type at INFO;
}
}
catch(Exception e) {
log "Failed to parse external id for : " + id at INFO;
log e.getMessage() at WARN;
}
}
send request to GenericRequest.SEND_CHANNEL;
}
Notice that I try to do something sequential in the snippet. I initiate the request to fetch all child devices, every time I find a device i create an event and save it in the map. Eventually when I get the last device, I send the deviceFoundTerminate event. Then I have a listener for this specific event which then calls the function to figure out the external ids. The external id process does not begin till I have all the child devices of interest already in our childDeviceEvents map.
Another important point is that I create a request in one action and listen for completion in another. Check the getAndUpdateExternalId action where i create a request to fetch the external id. The completion part exists in the initiateExternalIdFetchForFoundDevices action. Lastly the example above also shows an example of calling an endpoint in c8y. We do this for external ids since there is no dedicated channel where we can query for this.
Finally consider we have a managed object which has a fragment that contains timestamps.
"timeFrame": [
{
"timeStampStart": "1970-01-01T00:00:00.000Z",
"timeStampEnd": "1970-01-01T23:59:59.000Z"
},
{
"timeStampStart": "1970-01-01T00:00:00.000Z",
"timeStampEnd": "1970-01-01T23:59:59.000Z"
}
],
We need to figure out if an incoming measurement falls within those timestamps. The following code to read the timestamps:
// event definition to hold the timestamps against the device
event Device{
string id;
sequence<dictionary<string, string>> timeslots;
}
// timeframe
if(mo.params.hasKey(TIME_FRAME_FRAGMENT)) {
sequence<dictionary<string, string>> timeslots := new sequence<dictionary<string, string>>;
sequence<any> timeFrameArray := <sequence<any>> mo.params[TIME_FRAME_FRAGMENT];
any item;
for item in timeFrameArray {
dictionary<any, any> temp := <dictionary<any, any>> item;
dictionary<string, string> timeslotsItem := new dictionary<string, string>;
timeslotsItem["timeStampStart"] := AnyExtractor(temp).getString("timeStampStart");
timeslotsItem["timeStampEnd"] := AnyExtractor(temp).getString("timeStampEnd");
timeslots.append(timeslotsItem);
}
device.timeslots := timeslots;
}
Each day consists of 86400 seconds. When we get a measurement, we can take mod to figure out where it falls. If it falls within the time slots, then it is within the time frame.
action isMeasurementWithinTimeFrame(string sensorId, float timestamp) returns boolean {
try {
// measurement timestamp
//float epochTime := TimeFormat.parseTime("yyyy-MM-dd'T'HH:mm:ss.sss'Z'", timestamp);
integer time := timestamp.integralPart() % 86400;
sequence<dictionary<string,string>> timeSlots:= siloDevices[sensorId].timeslots;
if(timeSlots.size() = 0) {
// no time slots defined
return true;
}
dictionary<string, string> temp;
for temp in timeSlots {
float start := TimeFormat.parseTime("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", temp[TIME_STAMP_START_FRAGMENT]);
float end := TimeFormat.parseTime("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'", temp[TIME_STAMP_END_FRAGMENT]);
// case equal?
if(end < start) {
log "End date cant be before start date. defaulting to true for time slots" at WARN;
return true;
}
if(time > start.integralPart() and time < end.integralPart()) {
return true;
}
}
return false;
} else {
log "The given sensor was not found in local cache: " + sensorId at WARN;
log "default timeslot is: 00 00 to 24 00" at INFO;
return true;
}
} catch (Exception e) {
log e.getMessage() at ERROR;
log "An error occurred while figuring out if timestamp falls in timeslots for sensorId: " + sensorId at WARN;
return false;
}
}