Writing a Custom Window

To write a custom window, create a class extending org.wso2.siddhi.core.query.processor.window.WindowProcessor add the SiddhiExtionsion annotation, compile that class, and add the JAR file to the class path <CEP_HOME>/repository/components/lib/ directory. Then add the fully-qualified class name for the implementation class in a new line, to the siddhi.extension file located at <CEP_HOME>/repository/conf/siddhi/ directory.

For example, if you have created the extension with namespace custom and function name lastUnique, you can use that in the query as follows:

from StockExchangeStream[price >= 20]#window.custom:lastUnique(symbol,5) 
select symbol, price
insert into StockQuote;

import org.wso2.siddhi.core.config.SiddhiContext;
import org.wso2.siddhi.core.event.StreamEvent;
import org.wso2.siddhi.core.event.in.InEvent;
import org.wso2.siddhi.core.event.in.InListEvent;
import org.wso2.siddhi.core.query.QueryPostProcessingElement;
import org.wso2.siddhi.core.query.processor.window.WindowProcessor;
import org.wso2.siddhi.query.api.definition.AbstractDefinition;
import org.wso2.siddhi.query.api.expression.Expression;
import org.wso2.siddhi.query.api.expression.Variable;
import org.wso2.siddhi.query.api.expression.constant.IntConstant;
import org.wso2.siddhi.query.api.extension.annotation.SiddhiExtension;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;

@SiddhiExtension(namespace = "custom", function = "lastUnique")
public class CustomWindowExtension extends WindowProcessor {
    String variable = "";
    int variablePosition = 0;
    int noValue = 0;
    Map<Object, InEvent> uniqueWindow = null;
    
    /**
     *This method called when processing an event
     */
    @Override
    protected void processEvent(InEvent inEvent) {
        acquireLock();
        try {
            doProcessing(inEvent);
        } finally
        {
            releaseLock();
        }
    }
    
    /**
     *This method called when processing an event list
     */
	@Override
    protected void processEvent(InListEvent inListEvent) {
        for (int i = 0; i < inListEvent.getActiveEvents(); i++) {
            InEvent inEvent = (InEvent) inListEvent.getEvent(i);
            processEvent(inEvent);
        }
    }
    
    /**
     * This method iterate through the events which are in window
     */
	@Override
    public Iterator<StreamEvent> iterator() {
        return null;
    }
    
    /**
     * This method iterate through the events which are in window but used in distributed processing
     */
	@Override
    public Iterator<StreamEvent> iterator(String s) {
        return null;
    }
    
    /**
     * This method used to return the current state of the window, Used for persistence of data
     */
	@Override
    protected Object[] currentState() {
        return new Object[]{uniqueWindow};
    }
    
    /**
     * This method is used to restore from the persisted state
     */
	@Override
    protected void restoreState(Object[] objects) {
    }
    
    /**
     * Method called when initialising the extension
     */
	@Override
    protected void init(Expression[] expressions,
                        QueryPostProcessingElement queryPostProcessingElement,
                        AbstractDefinition abstractDefinition, String s, boolean b,
                        SiddhiContext siddhiContext) {
        if (expressions.length != 2) {
            log.error("Parameters count is not matching, There should be two parameters ");
        }
        variable = ((Variable) expressions[0]).getAttributeName();
        noValue = ((IntConstant) expressions[1]).getValue();
        uniqueWindow = new LinkedHashMap<Object, InEvent>();
        variablePosition = abstractDefinition.getAttributePosition(variable);
    }
    private void doProcessing(InEvent event) {
        event.getData(variablePosition);
        log.info(event);
        Object eventKey = event.getData(variablePosition);
        if (uniqueWindow.containsKey(eventKey)) {
            InEvent firstEvent = uniqueWindow.remove(eventKey);
            uniqueWindow.put(eventKey, event);
            if (uniqueWindow.size() == noValue) {
                nextProcessor.process(firstEvent);
            }
        } else {
            if (uniqueWindow.size() < noValue) {
                uniqueWindow.put(eventKey, event);
            } else if (uniqueWindow.size() == noValue) {
                Object firstKey = uniqueWindow.keySet().toArray()[0];
                InEvent firstEvent = uniqueWindow.remove(firstKey);
                uniqueWindow.put(eventKey, event);
                nextProcessor.process(firstEvent);
            }
        }
    }
    @Override
    public void destroy() {
    }
}

Sample project file can be downloaded from here.