CODEWITHSUNDEEP
wso2complex-event-processingsiddhiwso2-cep
Johnny
Johnny

Reputation: 7321

Why do all events come in bulk rather than one by one?

I'm testing a very simple CEP query with an externally timed window. The query is define stream LoginEvents (timeStamp long, ip string, phone string); @info(name = 'query1') from LoginEvents#window.externalTime(timeStamp,5 sec) select timeStamp, ip insert all events into uniqueIps;;

Looking at the unit test here, I thought what would happen is that the callback would be invoked 9 times, 5 times for incoming events and 4 for expiring. Instead, it's only invoked once. Why is that and how can I get to the state where the callback is invoked for every event?

Upvotes: 0

Views: 138

Answers (2)

Johnny
Johnny

Reputation: 7321

Here's my implementation. I had to copy the send method as it's private. You can replace my implementation of the timestamp getter as it's very specific to my use case.

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wso2.siddhi.core.event.ComplexEvent;
import org.wso2.siddhi.core.event.ComplexEventChunk;
import org.wso2.siddhi.core.event.Event;
import org.wso2.siddhi.core.event.stream.StreamEvent;
import org.wso2.siddhi.core.query.output.callback.QueryCallback;
import java.util.Arrays;

public abstract class CustomQueryCallback extends QueryCallback {

    private static final Logger log = LoggerFactory.getLogger(CustomQueryCallback.class);

    public void receiveStreamEvent(ComplexEventChunk complexEventChunk) {
        while (complexEventChunk.hasNext()) {
            ComplexEvent streamEvent = complexEventChunk.next();
            Event event = new Event(streamEvent.getOutputData().length).copyFrom(streamEvent);
            Event[] events = new Event[]{event};
            long timestamp = (streamEvent.getType() == StreamEvent.Type.EXPIRED ? streamEvent.getTimestamp() : (long) streamEvent.getOutputData()[2]);
            if (streamEvent.getType() == StreamEvent.Type.EXPIRED){
                send(timestamp, null, events);
            } else {
                send(timestamp, events, null);
            }
        }
    }

    private void send(long timeStamp, Event[] currentEvents, Event[] expiredEvents) {
        try {
            receive(timeStamp, currentEvents, expiredEvents);
        } catch (RuntimeException e) {
            log.error("Error on sending events" + Arrays.deepToString(currentEvents) + ", " + Arrays.deepToString(expiredEvents), e);
        }
    }

}

Upvotes: 1

suho
suho

Reputation: 912

Here all events are sent to Siddhi without any time delay, and hence Siddhi processes all these events together. Thats why the events are returned as a bulk.

If you want a callback to be invoked for every event then you have to extend either StreamCallback or QueryCallback and iterate the returned event array and invoke the callback for each event.

Upvotes: 1

Related Questions