import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
/**
 * Initialize the mapper and the mapping configurations.
 *
 * @param streamDefinition          The stream definition
 * @param optionHolder              Option holder containing static and dynamic options
 * @param payloadTemplateBuilderMap Unmapped list of payloads for reference
 */
@Override
public void init(StreamDefinition streamDefinition, OptionHolder optionHolder,
                 Map<String, TemplateBuilder> payloadTemplateBuilderMap, ConfigReader mapperConfigReader,
                 SiddhiAppContext siddhiAppContext) {

    attributeNameArray = streamDefinition.getAttributeNameArray();
    enclosingElement = optionHolder.validateAndGetStaticValue(ENCLOSING_ELEMENT_IDENTIFIER, null);
    isJsonValidationEnabled = Boolean.parseBoolean(optionHolder
            .validateAndGetStaticValue(JSON_VALIDATION_IDENTIFIER, "false"));

    //if @payload() is added there must be at least 1 element in it, otherwise a SiddhiParserException raised
    if (payloadTemplateBuilderMap != null && payloadTemplateBuilderMap.size() != 1) {
        throw new SiddhiAppCreationException("Json sink-mapper does not support multiple @payload mappings, " +
                "error at the mapper of '" + streamDefinition.getId() + "'");
    }
    if (payloadTemplateBuilderMap != null &&
            payloadTemplateBuilderMap.get(payloadTemplateBuilderMap.keySet().iterator().next()).isObjectMessage()) {
        throw new SiddhiAppCreationException("Json sink-mapper does not support object @payload mappings, " +
                "error at the mapper of '" + streamDefinition.getId() + "'");
    }
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的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.config.SiddhiAppContext; //导入依赖的package包/类
public final void init(StreamDefinition streamDefinition,
                       String type,
                       OptionHolder mapOptionHolder,
                       List<Element> unmappedPayloadList,
                       Sink sink, ConfigReader mapperConfigReader,
                       LatencyTracker mapperLatencyTracker,
                       SiddhiAppContext siddhiAppContext) {
    this.mapperLatencyTracker = mapperLatencyTracker;
    this.siddhiAppContext = siddhiAppContext;
    sink.setTrpDynamicOptions(trpDynamicOptions);
    this.sinkListener = sink;
    this.optionHolder = mapOptionHolder;
    this.type = type;
    if (unmappedPayloadList != null && !unmappedPayloadList.isEmpty()) {
        templateBuilderMap = new HashMap<>();
        for (Element e : unmappedPayloadList) {
            TemplateBuilder templateBuilder = new TemplateBuilder(streamDefinition, e.getValue());
            if (templateBuilderMap.containsKey(e.getKey())) {
                throw new SiddhiAppCreationException("Duplicate Keys, " + e.getKey() + ", in @payload() ");
            }
            templateBuilderMap.put(e.getKey(), templateBuilder);
        }
    }

    init(streamDefinition, mapOptionHolder, templateBuilderMap, mapperConfigReader, siddhiAppContext);
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
private static ConditionExpressionExecutor generateHavingExecutor(Expression expression,
                                                                  MetaComplexEvent metaComplexEvent,
                                                                  SiddhiAppContext siddhiAppContext,
                                                                  Map<String, Table> tableMap,
                                                                  List<VariableExpressionExecutor>
                                                                          variableExpressionExecutors, String
                                                                          queryName) {
    ConditionExpressionExecutor havingConditionExecutor = null;
    if (expression != null) {
        havingConditionExecutor = (ConditionExpressionExecutor) ExpressionParser.parseExpression(expression,
                metaComplexEvent, SiddhiConstants.HAVING_STATE, tableMap, variableExpressionExecutors,
                siddhiAppContext, false, 0, queryName);

    }
    return havingConditionExecutor;
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
/**
 * Construct a Window object.
 *
 * @param windowDefinition definition of the window
 * @param siddhiAppContext siddhi app context of Siddhi
 */
public Window(WindowDefinition windowDefinition, SiddhiAppContext siddhiAppContext) {
    this.windowDefinition = windowDefinition;
    this.siddhiAppContext = siddhiAppContext;
    this.elementId = siddhiAppContext.getElementIdGenerator().createNewId();
    this.lockWrapper = new LockWrapper(windowDefinition.getId());
    this.lockWrapper.setLock(new ReentrantLock());
    if (siddhiAppContext.getStatisticsManager() != null) {
        latencyTrackerFind = QueryParserHelper.createLatencyTracker(siddhiAppContext, windowDefinition.getId(),
                SiddhiConstants.METRIC_INFIX_WINDOWS, SiddhiConstants.METRIC_TYPE_FIND);
        latencyTrackerInsert = QueryParserHelper.createLatencyTracker(siddhiAppContext, windowDefinition.getId(),
                SiddhiConstants.METRIC_INFIX_WINDOWS, SiddhiConstants.METRIC_TYPE_INSERT);

        throughputTrackerFind = QueryParserHelper.createThroughputTracker(siddhiAppContext,
                windowDefinition.getId(), SiddhiConstants.METRIC_INFIX_WINDOWS, SiddhiConstants.METRIC_TYPE_FIND);
        throughputTrackerInsert = QueryParserHelper.createThroughputTracker(siddhiAppContext,
                windowDefinition.getId(), SiddhiConstants.METRIC_INFIX_WINDOWS, SiddhiConstants.METRIC_TYPE_INSERT);
    }
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
private void createJoinInputStreamExecutors(JoinInputStream inputStream, Partition partition, MetaStateEvent
        metaEvent, List<VariableExpressionExecutor> executors, SiddhiAppContext siddhiAppContext, String
        queryName) {
    createExecutors(inputStream.getLeftInputStream(), partition, metaEvent.getMetaStreamEvent(0), executors,
            siddhiAppContext, queryName);
    int size = executors.size();
    for (VariableExpressionExecutor variableExpressionExecutor : executors) {
        variableExpressionExecutor.getPosition()[SiddhiConstants.STREAM_EVENT_CHAIN_INDEX] = 0;
    }
    createExecutors(inputStream.getRightInputStream(), partition, metaEvent.getMetaStreamEvent(1), executors,
            siddhiAppContext, queryName);
    for (int i = size; i < executors.size(); i++) {
        executors.get(i).getPosition()[SiddhiConstants.STREAM_EVENT_CHAIN_INDEX] = 1;
    }

}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
public void initAggregator(ExpressionExecutor[] attributeExpressionExecutors, SiddhiAppContext
            siddhiAppContext, ConfigReader configReader) {
        this.configReader = configReader;
        try {
            this.siddhiAppContext = siddhiAppContext;
            this.attributeExpressionExecutors = attributeExpressionExecutors;
            this.attributeSize = attributeExpressionExecutors.length;
            if (elementId == null) {
                elementId = "AttributeAggregator-" + siddhiAppContext.getElementIdGenerator().createNewId();
            }
            //Not added to Snapshotable as the AggregationAttributeExecutors are added
//            siddhiAppContext.getSnapshotService().addSnapshotable(this);
            init(attributeExpressionExecutors, configReader, siddhiAppContext);
        } catch (Throwable t) {
            throw new SiddhiAppCreationException(t);
        }
    }
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
public PartitionRuntime(ConcurrentMap<String, AbstractDefinition> streamDefinitionMap, ConcurrentMap<String,
        StreamJunction> streamJunctionMap, Partition partition, SiddhiAppContext siddhiAppContext) {
    this.siddhiAppContext = siddhiAppContext;
    if (partition.getPartitionTypeMap().isEmpty()) {
        throw new SiddhiAppCreationException("Partition must have at least one executor. But found none.");
    }
    try {
        Element element = AnnotationHelper.getAnnotationElement("info", "name",
                partition.getAnnotations());
        if (element != null) {
            this.partitionId = element.getValue();
        }
    } catch (DuplicateAnnotationException e) {
        throw new DuplicateAnnotationException(e.getMessageWithOutContext() + " for the same Query " +
                partition.toString(), e, e.getQueryContextStartIndex(), e.getQueryContextEndIndex(),
                siddhiAppContext.getName(), siddhiAppContext.getSiddhiAppString());
    }
    if (partitionId == null) {
        this.partitionId = UUID.randomUUID().toString();
    }
    elementId = "PartitionRuntime-" + siddhiAppContext.getElementIdGenerator().createNewId();
    this.partition = partition;
    this.streamDefinitionMap = streamDefinitionMap;
    this.streamJunctionMap = streamJunctionMap;
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
@Override
public void initTransport(OptionHolder sinkOptionHolder, List<OptionHolder> destinationOptionHolders, Annotation
        sinkAnnotation, ConfigReader sinkConfigReader, SiddhiAppContext siddhiAppContext) {
    String transportType = sinkOptionHolder.validateAndGetStaticValue(SiddhiConstants.ANNOTATION_ELEMENT_TYPE);
    Extension sinkExtension = DefinitionParserHelper.constructExtension
            (streamDefinition, SiddhiConstants.ANNOTATION_SINK, transportType, sinkAnnotation, SiddhiConstants
                    .NAMESPACE_SINK);

    destinationOptionHolders.forEach(destinationOption -> {
        Sink sink = (Sink) SiddhiClassLoader.loadExtensionImplementation(
                sinkExtension, SinkExecutorExtensionHolder.getInstance(siddhiAppContext));
        destinationOption.merge(sinkOptionHolder);
        sink.initOnlyTransport(streamDefinition, destinationOption, sinkConfigReader, siddhiAppContext);
        transports.add(sink);
    });
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
/**
 * {@inheritDoc}
 */
@Override
public CompiledCondition compileCondition(Expression condition, MatchingMetaInfoHolder matchingMetaInfoHolder,
                                          SiddhiAppContext siddhiAppContext,
                                          List<VariableExpressionExecutor> variableExpressionExecutors,
                                          Map<String, Table> tableMap, String queryName) {
    if (this.internalWindowProcessor instanceof FindableProcessor) {
        return ((FindableProcessor) this.internalWindowProcessor).compileCondition(condition,
                matchingMetaInfoHolder, siddhiAppContext, variableExpressionExecutors, tableMap,
                queryName);
    } else {
        throw new OperationNotSupportedException("Cannot construct finder for the window " + this
                .windowDefinition.getWindow());
    }

}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
public void initExecutor(ExpressionExecutor[] attributeExpressionExecutors, SiddhiAppContext
        siddhiAppContext, String queryName, ConfigReader configReader) {
    this.configReader = configReader;
    try {
        this.siddhiAppContext = siddhiAppContext;
        this.attributeExpressionExecutors = attributeExpressionExecutors;
        attributeSize = attributeExpressionExecutors.length;
        this.queryName = queryName;
        if (elementId == null) {
            elementId = "FunctionExecutor-" + siddhiAppContext.getElementIdGenerator().createNewId();
        }
        siddhiAppContext.getSnapshotService().addSnapshotable(queryName, this);
        init(attributeExpressionExecutors, configReader, siddhiAppContext);
    } catch (Throwable t) {
        throw new SiddhiAppCreationException(t);
    }
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
/**
 * Creates an object size calculator that can calculate object sizes for a given
 * {@code memoryLayoutSpecification}.
 *
 * @param memoryLayoutSpecification a description of the JVM memory layout.
 */
public ObjectSizeCalculator(MemoryLayoutSpecification memoryLayoutSpecification) {
    Preconditions.checkNotNull(memoryLayoutSpecification);
    arrayHeaderSize = memoryLayoutSpecification.getArrayHeaderSize();
    objectHeaderSize = memoryLayoutSpecification.getObjectHeaderSize();
    objectPadding = memoryLayoutSpecification.getObjectPadding();
    referenceSize = memoryLayoutSpecification.getReferenceSize();
    superclassFieldPadding = memoryLayoutSpecification.getSuperclassFieldPadding();
    ignoreCalculation.add(SiddhiAppContext.class);
    ignoreCalculation.add(StreamJunction.class);
    ignoreCalculation.add(OutputCallback.class);
    ignoreCalculation.add(StreamCallback.class);
    ignoreCalculation.add(ThroughputTracker.class);
    ignoreCalculation.add(LatencyTracker.class);
    ignoreCalculation.add(MemoryUsageTracker.class);
    ignoreCalculation.add(BufferedEventsTracker.class);
    ignoreCalculation.add(StatisticsManager.class);
    ignoreCalculation.add(MemoryCalculable.class);
    ignoreCalculation.add(AbstractRecordTable.class);
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
protected AbstractExtensionHolder(Class clazz, SiddhiAppContext siddhiAppContext) {
    Map<String, Class> extensions = siddhiAppContext.getSiddhiContext().getSiddhiExtensions();
    if (extensions != null) {
        for (String extensionKey : siddhiAppContext.getSiddhiContext().getSiddhiExtensions().keySet()) {
            Class extension = extensions.get(extensionKey);
            if (clazz.isAssignableFrom(extension)) {
                if (extensionMap.containsKey(extensionKey)) {
                    log.error("Extension class " + extension.getName() + " not loaded, as there is already an" +
                                      " matching extension '" + extensionKey + "' implemented as " + extensionMap
                            .get
                                    (extensionKey).getName());
                } else {
                    extensionMap.put(extensionKey, extension);
                }

            }

        }
    }
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
@Override
public void init(StreamDefinition streamDefinition, OptionHolder optionHolder, List<AttributeMapping> list,
                 ConfigReader configReader, SiddhiAppContext siddhiAppContext) {

    this.streamDefinition = streamDefinition;
    this.streamAttributes = this.streamDefinition.getAttributeList();
    attributesSize = this.streamDefinition.getAttributeList().size();
    this.mappingPositions = new MappingPositionData[attributesSize];
    failOnMissingAttribute = Boolean.parseBoolean(optionHolder.
            validateAndGetStaticValue(FAIL_ON_MISSING_ATTRIBUTE_IDENTIFIER, "true"));
    factory = new JsonFactory();
    if (list != null && list.size() > 0) {
        isCustomMappingEnabled = true;
        enclosingElement = optionHolder.validateAndGetStaticValue(ENCLOSING_ELEMENT_IDENTIFIER,
                DEFAULT_ENCLOSING_ELEMENT);
        for (int i = 0; i < list.size(); i++) {
            AttributeMapping attributeMapping = list.get(i);
            String attributeName = attributeMapping.getName();
            int position;
            if (attributeName != null) {
                position = this.streamDefinition.getAttributePosition(attributeName);
            } else {
                position = i;
            }
            this.mappingPositions[i] = new MappingPositionData(position, attributeMapping.getMapping());
        }
    } else {
        for (int i = 0; i < attributesSize; i++) {
            this.mappingPositions[i] = new MappingPositionData(i, DEFAULT_JSON_MAPPING_PREFIX + this
                    .streamDefinition.getAttributeList().get(i).getName());
        }
    }
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的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.config.SiddhiAppContext; //导入依赖的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.config.SiddhiAppContext; //导入依赖的package包/类
public StreamJunction(StreamDefinition streamDefinition, ExecutorService executorService, int bufferSize,
                      SiddhiAppContext siddhiAppContext) {
    this.streamDefinition = streamDefinition;
    this.bufferSize = bufferSize;
    this.executorService = executorService;
    this.siddhiAppContext = siddhiAppContext;
    if (siddhiAppContext.getStatisticsManager() != null) {
        this.throughputTracker = QueryParserHelper.createThroughputTracker(siddhiAppContext,
                streamDefinition.getId(),
                SiddhiConstants.METRIC_INFIX_STREAMS, null);
    }
    try {
        Annotation annotation = AnnotationHelper.getAnnotation(SiddhiConstants.ANNOTATION_ASYNC,
                streamDefinition.getAnnotations());
        if (annotation != null) {
            async = true;
            String bufferSizeString = annotation.getElement(SiddhiConstants.ANNOTATION_ELEMENT_BUFFER_SIZE);
            if (bufferSizeString != null) {
                this.bufferSize = Integer.parseInt(bufferSizeString);
            }
        }

    } catch (DuplicateAnnotationException e) {
        throw new DuplicateAnnotationException(e.getMessageWithOutContext() + " for the same Stream " +
                streamDefinition.getId(), e, e.getQueryContextStartIndex(), e.getQueryContextEndIndex(),
                siddhiAppContext.getName(), siddhiAppContext.getSiddhiAppString());
    }
    isTraceEnabled = log.isTraceEnabled();
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
public IncrementalDataAggregator(List<TimePeriod.Duration> incrementalDurations,
                                 TimePeriod.Duration aggregateForDuration, List<ExpressionExecutor> baseExecutors,
                                 ExpressionExecutor timestampExecutor, MetaStreamEvent metaStreamEvent,
                                 SiddhiAppContext siddhiAppContext) {
    this.incrementalDurations = incrementalDurations;
    this.aggregateForDuration = aggregateForDuration;
    this.timestampExecutor = timestampExecutor;
    this.timeZoneExecutor = baseExecutors.get(0);
    StreamEventPool streamEventPool = new StreamEventPool(metaStreamEvent, 10);
    this.baseIncrementalValueStore = new BaseIncrementalValueStore(-1, baseExecutors, streamEventPool,
            siddhiAppContext, null);
    this.baseIncrementalValueStoreMap = new HashMap<>();
    this.baseIncrementalValueGroupByStoreMap = new HashMap<>();
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
public static StreamProcessorExtensionHolder getInstance(SiddhiAppContext siddhiAppContext) {
    ConcurrentHashMap<Class, AbstractExtensionHolder> extensionHolderMap = siddhiAppContext.getSiddhiContext
            ().getExtensionHolderMap();
    AbstractExtensionHolder abstractExtensionHolder = extensionHolderMap.get(clazz);
    if (abstractExtensionHolder == null) {
        abstractExtensionHolder = new StreamProcessorExtensionHolder(siddhiAppContext);
        extensionHolderMap.putIfAbsent(clazz, abstractExtensionHolder);
    }
    return (StreamProcessorExtensionHolder) extensionHolderMap.get(clazz);
}
 

import org.wso2.siddhi.core.config.SiddhiAppContext; //导入依赖的package包/类
@Override
protected void init(ExpressionExecutor[] attributeExpressionExecutors, ConfigReader configReader,
                    SiddhiAppContext siddhiAppContext) {
    if (!(attributeExpressionExecutors.length == 0 || attributeExpressionExecutors.length == 1)) {
        throw new SiddhiAppValidationException("incrementalAggregator:getTimeZone() function " +
                "accepts zero or one argument, but found " + attributeExpressionExecutors.length);
    }
    if (attributeExpressionExecutors.length == 1) {
        if (attributeExpressionExecutors[0].getReturnType() != Attribute.Type.STRING) {
            throw new SiddhiAppValidationException("Time zone can be retrieved, only from " +
                    "string values, but found " +  attributeExpressionExecutors[0].getReturnType());
        }
    }
}