define stream TempStream (deviceID long, roomNo int, temp double);

from TempStream 
select roomNo, temp * 9/5 + 32 as temp, 'F' as scale, roomNo >= 100 and roomNo < 110 as isServerRoom
insert into RoomTempStream;

SiddhiManager siddhiManager = new SiddhiManager();

String executionPlan = "" +
        "ddefine stream TempStream (deviceID int, roomNo int, temp float);" +
        "" +
        "@info(name = 'query1') " +
        "from TempStream " +
        "select roomNo, temp * 9/5 + 32 as temp, 'F' as scale, roomNo >= 100 and roomNo < 110 as isServerRoom " +
        "insert into RoomTempStream;";

ExecutionPlanRuntime executionPlanRuntime = siddhiManager.createExecutionPlanRuntime(executionPlan);

executionPlanRuntime.addCallback("query1", new QueryCallback() {
    @Override
    public void receive(long timeStamp, Event[] inEvents, Event[] removeEvents) {
        EventPrinter.print(timeStamp, inEvents, removeEvents);
    }
});

InputHandler inputHandler = executionPlanRuntime.getInputHandler("TempStream");
executionPlanRuntime.start();

inputHandler.send(new Object[] {12344, 201, 28.2f});
inputHandler.send(new Object[] {12345, 202, 22.2f});
inputHandler.send(new Object[] {12346, 203, 24.2f});

//Shutting down the runtime
executionPlanRuntime.shutdown();

//Shutting down Siddhi
siddhiManager.shutdown();

from TempStream#window.time(1 min)
select roomNo, avg(temp) as avgTemp
group by roomNo
insert all events into AvgRoomTempStream ;

from TempStream[temp > 30.0]#window.time(1 min) as T 
  join RegulatorStream[isOn == false]#window.length(1) as R
  on T.roomNo == R.roomNo
select T.roomNo, T.temp, R.deviceID, 'start' as action
insert into RegulatorActionStream ;

from every a1 = atmStatsStream[amountWithdrawed < 100]
    -> b1 = atmStatsStream[amountWithdrawed > 10000 and a1.cardNo == b1.cardNo]
    within 1 day
select a1.cardNo as cardNo, a1.cardHolderName as cardHolderName, b1.amountWithdrawed as amountWithdrawed, b1.location as location, b1.cardHolderMobile as cardHolderMobile
insert into possibleFraudStream;

from every e1=FilteredStockStream[price>20], 
           e2=FilteredStockStream[((e2[last].price is null) and price>=e1.price) or ((not (e2[last].price is null)) and price>=e2[last].price)],

e3=FilteredStockStream[price<e2[last].price] 
select e1.price as priceInitial, e2[last].price as pricePeak, e3.price as priceAfterPeak 
insert into PeakStream ;

public class SiddhiBolt implements IRichBolt {
protected OutputCollector collector;
    protected  SiddhiManager siddhiManager = null;
    protected String executionPlan = null;
    ExecutionPlanRuntime executionPlanRuntime = null;
    protected HashMap<String,InputHandler> handlers = null;

    public SiddhiBolt(String plan) {
        this.executionPlan = plan;
    }

    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        this.siddhiManager = new SiddhiManager();
        this.executionPlanRuntime = siddhiManager.createExecutionPlanRuntime(executionPlan);
        addCallbacks();
        handlers = new HashMap<String,InputHandler>();
        executionPlanRuntime.start();
    }

    public void execute(Tuple tuple) {
        String inputStream = tuple.getSourceStreamId();
        InputHandler inputHandler = getInputHandler(inputStream);

        List<Object> values = tuple.getValues();
        Object[] objects = values.toArray();

        try {
            inputHandler.send(objects);
        }catch (Exception e){
            LOG.error("Send stream event error: ", e);
        }
        // collector.fail(tuple); //test replay
        collector.ack(tuple); // remember ack the tuple
        // Make sure that add anchor tuple if you want to track it
        // collector.emit(streamid, tuple,new Values(counters, now));
    }

    public InputHandler getInputHandler(String streamName){
        InputHandler handler = null;
        if(handlers.containsKey(streamName))
            handler = handlers.get(streamName);
        else {
            handler = executionPlanRuntime.getInputHandler(streamName);
            if (handler != null) {
                handlers.put(streamName, handler);
            }
        }
        return handler;
    }

    //Need Override
    public void addCallbacks( ){
        //StreamCallback example
        executionPlanRuntime.addCallback("outputStream", new StreamCallback() {

            @Override
            public void receive(Event[] events) {
                LOG.info("receive events: " + events.length);
                for (Event e:events)
                    LOG.info(e);
            }
        });

        //QueryCallback example
        executionPlanRuntime.addCallback("query1", new QueryCallback() {
            @Override
            public void receive(long timeStamp, Event[] inEvents, Event[] removeEvents) {
                printEvents(timeStamp, inEvents, removeEvents);
            }
        });
    }

    public void printEvents(long timeStamp, Event[] inEvents, Event[] removeEvents){
        StringBuilder sb = new StringBuilder();
        sb.append("Events{ @timeStamp = ").append(timeStamp).append(", inEvents = ").append(
            Arrays.deepToString(inEvents)).append(", RemoveEvents = ").append(Arrays.deepToString(removeEvents)).append(" }");
        LOG.info(sb.toString());
    }

    public void cleanup() {
        //Shutting down the runtime
        executionPlanRuntime.shutdown();

        //Shutting down Siddhi
        siddhiManager.shutdown();
    }
}

2015-12-15