import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
/**
 * The initialization method for FunctionExecutor, this method will be called before the other methods
 *  @param attributeExpressionExecutors are the executors of each function parameters
 * @param configReader
 * @param siddhiAppContext         the context of the siddhi app
 */
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader,
                    SiddhiAppContext siddhiAppContext) {
    for (ExpressionExecutor expressionExecutor : attributeExpressionExecutors) {
        Attribute.Type attributeType = expressionExecutor.getReturnType();
        if (attributeType == Attribute.Type.DOUBLE) {
            returnType = attributeType;

        } else if ((attributeType == Attribute.Type.STRING) || (attributeType == Attribute.Type.BOOL)) {
            throw new SiddhiAppCreationException("Plus cannot have parameters with types String or Bool");
        } else {
            returnType = Attribute.Type.LONG;
        }
    }
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
/**
 * The initialization method for FunctionExecutor
 *
 * @param attributeExpressionExecutors are the executors of each attributes in the function
 * @param configReader                 this hold the {@link MaxAttributeAggregator} configuration reader.
 * @param siddhiAppContext             Siddhi app runtime context
 */
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader,
                    SiddhiAppContext siddhiAppContext) {
    if (attributeExpressionExecutors.length != 1) {
        throw new OperationNotSupportedException("Max aggregator has to have exactly 1 parameter, currently " +
                attributeExpressionExecutors.length + " parameters provided");
    }
    Attribute.Type type = attributeExpressionExecutors[0].getReturnType();
    switch (type) {
        case FLOAT:
            maxOutputAttributeAggregator = new MaxAttributeAggregatorFloat();
            break;
        case INT:
            maxOutputAttributeAggregator = new MaxAttributeAggregatorInt();
            break;
        case LONG:
            maxOutputAttributeAggregator = new MaxAttributeAggregatorLong();
            break;
        case DOUBLE:
            maxOutputAttributeAggregator = new MaxAttributeAggregatorDouble();
            break;
        default:
            throw new OperationNotSupportedException("Max not supported for " + type);
    }
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
/**
 * The initialization method for FunctionExecutor
 *
 * @param attributeExpressionExecutors are the executors of each attributes in the function
 * @param configReader                 this hold the {@link MinAttributeAggregator} configuration reader.
 * @param siddhiAppContext             Siddhi app runtime context
 */
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader,
                    SiddhiAppContext siddhiAppContext) {
    if (attributeExpressionExecutors.length != 1) {
        throw new OperationNotSupportedException("Min aggregator has to have exactly 1 parameter, currently " +
                attributeExpressionExecutors.length + " parameters provided");
    }
    Attribute.Type type = attributeExpressionExecutors[0].getReturnType();
    switch (type) {
        case FLOAT:
            minOutputAttributeAggregator = new MinAttributeAggregatorFloat();
            break;
        case INT:
            minOutputAttributeAggregator = new MinAttributeAggregatorInt();
            break;
        case LONG:
            minOutputAttributeAggregator = new MinAttributeAggregatorLong();
            break;
        case DOUBLE:
            minOutputAttributeAggregator = new MinAttributeAggregatorDouble();
            break;
        default:
            throw new OperationNotSupportedException("Min not supported for " + type);
    }
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
@Override
public void init(List<ExpressionExecutor> attributeExpressionExecutors, SiddhiContext siddhiContext) {
    if (attributeSize != 2) {
        throw new OperationNotSupportedException("IsMatch has to have 2 expressions regex and the attribute, " +
                "currently " + attributeSize + " expressions provided");
    }
    ExpressionExecutor regexExecutor = attributeExpressionExecutors.get(0);
    if (regexExecutor.getReturnType() != Attribute.Type.STRING &&
            regexExecutor instanceof ConstantExpressionExecutor) {
        throw new OperationNotSupportedException("IsMatch expects regex string input expression but found " +
                regexExecutor.getReturnType());
    }
    expressionExecutor = attributeExpressionExecutors.get(1);

    pattern = Pattern.compile((String) regexExecutor.execute(null));

}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
@Override
public Processor cloneProcessor(String key) {
    try {
        AggregateWindowProcessor streamProcessor = new AggregateWindowProcessor(aggregationRuntime, within, per);
        streamProcessor.inputDefinition = inputDefinition;
        ExpressionExecutor[] innerExpressionExecutors = new ExpressionExecutor[attributeExpressionLength];
        ExpressionExecutor[] attributeExpressionExecutors1 = this.attributeExpressionExecutors;
        for (int i = 0; i < attributeExpressionLength; i++) {
            innerExpressionExecutors[i] = attributeExpressionExecutors1[i].cloneExecutor(key);
        }
        streamProcessor.attributeExpressionExecutors = innerExpressionExecutors;
        streamProcessor.attributeExpressionLength = attributeExpressionLength;
        streamProcessor.additionalAttributes = additionalAttributes;
        streamProcessor.complexEventPopulater = complexEventPopulater;
        streamProcessor.init(inputDefinition, attributeExpressionExecutors, configReader, siddhiAppContext,
                outputExpectsExpiredEvents);
        streamProcessor.start();
        return streamProcessor;

    } catch (Exception e) {
        throw new SiddhiAppRuntimeException("Exception in cloning " + this.getClass().getCanonicalName(), e);
    }
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader, boolean
        outputExpectsExpiredEvents, SiddhiAppContext siddhiAppContext) {
    support = Double.parseDouble(String.valueOf(((ConstantExpressionExecutor) attributeExpressionExecutors[0])
            .getValue()));
    if (attributeExpressionExecutors.length > 1) {
        error = Double.parseDouble(String.valueOf(((ConstantExpressionExecutor) attributeExpressionExecutors[1])
                .getValue()));
    } else {
        error = support / 10; // recommended error is 10% of 20$ of support value;
    }
    if ((support > 1 || support < 0) || (error > 1 || error < 0)) {
        log.error("Wrong argument has provided, Error executing the window");
    }
    variableExpressionExecutors = new VariableExpressionExecutor[attributeExpressionExecutors.length - 2];
    if (attributeExpressionExecutors.length > 2) {  // by-default all the attributes will be compared
        for (int i = 2; i < attributeExpressionExecutors.length; i++) {
            variableExpressionExecutors[i - 2] = (VariableExpressionExecutor) attributeExpressionExecutors[i];
        }
    }
    windowWidth = Math.ceil(1 / error);
    currentBucketId = 1;
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
private void buildStreamVariableExecutor(Variable variable, int streamEventChainIndex,
                                         ExpressionVisitor expressionVisitor, Attribute.Type type) {
    String id = variable.getAttributeName();
    if (variable.getStreamId() != null) {
        id = variable.getStreamId() + "." + id;
    }
    expressionVisitor.beginVisitStreamVariable(id, variable.getStreamId(), variable.getAttributeName(), type);
    if (!variableExpressionExecutorMap.containsKey(id)) {
        ExpressionExecutor variableExpressionExecutor = ExpressionParser.parseExpression(
                variable, matchingMetaInfoHolder.getMetaStateEvent(), streamEventChainIndex, tableMap,
                variableExpressionExecutors, siddhiAppContext, false, 0, queryName);
        variableExpressionExecutorMap.put(id, variableExpressionExecutor);
    }
    expressionVisitor.endVisitStreamVariable(id, variable.getStreamId(), variable.getAttributeName(), type);

}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
private ComplexEventChunk<StreamEvent> createAggregateSelectionEventChunk(
        ComplexEventChunk<StreamEvent> complexEventChunkToHoldMatches,
        List<ExpressionExecutor> outputExpressionExecutors) {
    ComplexEventChunk<StreamEvent> aggregateSelectionComplexEventChunk = new ComplexEventChunk<>(true);
    StreamEvent resetEvent = streamEventPoolForTableMeta.borrowEvent();
    resetEvent.setType(ComplexEvent.Type.RESET);

    while (complexEventChunkToHoldMatches.hasNext()) {
        StreamEvent streamEvent = complexEventChunkToHoldMatches.next();
        StreamEvent newStreamEvent = streamEventPoolForAggregateMeta.borrowEvent();
        Object outputData[] = new Object[newStreamEvent.getOutputData().length];
        for (int i = 0; i < outputExpressionExecutors.size(); i++) {
            outputData[i] = outputExpressionExecutors.get(i).execute(streamEvent);
        }
        newStreamEvent.setTimestamp(streamEvent.getTimestamp());
        newStreamEvent.setOutputData(outputData);
        aggregateSelectionComplexEventChunk.add(newStreamEvent);
    }

    for (ExpressionExecutor expressionExecutor : outputExpressionExecutors) {
        expressionExecutor.execute(resetEvent);
    }

    return aggregateSelectionComplexEventChunk;
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
/**
 * @param inputDefinition
 * @param attributeExpressionExecutors
 * @param startIndex                   starting index
 * @param noOfFeatures
 * @return
 */
public static List<VariableExpressionExecutor> extractAndValidateFeatures(
        AbstractDefinition inputDefinition, ExpressionExecutor[]
        attributeExpressionExecutors,
        int startIndex, int noOfFeatures) {

    List<VariableExpressionExecutor> featureVariableExpressionExecutors = new ArrayList<>();

    // feature values start
    for (int i = startIndex; i < (startIndex + noOfFeatures); i++) {
        if (attributeExpressionExecutors[i] instanceof VariableExpressionExecutor) {
            featureVariableExpressionExecutors.add((VariableExpressionExecutor)
                    attributeExpressionExecutors[i]);
            // other attributes should be numeric type.
            String attributeName = ((VariableExpressionExecutor)
                    attributeExpressionExecutors[i]).getAttribute().getName();
            Attribute.Type featureAttributeType = inputDefinition.
                    getAttributeType(attributeName);

            //feature attributes not numerical type
            if (!isNumeric(featureAttributeType)) {
                throw new SiddhiAppValidationException("model.features in " + (i + 1) + "th parameter is not "
                        + "a numerical type attribute. Found " + attributeExpressionExecutors[i].getReturnType()
                        + ". Check the input stream definition.");
            }
        } else {
            throw new SiddhiAppValidationException((i + 1) + "th parameter is not " +
                    "an attribute (VariableExpressionExecutor) present in the stream definition. Found a "
                    + attributeExpressionExecutors[i].getClass().getCanonicalName());
        }
    }
    return featureVariableExpressionExecutors;
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
/**
 * The initialization method for FunctionExecutor, this method will be called before the other methods
 */
@Override
protected void init(ExpressionExecutor[] expressionExecutors, ConfigReader configReader, SiddhiAppContext siddhiAppContext) {
	for (ExpressionExecutor expressionExecutor : attributeExpressionExecutors) {
		Attribute.Type attributeType = expressionExecutor.getReturnType();
		if (attributeType == Attribute.Type.DOUBLE) {
			returnType = attributeType;

		} else if ((attributeType == Attribute.Type.STRING) || (attributeType == Attribute.Type.BOOL)) {
			throw new SiddhiAppCreationException("Plus cannot have parameters with types String or Bool");
		} else {
			returnType = Attribute.Type.LONG;
		}
	}
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
/**
 * The initialization method for FunctionExecutor, this method will be called before the other methods
 */
@Override
protected void init(ExpressionExecutor[] expressionExecutors, ConfigReader configReader, SiddhiAppContext siddhiAppContext) {
    for (ExpressionExecutor expressionExecutor : attributeExpressionExecutors) {
        Attribute.Type attributeType = expressionExecutor.getReturnType();
        if (attributeType == Attribute.Type.DOUBLE) {
            returnType = attributeType;

        } else if ((attributeType == Attribute.Type.STRING) || (attributeType == Attribute.Type.BOOL)) {
            throw new SiddhiAppCreationException("Plus cannot have parameters with types String or Bool");
        } else {
            returnType = Attribute.Type.LONG;
        }
    }
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader,
                    SiddhiAppContext siddhiAppContext) {
    if (attributeExpressionExecutors.length != 1) {
        throw new SiddhiAppValidationException("Invalid no of arguments passed to instanceOfString() " +
                "function, required only 1, but found " + attributeExpressionExecutors.length);
    }
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
/**
 * The initialization method for FunctionExecutor
 *
 * @param attributeExpressionExecutors are the executors of each attributes in the function
 * @param configReader                 this hold the {@link SumAttributeAggregator} configuration reader.
 * @param siddhiAppContext             Siddhi app runtime context
 */
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader,
                    SiddhiAppContext siddhiAppContext) {
    if (attributeExpressionExecutors.length != 1) {
        throw new OperationNotSupportedException("Sum aggregator has to have exactly 1 parameter, currently " +
                attributeExpressionExecutors.length
                + " parameters provided");
    }
    Attribute.Type type = attributeExpressionExecutors[0].getReturnType();
    switch (type) {
        case FLOAT:
            sumOutputAttributeAggregator = new SumAttributeAggregatorFloat();
            break;
        case INT:
            sumOutputAttributeAggregator = new SumAttributeAggregatorInt();
            break;
        case LONG:
            sumOutputAttributeAggregator = new SumAttributeAggregatorLong();
            break;
        case DOUBLE:
            sumOutputAttributeAggregator = new SumAttributeAggregatorDouble();
            break;
        default:
            throw new OperationNotSupportedException("Sum not supported for " + type);
    }

}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
@Test(expected = OperationNotSupportedException.class)
public void testConditionExpressionExecutorValidation() {
    StreamDefinition streamDefinition = StreamDefinition.id("cseEventStream").attribute("symbol", Attribute.Type.STRING).attribute("price", Attribute.Type.FLOAT).attribute("volume", Attribute.Type.INT);

    VariableExpressionExecutor volumeVariableExpressionExecutor = new VariableExpressionExecutor("volume", streamDefinition);
    ConstantExpressionExecutor constantExpressionExecutor = new ConstantExpressionExecutor(10f, Attribute.Type.FLOAT);
    ExpressionExecutor compareGreaterThanExecutor = new GreaterThanCompareConditionExpressionExecutorIntInt(new ConstantExpressionExecutor(10, Attribute.Type.INT), volumeVariableExpressionExecutor);
    ExpressionExecutor andExecutor = new AndConditionExpressionExecutor(constantExpressionExecutor, compareGreaterThanExecutor);
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader, boolean
        outputExpectsExpiredEvents, SiddhiAppContext siddhiAppContext) {
    // nothing to be done
    this.configReader = configReader;
    this.outputExpectsExpiredEvents = outputExpectsExpiredEvents;
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader,
                    boolean outputExpectsExpiredEvents, SiddhiAppContext siddhiAppContext) {
    // nothing to be done
    this.configReader = configReader;
    this.outputExpectsExpiredEvents = outputExpectsExpiredEvents;
    this.siddhiAppContext = siddhiAppContext;
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader, boolean
        outputExpectsExpiredEvents, SiddhiAppContext siddhiAppContext) {
    this.outputExpectsExpiredEvents = outputExpectsExpiredEvents;
    this.siddhiAppContext = siddhiAppContext;
    if (outputExpectsExpiredEvents) {
        expiredEventChunk = new ComplexEventChunk<StreamEvent>(false);
    }
    if (attributeExpressionExecutors.length == 1) {
        length = (Integer) (((ConstantExpressionExecutor) attributeExpressionExecutors[0]).getValue());
    } else {
        throw new SiddhiAppValidationException("Length batch window should only have one parameter (<int> " +
                "windowLength), but found " + attributeExpressionExecutors.length + " input attributes");
    }
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader,
                    SiddhiAppContext siddhiAppContext) {
    if (attributeExpressionExecutors.length != 1) {
        throw new SiddhiAppValidationException("Invalid no of arguments passed to instanceOfInteger() " +
                "function, required only 1, but found " + attributeExpressionExecutors.length);
    }
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
public FilterProcessor(ExpressionExecutor conditionExecutor) {
    if (Attribute.Type.BOOL.equals(conditionExecutor.getReturnType())) {
        this.conditionExecutor = conditionExecutor;
    } else {
        throw new OperationNotSupportedException("Return type of " + conditionExecutor.toString() + " should be " +
                "of type BOOL. " +
                "Actual type: " + conditionExecutor.getReturnType().toString());
    }
}
 

import org.wso2.siddhi.core.executor.ExpressionExecutor; //导入依赖的package包/类
private void process(StreamEvent streamEvent, BaseIncrementalValueStore baseIncrementalValueStore) {
    List<ExpressionExecutor> expressionExecutors = baseIncrementalValueStore.getExpressionExecutors();
    for (int i = 0; i < expressionExecutors.size(); i++) { // keeping timestamp value location as null
        ExpressionExecutor expressionExecutor = expressionExecutors.get(i);
        baseIncrementalValueStore.setValue(expressionExecutor.execute(streamEvent), i + 1);
    }
    baseIncrementalValueStore.setProcessed(true);
}