To write a custom Window, create a class extending "org.wso2.siddhi.core.query.processor.stream.window.WindowProcessor" and create an appropriate .siddhiext extension mapping file, compile the class, and build the jar containing the .class and .siddhiext files. Add them to the Siddhi class path. In the case of running them on WSO2 DAS add the jar to <DAS_HOME>/repository/components/lib.
For example, Window extension created with namespace "custom" and function name "lastUnique" can be referred in the query as follows:
from StockExchangeStream[price >= 20]#window.custom:lastUnique(symbol,5) select symbol, price insert into StockQuote;
For the Window extension to be used in a Join Query the Window Extension should be findable. To make the Window findable it should implement the "org.wso2.siddhi.core.query.processor.stream.window.FindableProcessor" Interface.
E.g. Implementation can be found below;
/*
* Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.wso2.siddhi.core.query.processor.stream.window;
import org.wso2.siddhi.core.config.ExecutionPlanContext;
import org.wso2.siddhi.core.event.ComplexEvent;
import org.wso2.siddhi.core.event.ComplexEventChunk;
import org.wso2.siddhi.core.event.MetaComplexEvent;
import org.wso2.siddhi.core.event.stream.StreamEvent;
import org.wso2.siddhi.core.event.stream.StreamEventCloner;
import org.wso2.siddhi.core.executor.ExpressionExecutor;
import org.wso2.siddhi.core.executor.VariableExpressionExecutor;
import org.wso2.siddhi.core.query.processor.Processor;
import org.wso2.siddhi.core.table.EventTable;
import org.wso2.siddhi.core.util.collection.operator.Finder;
import org.wso2.siddhi.core.util.parser.CollectionOperatorParser;
import org.wso2.siddhi.query.api.expression.Expression;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class LastUniqueWindowProcessor extends WindowProcessor implements FindableProcessor{
private ConcurrentHashMap<String, StreamEvent> map = new ConcurrentHashMap<String, StreamEvent>();
private VariableExpressionExecutor[] variableExpressionExecutors;
/**
* The init method of the WindowProcessor, this method will be called before other methods
*
* @param attributeExpressionExecutors the executors of each function parameters
* @param executionPlanContext the context of the execution plan
*/
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ExecutionPlanContext executionPlanContext) {
variableExpressionExecutors = new VariableExpressionExecutor[attributeExpressionExecutors.length];
for (int i = 0; i < attributeExpressionExecutors.length; i++) {
variableExpressionExecutors[i] =(VariableExpressionExecutor) attributeExpressionExecutors[i];
}
}
/**
* The main processing method that will be called upon event arrival
*
* @param streamEventChunk the stream event chunk that need to be processed
* @param nextProcessor the next processor to which the success events need to be passed
* @param streamEventCloner helps to clone the incoming event for local storage or modification
*/
@Override
protected synchronized void process(ComplexEventChunk<StreamEvent> streamEventChunk, Processor nextProcessor, StreamEventCloner streamEventCloner) {
ComplexEventChunk<StreamEvent> complexEventChunk = new ComplexEventChunk<StreamEvent>();
StreamEvent streamEvent = streamEventChunk.getFirst();
while (streamEvent != null) {
StreamEvent clonedEvent = streamEventCloner.copyStreamEvent(streamEvent);
clonedEvent.setType(StreamEvent.Type.EXPIRED);
StreamEvent oldEvent = map.put(generateKey(clonedEvent), clonedEvent);
if (oldEvent != null) {
complexEventChunk.add(oldEvent);
}
StreamEvent next = streamEvent.getNext();
streamEvent.setNext(null);
complexEventChunk.add(streamEvent);
streamEvent = next;
}
nextProcessor.process(complexEventChunk);
}
/**
* This will be called only once and this can be used to acquire
* required resources for the processing element.
* This will be called after initializing the system and before
* starting to process the events.
*/
@Override
public void start() {
//Do nothing
}
/**
* This will be called only once and this can be used to release
* the acquired resources for processing.
* This will be called before shutting down the system.
*/
@Override
public void stop() {
//Do nothing
}
/**
* Used to collect the serializable state of the processing element, that need to be
* persisted for the reconstructing the element to the same state on a different point of time
*
* @return stateful objects of the processing element as an array
*/
@Override
public Object[] currentState() {
return new Object[]{map};
}
/**
* Used to restore serialized state of the processing element, for reconstructing
* the element to the same state as if was on a previous point of time.
*
* @param state the stateful objects of the element as an array on
* the same order provided by currentState().
*/
@Override
public void restoreState(Object[] state) {
map = (ConcurrentHashMap<String, StreamEvent>) state[0];
}
/**
* To find events from the processor event pool, that the matches the matchingEvent based on finder logic.
*
* @param matchingEvent the event to be matched with the events at the processor
* @param finder the execution element responsible for finding the corresponding events that matches
* the matchingEvent based on pool of events at Processor
* @return the matched events
*/
@Override
public synchronized StreamEvent find(ComplexEvent matchingEvent, Finder finder) {
return finder.find(matchingEvent, map.values(),streamEventCloner);
}
/**
* To construct a finder having the capability of finding events at the processor that corresponds to the incoming
* matchingEvent and the given matching expression logic.
*
* @param expression the matching expression
* @param metaComplexEvent the meta structure of the incoming matchingEvent
* @param executionPlanContext current execution plan context
* @param variableExpressionExecutors the list of variable ExpressionExecutors already created
* @param eventTableMap map of event tables
* @param matchingStreamIndex the stream index of the incoming matchingEvent
* @param withinTime the maximum time gap between the events to be matched
* @return finder having the capability of finding events at the processor against the expression and incoming
* matchingEvent
*/
@Override
public Finder constructFinder(Expression expression, MetaComplexEvent metaComplexEvent, ExecutionPlanContext executionPlanContext, List<VariableExpressionExecutor> variableExpressionExecutors, Map<String, EventTable> eventTableMap, int matchingStreamIndex, long withinTime) {
return CollectionOperatorParser.parse(expression, metaComplexEvent, executionPlanContext, variableExpressionExecutors, eventTableMap, matchingStreamIndex, inputDefinition, withinTime);
}
private String generateKey(StreamEvent event) {
StringBuilder stringBuilder = new StringBuilder();
for (VariableExpressionExecutor executor : variableExpressionExecutors) {
stringBuilder.append(event.getAttribute(executor.getPosition()));
}
return stringBuilder.toString();
}
}