How can i pick out value from consecutive value of

Question

for example, that is data:

1,1470732420000,0
2,1470732421000,0
3,1470732422000,0
4,1470732423000,86
5,1470732424000,87
6,1470732425000,88
7,1470732426000,84
8,1470732427000,0
9,1470732428000,0
10,1470732429000,0
11,1470732430000,89
12,1470732431000,89
13,1470732432000,87
14,1470732433000,89
15,1470732434000,85
16,1470732435000,89
17,1470732436000,89
18,1470732437000,87
19,1470732438000,86
20,1470732439000,88
21,1470732440000,0
22,1470732441000,0
23,1470732442000,0
24,1470732443000,87
25,1470732444000,85
26,1470732445000,86
27,1470732446000,0
28,1470732447000,0
29,1470732448000,0
30,1470732449000,0

column one is id,column two is timestamp,column three is value,1 sec interval between the timestamp.

i want monitoring the value of event,if i found out value>=85(e.g. id=4), i will starting counting,if the next two consecutive value>=85(e.g. id=5/id=6),then i will put the third value of event to OutputStream.(e.g. id=6,value=88,timestamp=1470732425000)

at the same time i clear the counting and wait value lower than 85(e.g. id=7,value=84), then i will monitoring again,when i found out value>=85(e.g. id=11,value=89) i will starting counting,if the next two consecutive value>=85(e.g. id=12/id=13),then i will put the third value of event to OutputStream.(e.g. id=13,value=87,timestamp=1470732432000)...

all this is i wanna do,before i post this ask, i've got an answer in this post,i've tried this code:

from every a1=InputStream[value>=85], a2=InputStream[value>=85]+, a3=InputStream[value<85]
select a2[1].id, a2[1].value
having (not (a2[1] is null))
insert into OutPutStream;

and it works,but i found out it will insert the value into OutputStream after the value<=85,and what i want is if i got three consecutive value>=85 then i insert into the value immediately.(i don't want to wait if the next value>=85 all the times) in fact, i just wanna record value of third seconds in three consecutive seconds value(>=85) . i'm using wso2das-3.1.0-SNAPSHOT.

Answer 1

Though DAS (Siddhi) supports sequence/pattern processing, for your requirement you might need to write a custom extension. I have written a sample window processor extension to cater your requirement (source code). Download and place siddhi-extension-condition-window-1.0.jar in <das_home>/repository/components/lib/ directory and restart the server. Refer to the test case to get an idea of the usage of the extension.