import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的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.query.api.definition.StreamDefinition; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * <p/>
 * <p>The default implementation returns the result of calling
 * {@link #visitChildren} on {@code ctx}.</p>
 *
 * @param ctx
 */
@Override
public Object visitDefinitionStream(@NotNull SiddhiQLGrammarParser.DefinitionStreamContext ctx) {

    StreamDefinition streamDefinition = QueryFactory.createStreamDefinition();
    streamDefinition.name(ctx.source().getText());

    List<Attribute> attributeList = (List<Attribute>) visitDefinition(ctx.definition());
    if (attributeList != null) {
        for (Attribute attribute : attributeList) {
            streamDefinition.attribute(attribute.getName(), attribute.getType());
        }
    } else {
        throw new SiddhiParserException("Operation not supported :fromSource and likeSource");
    }
    return streamDefinition;
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
@Override
public List<QueryEventSource> constructQueryEventSourceList(
        ConcurrentMap<String, AbstractDefinition> streamTableDefinitionMap,
        List<QueryEventSource> queryEventSources) {
    AbstractDefinition definition = streamTableDefinitionMap.get(streamId);
    if (definition == null) {
        throw new SourceNotExistException("Stream definition not exist! No steam defined with stream ID: " + streamId);
    }
    if (definition instanceof TableDefinition) {
        throw new SourceNotExistException(streamId + " is not a Stream but a Table, and it cant have window");
    }
    streamDefinition = (StreamDefinition) definition;
    queryEventSource = new QueryEventSource(streamId, streamReferenceId,
                                            streamDefinition,
                                            filter, transformer, window);

    queryEventSources.add(queryEventSource);
    return queryEventSources;
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
public ExecutionPlanReference addExecutionPlan(String executionPlan) throws SiddhiParserException {
    List<ExecutionPlan> executionPlanList = SiddhiCompiler.parse(executionPlan);
    ExecutionPlanReference executionPlanReference = new ExecutionPlanReference();
    for (ExecutionPlan plan : executionPlanList) {
        if (plan instanceof Query) {
            executionPlanReference.addQueryReference(addQuery((Query) plan));
        } else if (plan instanceof StreamDefinition) {
            executionPlanReference.addInputHandler(defineStream((StreamDefinition) plan));
        } else if (plan instanceof TableDefinition) {
            defineTable((TableDefinition) plan);
        } else if (plan instanceof PartitionDefinition) {
            definePartition((PartitionDefinition) plan);
        } else {
            throw new OperationNotSupportedException(plan.getClass().getName() + " is not supported as an execution plan element ");
        }
    }

    return executionPlanReference;
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
public static StreamDefinition parseStreamDefinition(String source) throws SiddhiParserException {
    try {
        SiddhiQLGrammarLexer lexer = new SiddhiQLGrammarLexer();
        lexer.setCharStream(new ANTLRStringStream(source));
        CommonTokenStream tokens = new CommonTokenStream(lexer);
        SiddhiQLGrammarParser parser = new SiddhiQLGrammarParser(tokens);

        SiddhiQLGrammarParser.definitionStreamFinal_return r = parser.definitionStreamFinal();
        CommonTree t = (CommonTree) r.getTree();

        CommonTreeNodeStream nodes = new CommonTreeNodeStream(t);
        nodes.setTokenStream(tokens);
        SiddhiQLGrammarWalker walker = new SiddhiQLGrammarWalker(nodes);
        return walker.definitionStreamFinal();
    } catch (Throwable e) {
        throw new SiddhiParserException(e.getMessage(), e);
    }
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
public static AbstractDefinition getStreamDefinition(String executionPlan, String streamId) {
	SiddhiManager siddhiManager = null;
	SiddhiAppRuntime runtime = null;
	try {
		siddhiManager = new SiddhiManager();
		runtime = siddhiManager.createSiddhiAppRuntime(executionPlan);
		Map<String, StreamDefinition> definitionMap = runtime.getStreamDefinitionMap();
		if (definitionMap.containsKey(streamId)) {
			return definitionMap.get(streamId);
		} else {
			throw new IllegalArgumentException("Unknown stream id" + streamId);
		}
	} finally {
		if (runtime != null) {
			runtime.shutdown();
		}
		if (siddhiManager != null) {
			siddhiManager.shutdown();
		}
	}
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
@Test
public void testStreamSchemaWithPojo() {
	TypeInformation<Event> typeInfo = TypeExtractor.createTypeInfo(Event.class);
	assertTrue("Type information should be PojoTypeInfo", typeInfo instanceof PojoTypeInfo);

	SiddhiStreamSchema<Event> schema = new SiddhiStreamSchema<>(typeInfo, "id", "timestamp", "name", "price");
	assertEquals(4, schema.getFieldIndexes().length);

	StreamDefinition streamDefinition = schema.getStreamDefinition("test_stream");
	assertArrayEquals(new String[]{"id", "timestamp", "name", "price"}, streamDefinition.getAttributeNameArray());

	assertEquals(Attribute.Type.INT, streamDefinition.getAttributeType("id"));
	assertEquals(Attribute.Type.LONG, streamDefinition.getAttributeType("timestamp"));
	assertEquals(Attribute.Type.STRING, streamDefinition.getAttributeType("name"));
	assertEquals(Attribute.Type.DOUBLE, streamDefinition.getAttributeType("price"));

	assertEquals("define stream test_stream (id int,timestamp long,name string,price double);", schema.getStreamDefinitionExpression("test_stream"));
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
public static AbstractDefinition getStreamDefinition(String executionPlan, String streamId) {
    SiddhiManager siddhiManager = null;
    SiddhiAppRuntime runtime = null;
    try {
        siddhiManager = new SiddhiManager();
        runtime = siddhiManager.createSiddhiAppRuntime(executionPlan);
        Map<String, StreamDefinition> definitionMap = runtime.getStreamDefinitionMap();
        if (definitionMap.containsKey(streamId)) {
            return definitionMap.get(streamId);
        } else {
            throw new IllegalArgumentException("Unknown stream id" + streamId);
        }
    } finally {
        if (runtime != null) {
            runtime.shutdown();
        }
        if (siddhiManager != null) {
            siddhiManager.shutdown();
        }
    }
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
@Test
public void testStreamSchemaWithPojo() {
    TypeInformation<Event> typeInfo = TypeExtractor.createTypeInfo(Event.class);
    assertTrue("Type information should be PojoTypeInfo", typeInfo instanceof PojoTypeInfo);

    SiddhiStreamSchema<Event> schema = new SiddhiStreamSchema<>(typeInfo, "id", "timestamp", "name", "price");
    assertEquals(4, schema.getFieldIndexes().length);

    StreamDefinition streamDefinition = schema.getStreamDefinition("test_stream");
    assertArrayEquals(new String[]{"id", "timestamp", "name", "price"}, streamDefinition.getAttributeNameArray());

    assertEquals(Attribute.Type.INT, streamDefinition.getAttributeType("id"));
    assertEquals(Attribute.Type.LONG, streamDefinition.getAttributeType("timestamp"));
    assertEquals(Attribute.Type.STRING, streamDefinition.getAttributeType("name"));
    assertEquals(Attribute.Type.DOUBLE, streamDefinition.getAttributeType("price"));

    assertEquals("define stream test_stream (id int,timestamp long,name string,price double);", schema.getStreamDefinitionExpression("test_stream"));
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
@Test
public void testMultilevelNestedAnnotations1() throws SiddhiParserException {
    StreamDefinition streamDefinition = SiddhiCompiler.parseStreamDefinition(
            "@sink(url='http://foo.com/test/{{data}}', " +
                    "   @map(type='xml', " +
                    "@payload('<test><time>{{time}}</time></test>')" +
                    "   )" +
                    ") " +
                    "define stream fooStream (id int, name string);"
    );

    AssertJUnit.assertEquals(
            StreamDefinition
                    .id("fooStream")
                    .attribute("id", Attribute.Type.INT)
                    .attribute("name", Attribute.Type.STRING)
                    .annotation(Annotation.annotation("sink")
                            .element("url", "http://foo.com/test/{{data}}")
                            .annotation(Annotation.annotation("map")
                                    .element("type", "xml")
                                    .annotation(Annotation.annotation("payload")
                                            .element("<test><time>{{time}}</time></test>")))),
            streamDefinition);
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
public final void init(StreamDefinition streamDefinition, String mapType, OptionHolder mapOptionHolder,
                       List<AttributeMapping> attributeMappings, String sourceType,
                       List<AttributeMapping> transportMappings, SourceHandler sourceHandler,
                       ConfigReader configReader, SiddhiAppContext siddhiAppContext) {
    this.streamDefinition = streamDefinition;
    this.mapType = mapType;
    this.sourceType = sourceType;
    this.transportMappings = transportMappings;
    if (sourceHandler != null) {
        sourceHandler.initSourceHandler(siddhiAppContext.getElementIdGenerator().createNewId(), streamDefinition);
    }
    this.sourceHandler = sourceHandler;
    this.siddhiAppContext = siddhiAppContext;
    if (siddhiAppContext.getStatisticsManager() != null) {
        this.throughputTracker = QueryParserHelper.createThroughputTracker(siddhiAppContext,
                streamDefinition.getId(),
                SiddhiConstants.METRIC_INFIX_SOURCES, sourceType);
        this.mapperLatencyTracker = QueryParserHelper.createLatencyTracker(siddhiAppContext,
                streamDefinition.getId(),
                SiddhiConstants.METRIC_INFIX_SOURCE_MAPPERS,
                sourceType + SiddhiConstants.METRIC_DELIMITER + mapType);
    }
    init(streamDefinition, mapOptionHolder, attributeMappings, configReader, siddhiAppContext);
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
public static QuerySelector constructQuerySelector(Map<String, StreamDefinition> tempStreamDefinitionMap, OutputStream outStream, Selector selector, OutputRateManager outputRateManager
        , SiddhiContext siddhiContext, MetaStreamEvent metaStreamEvent, List<VariableExpressionExecutor> variableExpressionExecutors) {
    boolean currentOn = false;
    boolean expiredOn = false;
    String id = null;

    if (outStream != null) {
        if (outStream.getOutputEventType() == OutputStream.OutputEventType.CURRENT_EVENTS || outStream.getOutputEventType() == OutputStream.OutputEventType.ALL_EVENTS) {
            currentOn = true;
        }
        if (outStream.getOutputEventType() == OutputStream.OutputEventType.EXPIRED_EVENTS || outStream.getOutputEventType() == OutputStream.OutputEventType.ALL_EVENTS) {
            expiredOn = true;
        }

        id = outStream.getId();
    } else {
        currentOn = true;
        expiredOn = true;
    }
    return new QuerySelector(id, selector, outputRateManager, siddhiContext, currentOn, expiredOn, tempStreamDefinitionMap, metaStreamEvent, variableExpressionExecutors);


}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
@Test(expectedExceptions = SiddhiAppCreationException.class)
public void testFilterQuery48() throws InterruptedException {
    log.info("Filter test48");

    SiddhiManager siddhiManager = new SiddhiManager();

    StreamDefinition cseEventStream = StreamDefinition.id("cseEventStream").attribute("symbol", Attribute.Type
            .STRING).attribute("price", Attribute.Type.FLOAT).attribute("available", Attribute.Type.BOOL);
    Query query = new Query();
    query.from(InputStream.stream("cseEventStream").
            filter(Expression.not(Expression.variable("price"))));
    query.annotation(Annotation.annotation("info").element("name", "query1"));
    query.select(
            Selector.selector().
                    select("symbol", Expression.variable("symbol")).
                    select("price", Expression.variable("price")).
                    select("available", Expression.variable("available"))
    );
    query.insertInto("StockQuote");

    SiddhiApp siddhiApp = new SiddhiApp("ep1");
    siddhiApp.defineStream(cseEventStream);
    siddhiApp.addQuery(query);
    SiddhiAppRuntime siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(siddhiApp);

}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
public int enlargeStream(String streamName, List<ColumnNameTypeValue> columns, Boolean raiseException) throws
        ServiceException {
    int addedColumns = 0;
    StreamDefinition streamMetaData = siddhiManager.getStreamDefinition(streamName);
    for (ColumnNameTypeValue columnNameTypeValue : columns) {
        if (!SiddhiUtils.columnAlreadyExistsInStream(columnNameTypeValue.getColumn(), streamMetaData)) {
            addedColumns++;
            // JPFM -- Updating the columns in streamStatusDao
            streamStatusDao.addColumn(streamName, columnNameTypeValue);
            streamMetaData.attribute(columnNameTypeValue.getColumn(), getSiddhiType(columnNameTypeValue.getType()));
        } else {
            if (raiseException) {
                throw new ServiceException(String.format("Alter stream error, Column %s already "
                                + "exists.",
                        columnNameTypeValue.getColumn()));
            }
        }
    }

    return addedColumns;
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
public static SingleStreamRuntime parseInputStream(SingleInputStream inputStream, SiddhiContext context, MetaStreamEvent metaStreamEvent, List<VariableExpressionExecutor> executors) {
    Processor processor = null;
    int i = 0;
    if (!inputStream.getStreamHandlers().isEmpty()) {
        for (StreamHandler handler : inputStream.getStreamHandlers()) {
            if (i == 0) {
                processor = generateProcessor(handler, context, metaStreamEvent, executors);
                i++;
            } else {
                processor.setToLast(generateProcessor(handler, context, metaStreamEvent, executors));
            }
        }
    }
    QueryStreamReceiver queryStreamReceiver = new QueryStreamReceiver((StreamDefinition) metaStreamEvent.getDefinition());
    SingleStreamRuntime singleStreamRuntime = new SingleStreamRuntime(queryStreamReceiver, processor);
    return singleStreamRuntime;
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
protected QueryCreator(String queryId, Query query, ConcurrentMap<String, AbstractDefinition> streamDefinitionMap, ConcurrentMap<String, AbstractDefinition> localStreamDefinitionMap, OutputRateManager outputRateManager, SiddhiContext siddhiContext) {
    this.queryId = queryId;
    this.query = query;
    this.streamDefinitionMap = streamDefinitionMap;
    this.localStreamDefinitionMap = localStreamDefinitionMap;
    this.outputRateManager = outputRateManager;
    this.siddhiContext = siddhiContext;
    if (getInputStream() instanceof SingleInputStream) {
        String streamId = ((SingleInputStream) getInputStream()).getStreamId();
        if (((SingleInputStream) getInputStream()).isInnerStream()) {
            this.defaultDefinition = (StreamDefinition) localStreamDefinitionMap.get(streamId);
        } else {
            this.defaultDefinition = (StreamDefinition) streamDefinitionMap.get(streamId);
        }
    } //TODO other streams
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
@Test(expectedExceptions = SiddhiAppCreationException.class)
public void testFilterQuery49() throws InterruptedException {
    log.info("Filter test49");

    SiddhiManager siddhiManager = new SiddhiManager();

    StreamDefinition cseEventStream = StreamDefinition.id("cseEventStream").attribute("symbol", Attribute.Type
            .STRING).attribute("price", Attribute.Type.FLOAT).attribute("available", Attribute.Type.BOOL);
    Query query = new Query();
    query.from(InputStream.stream("cseEventStream").
            filter(Expression.variable("price")));
    query.annotation(Annotation.annotation("info").element("name", "query1"));
    query.select(
            Selector.selector().
                    select("symbol", Expression.variable("symbol")).
                    select("price", Expression.variable("price")).
                    select("available", Expression.variable("available"))
    );
    query.insertInto("StockQuote");

    SiddhiApp siddhiApp = new SiddhiApp("ep1");
    siddhiApp.defineStream(cseEventStream);
    siddhiApp.addQuery(query);
    SiddhiAppRuntime siddhiAppRuntime = siddhiManager.createSiddhiAppRuntime(siddhiApp);

}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的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.query.api.definition.StreamDefinition; //导入依赖的package包/类
@Test
public void Test1() throws RecognitionException, SiddhiParserException {
    StreamDefinition streamDefinition = SiddhiCompiler.parseStreamDefinition("define stream cseStream ( symbol string, price int, volume float )");
    Assert.assertEquals(new StreamDefinition().
            name("cseStream").
            attribute("symbol", Attribute.Type.STRING).
            attribute("price", Attribute.Type.INT).
            attribute("volume", Attribute.Type.FLOAT).toString(),
                        streamDefinition.toString());
}
 

import org.wso2.siddhi.query.api.definition.StreamDefinition; //导入依赖的package包/类
@Test
public void Test2() throws RecognitionException, SiddhiParserException {
    StreamDefinition streamDefinition = SiddhiCompiler.parseStreamDefinition("define stream `define` ( `string` string, price int, volume float );");
    Assert.assertEquals(new StreamDefinition().
            name("define").
            attribute("string", Attribute.Type.STRING).
            attribute("price", Attribute.Type.INT).
            attribute("volume", Attribute.Type.FLOAT).toString(),
                        streamDefinition.toString());
}