<commit_before>package net.redborder.correlation.siddhi;

import com.lmax.disruptor.EventHandler;
import net.redborder.correlation.kafka.disruptor.MapEvent;

public class SiddhiHandler implements EventHandler<MapEvent> {
    @Override
    public void onEvent(MapEvent mapEvent, long sequence, boolean endOfBatch) throws Exception {
        System.out.println("Handler: " + mapEvent.getData().toString());
    }
}<commit_msg>First proof of concept for siddhi.<commit_after>package net.redborder.correlation.siddhi;

import com.lmax.disruptor.EventHandler;
import net.redborder.correlation.kafka.disruptor.MapEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.wso2.siddhi.core.ExecutionPlanRuntime;
import org.wso2.siddhi.core.SiddhiManager;
import org.wso2.siddhi.core.event.Event;
import org.wso2.siddhi.core.query.output.callback.QueryCallback;
import org.wso2.siddhi.core.stream.input.InputHandler;

import java.util.Map;

public class SiddhiHandler implements EventHandler<MapEvent> {
    private static final Logger log = LoggerFactory.getLogger(SiddhiHandler.class);
    private InputHandler inputHandler;

    public SiddhiHandler() {
        SiddhiManager siddhiManager = new SiddhiManager();

        String executionPlan = "@config(async = 'true')define stream testStream (src string, dst string, bytes int);" +
            "@info(name = 'queryTest') from testStream[bytes > 2990] select src,dst insert into outputTestStream;";

        ExecutionPlanRuntime executionPlanRuntime = siddhiManager.createExecutionPlanRuntime(executionPlan);

        executionPlanRuntime.addCallback("queryTest", new QueryCallback() {
            @Override
            public void receive(long l, Event[] events, Event[] events1) {
                log.info("Alert with more than 2990 bytes!!! {}", l);
            }
        });

        executionPlanRuntime.start();
        inputHandler = executionPlanRuntime.getInputHandler("testStream");
    }


    @Override
    public void onEvent(MapEvent mapEvent, long sequence, boolean endOfBatch) throws Exception {
        Map<String, Object> data = mapEvent.getData();
        String src = (String) data.get("src");
        String dst = (String) data.get("dst");
        Integer bytes = (Integer) data.get("bytes");
        inputHandler.send(new Object[] { src, dst, bytes });
    }
}
{'fraction_non_alphanumeric': 0.10708191126279863, 'fraction_numerical': 0.006825938566552901, 'mean_word_length': 3.968220338983051, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5217265', 'n_tokens_mistral': 682, 'n_tokens_neox': 666, 'n_words': 150}