Edit test case to retrieve values from aggregate
@@ -138,12 +138,38 @@ public class AggregationTestCase {
String query = "" +
"define aggregation test " +
"from cseEventStream " +
- "select symbol, avg(price1) as avgPrice, sum(price1) as totprice1, (quantity * volume) as mult " +
- "aggregate by timestamp every sec...year ;";
+ "select symbol as s, avg(price1) as avgPrice, sum(price1) as totprice1, count(price1) as countPrice, " +
+ "(quantity * volume) as mult " +
+ "aggregate by timestamp every sec...year ;" +
+
+ "define stream barStream (symbol string, value int, startTime string, " +
+ "endTime string, perValue string); " +
+ "" +
+ "@info(name = 'query1') " +
+ "from barStream as b join test as a " +
+ "within 1496289900000L, 1596434877000L " +
+ "per \"minutes\" " +
+ "select a.avgPrice, a.totprice1 as sumPrice, a.mult, a.countPrice " +
+ "insert all events into fooBar; ";
SiddhiAppRuntime siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(cseEventStream + query);
+ try {
+ siddhiAppRuntime.addCallback("query1", new QueryCallback() {
+ @Override
+ public void receive(long timestamp, Event[] inEvents, Event[] removeEvents) {
+ EventPrinter.print(timestamp, inEvents, removeEvents);
+ if (inEvents != null) {
+ inEventCount.addAndGet(inEvents.length);
+ }
+ if (removeEvents != null) {
+ removeEventCount.addAndGet(removeEvents.length);
+ }
+ eventArrived = true;
+ }
+ });
InputHandler inputHandler = siddhiAppRuntime.getInputHandler("cseEventStream");
+ InputHandler barStreamInputHandler = siddhiAppRuntime.getInputHandler("barStream");
siddhiAppRuntime.start();
// Thursday, June 1, 2017 4:05:50 AM
@@ -196,8 +222,14 @@ public class AggregationTestCase {
inputHandler.send(new Object[]{"CISCO", 260f, 44f, 200L, 16, 1596434876000L});
Thread.sleep(2000);
+
+ barStreamInputHandler.send(new Object[]{"IBM", 1, "2017-06-01 09:35:51 +05:30",
+ "2017-06-01 09:35:52 +05:30", "seconds"});
+ Thread.sleep(50);
+ } finally {
siddhiAppRuntime.shutdown();
}
+ }
@Test
