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 });
    }
}

@@ -655,16 +655,46 @@
 Handler;
+%0A    private byte %5B%5D snapshot;
 %0A%0A    pu
@@ -795,230 +795,250 @@
 ing 
-executionPlan = %22@config(async = 'true')define stream testStream (src string, dst string, bytes int);%22 +%0A            %22@info(name = 'queryTest') from testStream%5Bbytes %3E 2990%5D select src,dst insert into outputTestStream;%22
+query = %22@info(name = 'queryTest') from testStream%5Bbytes %3E 2990%5D select src,dst insert into outputTestStream;%22;%0A        String executionPlan = %22@config(async = 'true')define stream testStream (src string, dst string, bytes int);%22 + query 
 ;%0A%0A
+%0A
     
@@ -1447,24 +1447,24 @@
 me.start();%0A
-
         inpu
@@ -1522,24 +1522,1075 @@
 stStream%22);%0A
+        snapshot = executionPlanRuntime.snapshot();%0A        executionPlanRuntime.shutdown();%0A%0A        //A%C3%B1adir una query e inicializar el nuevo executionPlanRunTime%0A        //%0A        //%0A        //%0A        //%0A        /*%0A        query.concat(%22@info(name = 'queryTest2') from testStream%5Bbytes %3E 2990%5D select src insert into outputTestStream;%22);%0A        executionPlanRuntime = siddhiManager.createExecutionPlanRuntime(executionPlan);%0A        executionPlanRuntime.restore(snapshot);%0A        executionPlanRuntime.start();%0A        */%0A%0A        query.concat(%22@info(name = 'queryTest2') from testStream%5Bbytes %3E 2990%5D select src insert into outputTestStream;%22);%0A        executionPlanRuntime = siddhiManager.createExecutionPlanRuntime(executionPlan);%0A%0A        executionPlanRuntime.addCallback(%22queryTest2%22, new QueryCallback() %7B%0A            @Override%0A            public void receive(long l, Event%5B%5D events, Event%5B%5D events1) %7B%0A                log.info(%22Alert with more than 2990 bytes!!! %7B%7D%22, l);%0A            %7D%0A        %7D);%0A%0A        executionPlanRuntime.start();%0A%0A
     %7D%0A%0A%0A    

Add new query to executionPlan