CODEWITHSUNDEEP
complex-event-processingsiddhiwso2-streaming-integratorwso2-cep
Jun Li
Jun Li

Reputation: 31

Use External Window Time Stamp to Debug Siddhi Stream Query

I am planning to use the historical event traces (stored in JSON with my own event time stamp recorded for each event) to debug the Siddhi stream queries that I have just created. My stream starts with:

from MyInputEventStream#window.externalTime(my_own_timestamp, 10 min) select some_fields insert into MyOutpuStream;

and I will input my events from traces, one by one.

Supposed event 1 arrives at the specified my_own_timestamp = 1528905600000, which is 9 am PST time, June 13. and event 2 arrives at 11 minutes later, my_own_timestamp = 1528906260000. I believe that I will get the output at MyOutpuStream at 9:10 am, as time_stamp(e2) - time_stamp(e1) > 10 min, and e2 will trigger the system after the windows passes.

Now supposed event 1 arrives at my_own_timestamp = 1528905600000, that is, 9:00 am. But no events will arrive in the next 2 hours. Do I still get the output at 9:10 am, as in reality, the window time should expire at 9:10 am, independent of when the next event should arrive? But it seems that in this case, the internal timing system of Siddhi will have to incorporate my event input's time stamp, and then set the expiration time of the events based on the clock system of the process on which the Siddhi is running. Is this correct? could you help clarify it.

Upvotes: 0

Views: 213

Answers (1)

Grainier
Grainier

Reputation: 1654

You won't get an output at 9:10 am. Because if you use externalTime, the event expiration logic will entirely base on the timestamp that you defined. And it will wait for a timestamp that satisfies the time difference which is greater than or equal to expire the previous event.

What internally happens is;

def array previousEvents;

foreach currentEvent in currentEvents (events that are coming in):
    def currentTime = currentEvent.timestamp;
    foreach previousEvent in previousEvents:
        def previousTime = previousEvent.timestamp;
        def timeDiff = previousTime - currentTime + windowLength;
        if (timeDiff <= 0) {
            remove previousEvent from previousEvents;
            set expired timestamp of previousEvent to currentTime;
            expire previousEvent;
        }
    previousEvents.add(currentEvent);

Upvotes: 0

Related Questions