Using dynamic values in stream queries

One of the great features of Apama stream queries is that the values used in the stream query expression can be dynamically changed throughout the lifetime of the query. This is useful (for example) for setting dynamic thresholds or for changing the aggregation period of a query. The code examples below illustrate these cases.

01. TemperatureAlert alert;
02. from t in all Temperature(sensorId="T001") where t.temperature > threshold
03. select TemperatureAlert(t.sensorId,t.temperature): alert { emit alert; }

01. TemperatureRange range;
02. from t in all Temperature(sensorId="T001") within period every period
03. select TemperatureRange(t.sensorId,min(t.temperature),max(t.temperature)):
04. range {
05. print range.toString();
06. }

In the code examples above, if the variables threshold and period are local variables4, then the value used by the queries are the values of the local variables when the from statement is executed.5 Even if the local variable is assigned a new value at some later point in the program execution, the values used by the queries will be constant throughout the lifetime of the query.

However, if global variables6 or event member variables7 are used and, at a later time, the values of these variables are changed, then these value changes will affect the behavior of the stream queries. The full code examples for the dynamic use-cases are given below.

01. event Temperature { string sensorId; float temperature; }
02. event TemperatureAlert { string sensorId; float temperature; }
03. event ChangeThreshold { float temperature; }

01. monitor TemperatureAlertMonitor {
02. float threshold := 60.0; // a global variable is used
03. action onload() {
04. TemperatureAlert alert;
05. from t in all Temperature(sensorId="T001")
06. where t.temperature > threshold
07. select TemperatureAlert(t.sensorId,t.temperature): alert { emit alert; }
08. ChangeThreshold ct;
09. on all ChangeThreshold():ct { threshold := ct.temperature; }
10. }
11. }

01. event Temperature { string sensorId; float temperature; }
02. event TemperatureRange { string sensorId; float minTemperature;
03. float maxTemperature; }
04. event ChangePeriod { float period; }

01. using com.apama.aggregates.max; using com.apama.aggregates.min;
02. event TemperatureRangeService {
03. float period; // an event member variable is used
04. action init( string id, float _period ) {
05. period := _period;
06. TemperatureRange range;
07. from t in all Temperature(sensorId=id) within period every period
08. select TemperatureRange(id,min(t.temperature),max(t.temperature)):
09. range {
10. print range.toString();
11. }
12. }
13. action setPeriod(float _period ) { period := _period; }
14. }
15. monitor UsesTemperatureRangeService {
16. action onload() {
17. TemperatureRangeService trs := new TemperatureRangeService;
18. trs.init("S001",60.0);
19. ChangePeriod cp;
20. on all ChangePeriod ():cp { trs.setPeriod(cp.period); }
21. }
22. }