import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
public static boolean isValidTupleToAvroMapping(String tupleSchemaName, StreamSchema tupleSchema, Schema avroSchema)
		throws Exception {
	boolean validMapping = true;
	LOGGER.log(TraceLevel.TRACE,
			"Checking attributes in tuple schema " + tupleSchemaName + ": " + tupleSchema.getAttributeNames());
	for (String attributeName : tupleSchema.getAttributeNames()) {
		Attribute attribute = tupleSchema.getAttribute(attributeName);
		Field avroField = avroSchema.getField(attributeName);
		if (avroField != null)
			validMapping = validMapping
					& isValidAttributeToAvroMapping(attributeName, attribute.getType(), avroField.schema());
		else
			LOGGER.log(TraceLevel.INFO, "Attribute " + attributeName + " in schema " + tupleSchemaName
					+ " does not have a corresponding field in the Avro schema. It will not be mapped.");
	}
	return validMapping;
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
@ContextCheck(compile=false)
public static void checkOutputAttributeRuntime(OperatorContextChecker checker) {
	
	OperatorContext context = checker.getOperatorContext();
	StreamSchema schema = context.getStreamingOutputs().get(0).getStreamSchema();
	Attribute resultAttribute = schema.getAttribute(ANALYSISRESULT_ATTRIBUTE);
	
	//make sure that the output attribute is the right type based on the type of analysis
	String type = context.getParameterValues("analysisType").get(0);
	
	if(type.equals(AnalysisType.Prediction.name())) {
		if( resultAttribute.getType().getMetaType() != MetaType.FLOAT64) {
			tracer.log(TraceLevel.ERROR, "TRACE_M_WRONG_TYPE_ALS", new Object[]{"Prediction", "float64", resultAttribute.getType()});
			checker.setInvalidContext();
		}
	}
	else if(!isList(resultAttribute, Integer.class)) {
		tracer.log(TraceLevel.ERROR, "TRACE_M_WRONG_TYPE_ALS", new Object[]{type, "list<int32>", resultAttribute.getType()});
		checker.setInvalidContext();
	}
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
@ContextCheck
public static void checkControlPortInputAttribute(OperatorContextChecker checker) {
	OperatorContext context = checker.getOperatorContext();
	
	if(context.getNumberOfStreamingInputs() == 2) {
		StreamSchema schema = context.getStreamingInputs().get(1).getStreamSchema();
		
		//the first attribute must be of type rstring
		Attribute jsonAttr = schema.getAttribute(0);
		
		//check if the output attribute is present where the result will be stored
		if(jsonAttr != null && jsonAttr.getType().getMetaType() != MetaType.RSTRING) {
			tracer.log(TraceLevel.ERROR, "COMPILE_M_WRONG_TYPE", jsonAttr.getType());
			checker.setInvalidContext();
		}
	}
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
/**
 * Compile time to check to ensure that the output schema contains an attribute called
 * 'analysisResult'. The type of the attribute depends on the specific type of the
 * operator and is handled by the appropriate derived class.
 */
@ContextCheck
public static void checkOutputAttribute(OperatorContextChecker checker) {
	OperatorContext context = checker.getOperatorContext();
	
	if(context.getNumberOfStreamingOutputs() == 1) {
		StreamSchema schema = context.getStreamingOutputs().get(0).getStreamSchema();
		Attribute resultAttribute = schema.getAttribute(ANALYSISRESULT_ATTRIBUTE);
		
		//check if the output attribute is present where the result will be stored
		if(resultAttribute == null) {
			tracer.log(TraceLevel.ERROR, "COMPILE_M_MISSING_ATTRIBUTE", new Object[]{ ANALYSISRESULT_ATTRIBUTE});
			checker.setInvalidContext();
		}
	}
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
static Map<String,? extends Object> fixParameters(Map<String,? extends Object> params) {
    if (params == null)
        return null;
    
    Map<String,Object> fp = new HashMap<>(params);
    
    // Iterator over params as we may modify fp
    for (String name : params.keySet()) {
        Object value = fp.get(name);
        if (value instanceof StreamSchema) {
            fp.put(name, JParamTypes.create(TYPE_SPLTYPE, ((StreamSchema) value).getLanguageType()));
        } else if (value instanceof Attribute) {
            fp.put(name, JParamTypes.create(TYPE_ATTRIBUTE, ((Attribute) value).getName()));
        }
    }
    return fp;
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
public static SPLMapping<?> getSPLMapping(StreamSchema schema) {

        if (STRING.equals(schema)) {
            return SPLMapping.JavaString;
        }
        if (JAVA_OBJECT.equals(schema)) {
            return new SPLJavaObject(schema);
        }
        if (BLOB.equals(schema)) {
            return SPLMapping.JavaBlob;
        }
        if (XML.equals(schema)) {
            return SPLMapping.JavaXML;
        }

        return new SPLTuple(schema);
    }
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
@Test
public void testSPLContentsGood() throws Exception {
    final Topology topology = new Topology();
    TStream<String> source = topology.strings("A", "B", "C", "D");
    StreamSchema schema = SPLSchemas.STRING.extend("int32", "id");
    SPLStream tested = SPLStreams.convertStream(source, (s,t) -> {t.setString(0, s); t.setInt(1, s.charAt(0)); return t;}, schema);
    
    
    ExpectedTuples expected = new ExpectedTuples(schema);
    int id = "A".charAt(0);
    expected.addAsTuple(new RString("A"), id);
    expected.addAsTuple(new RString("B"), ++id);
    expected.addAsTuple(new RString("C"), ++id);
    expected.addAsTuple(new RString("D"), ++id);

    Condition<List<Tuple>> contents = expected.contents(tested);

    boolean passed = complete(topology.getTester(), contents, 10, TimeUnit.SECONDS);
    assertTrue(contents.toString(), contents.valid());
    assertTrue(passed);
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
@Test
public void testSPLContentsBad() throws Exception {
    final Topology topology = new Topology();
    TStream<String> source = topology.strings("A", "B", "C", "D");
    StreamSchema schema = SPLSchemas.STRING.extend("int32", "id");
    SPLStream tested = SPLStreams.convertStream(source, (s,t) -> {t.setString(0, s); t.setInt(1, s.charAt(0)); return t;}, schema);
    
    
    ExpectedTuples expected = new ExpectedTuples(schema);
    int id = "A".charAt(0);
    expected.addAsTuple(new RString("A"), id);
    expected.addAsTuple(new RString("B"), ++id);
    expected.addAsTuple(new RString("C"), 1241241);
    expected.addAsTuple(new RString("D"), ++id);

    Condition<List<Tuple>> contents = expected.contents(tested);

    boolean passed = complete(topology.getTester(), contents, 10, TimeUnit.SECONDS);
    assertFalse(passed);

    assertFalse(contents.toString(), contents.valid());
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
@Test
public void testConsistentPeriodic() throws Exception {
    Topology topology = new Topology("testConsistentPeriodic");
    
    StreamSchema schema = Type.Factory.getStreamSchema("tuple<uint64 id>");
    Map<String,Object> params = new HashMap<>();
    params.put("iterations", 200);
    params.put("period", 0.05);
    
    SPLStream b = SPL.invokeSource(topology, "spl.utility::Beacon", params, schema);
    
    ConsistentRegionConfig config = periodic(2);
    assertSame(b, b.setConsistent(config));
    
    Condition<Long> atLeast = topology.getTester().atLeastTupleCount(b, 200);
    complete(topology.getTester(), atLeast, 80, TimeUnit.SECONDS);
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
@Test
public void testConsistentOperatorDriven() throws Exception {
    Topology topology = new Topology("testConsistentOperatorDriven");
    
    StreamSchema schema = Type.Factory.getStreamSchema("tuple<uint64 id>");
    Map<String,Object> params = new HashMap<>();
    params.put("iterations", 300);
    params.put("triggerCount", SPL.createValue(37, Type.MetaType.UINT32));
    
    SPLStream b = SPL.invokeSource(topology, "spl.utility::Beacon", params, schema);
    
    ConsistentRegionConfig config = ConsistentRegionConfig.operatorDriven();
    assertSame(b, b.setConsistent(config));
    
    Condition<Long> atLeast = topology.getTester().atLeastTupleCount(b, 300);
    complete(topology.getTester(), atLeast, 40, TimeUnit.SECONDS);
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
/**
 * Check that the fileAttributeName parameter is an attribute of the right
 * type.
 * 
 * @param checker
 */
@ContextCheck(compile = false)
public static void checkFileAttributeName(OperatorContextChecker checker) {
	StreamSchema inputSchema = checker.getOperatorContext()
			.getStreamingInputs().get(0).getStreamSchema();
	List<String> fileAttrNameList = checker.getOperatorContext()
			.getParameterValues(IHdfsConstants.PARAM_FILE_NAME_ATTR);
	if (fileAttrNameList == null || fileAttrNameList.size() == 0) {
		// Nothing to check, because the parameter doesn't exist.
		return;
	}

	String fileAttrName = fileAttrNameList.get(0);
	Attribute fileAttr = inputSchema.getAttribute(fileAttrName);
	if (fileAttr == null) {
		checker.setInvalidContext(Messages.getString("HDFS_SINK_NO_ATTRIBUTE"), 
				new Object[] { fileAttrName });
	}
	if (MetaType.RSTRING != fileAttr.getType().getMetaType()
			&& MetaType.USTRING != fileAttr.getType().getMetaType()) {
		checker.setInvalidContext(
				Messages.getString("HDFS_SINK_INVALID_ATTR_FILENAME", fileAttr.getType().getMetaType()), 
				new Object[] {});
	}
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
@ContextCheck()
public static void checkInputPortSchema(OperatorContextChecker checker) {
	List<StreamingInput<Tuple>> streamingInputs = checker.getOperatorContext().getStreamingInputs();
	if (streamingInputs.size() > 0) {
		StreamSchema inputSchema = streamingInputs.get(0).getStreamSchema();
		if (inputSchema.getAttributeCount() > 1) {
			checker.setInvalidContext(
					Messages.getString("HDFS_DS_INVALID_INPUT_PORT"), 
					null);
		}

		if (inputSchema.getAttribute(0).getType().getMetaType() != MetaType.RSTRING) {
			checker.setInvalidContext(Messages.getString("HDFS_DS_INVALID_ATTRIBUTE", 
					 inputSchema.getAttribute(0).getType().getMetaType()), null);
		}

		ConsistentRegionContext crContext = checker.getOperatorContext().getOptionalContext(
				ConsistentRegionContext.class);
		if (crContext != null) {
			LOGGER.log( LogLevel.WARNING, Messages.getString("HDFS_DS_CONSISTENT_REGION_NOT_SUPPORTED")); 
		}
	}

}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
public static GenericRecord convertTupleToAvro(Tuple tuple, StreamSchema streamSchema, Schema avroSchema) {
	GenericRecord datum = new GenericData.Record(avroSchema);
	for (String attributeName : streamSchema.getAttributeNames()) {
		Attribute attribute = streamSchema.getAttribute(attributeName);
		Object tupleAttribute = tuple.getObject(attributeName);
		Field avroField = avroSchema.getField(attributeName);
		// If there is an Avro field associated with this attribute, convert
		if (avroField != null)
			datum.put(attributeName,
					convertAttributeToAvro(attributeName, tupleAttribute, attribute.getType(), avroField.schema()));
	}
	return datum;
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
/**
 * Check to ensure that an analysisResult attribute of type int32 is present on the output schema
 */
@ContextCheck
public static void checkOutputAttributeType(OperatorContextChecker checker) {
	
	OperatorContext context = checker.getOperatorContext();
	StreamSchema schema = context.getStreamingOutputs().get(0).getStreamSchema();
	Attribute resultAttribute = schema.getAttribute(ANALYSISRESULT_ATTRIBUTE);
	
	if(resultAttribute != null && resultAttribute.getType().getMetaType() != MetaType.INT32) {
		tracer.log(TraceLevel.ERROR, "COMPILE_M_WRONG_TYPE_FULL", new Object[]{ANALYSISRESULT_ATTRIBUTE, "int32", resultAttribute.getType()});
		checker.setInvalidContext();
	}
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
/**
 * Check to ensure that an analysisResult attribute of type float64 is present on the output schema
 */
@ContextCheck
public static void checkOutputAttributeType(OperatorContextChecker checker) {
	
	OperatorContext context = checker.getOperatorContext();
	StreamSchema schema = context.getStreamingOutputs().get(0).getStreamSchema();
	Attribute resultAttribute = schema.getAttribute(ANALYSISRESULT_ATTRIBUTE);
	
	if(resultAttribute != null && resultAttribute.getType().getMetaType() != MetaType.FLOAT64) {
		tracer.log(TraceLevel.ERROR, "COMPILE_M_WRONG_TYPE_FULL", new Object[]{ANALYSISRESULT_ATTRIBUTE, "float64", resultAttribute.getType()});
		checker.setInvalidContext();
	}
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
@ContextCheck (compile  = true)
public static void checkOutputAttributeType(OperatorContextChecker checker) {
	OperatorContext context = checker.getOperatorContext();

	StreamSchema schema = context.getStreamingOutputs().get(0).getStreamSchema();
	Attribute resultAttribute = schema.getAttribute(ANALYSISRESULT_ATTRIBUTE);
	
	if(resultAttribute != null && resultAttribute.getType().getMetaType() != MetaType.FLOAT64) {
		tracer.log(TraceLevel.ERROR, "COMPILE_M_WRONG_TYPE_FULL", new Object[]{ANALYSISRESULT_ATTRIBUTE, "float64", resultAttribute.getType()});
		checker.setInvalidContext();
	}
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
/**
 * Invoke an SPL operator with an arbitrary number
 * of input and output ports.
 * <BR>
 * Each input stream or window in {@code inputs} results in
 * a input port for the operator with the input port index
 * matching the position of the input in {@code inputs}.
 * If {@code inputs} is {@code null} or empty then the operator will not
 * have any input ports.
 * <BR>
 * Each SPL schema in {@code outputSchemas} an output port
 * for the operator with the output port index
 * matching the position of the schema in {@code outputSchemas}.
 * If {@code outputSchemas} is {@code null} or empty then the operator will not
 * have any output ports.
 * 
 * @param te Reference to Topology the operator will be in.
 * @param name Name for the operator invocation.
 * @param kind
 *            SPL kind of the operator to be invoked.
 * @param inputs Input streams to be connected to the operator. May be {@code null} if no input  streams are required.
 * @param outputSchemas Schemas of the output streams. May be {@code null} if no output streams are required.
 * @param params
 *            Parameters for the SPL Java Primitive operator, ignored if {@code null}.
 * @return List of {@code SPLStream} instances that represent the outputs of the operator.
 */
public static List<SPLStream> invokeOperator(
        TopologyElement te,
        String name,
        String kind,
        List<? extends SPLInput> inputs,
        List<StreamSchema> outputSchemas,
        Map<String, ? extends Object> params) {
    
    BOperatorInvocation op = te.builder().addSPLOperator(name, kind, fixParameters(params));
    
    SourceInfo.setSourceInfo(op, SPL.class);
    
    if (inputs != null && !inputs.isEmpty()) {
        for (SPLInput input : inputs)
            SPL.connectInputToOperator(input, op);
    }
    
    if (outputSchemas == null || outputSchemas.isEmpty())
        return Collections.emptyList();
    
    List<SPLStream> streams = new ArrayList<>(outputSchemas.size());
    for (StreamSchema outputSchema : outputSchemas)
        streams.add(newSPLStream(te, op, outputSchema, outputSchemas.size() == 1));
        
    return streams;
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
private static SPLStream _subscribe(TopologyElement te, Object topic, StreamSchema schema) {
    Map<String, Object> params = new HashMap<>();
            
    params.put("topic", requireNonNull(topic));
    params.put("streamType", requireNonNull(schema));

    SPLStream stream = SPL.invokeSource(te,
            "com.ibm.streamsx.topology.topic::Subscribe",
            params, schema);

    return stream;
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
/**
 * Invoke an SPL Java primitive operator with an arbitrary number
 * of input and output ports.
 * <BR>
 * Each input stream or window in {@code inputs} results in
 * a input port for the operator with the input port index
 * matching the position of the input in {@code inputs}.
 * If {@code inputs} is {@code null} or empty then the operator will not
 * have any input ports.
 * <BR>
 * Each SPL schema in {@code outputSchemas} an output port
 * for the operator with the output port index
 * matching the position of the schema in {@code outputSchemas}.
 * If {@code outputSchemas} is {@code null} or empty then the operator will not
 * have any output ports.
 * 
 * @param te Reference to Topology the operator will be in.
 * @param opClass Class of the operator to be invoked.
 * @param inputs Input streams to be connected to the operator. May be {@code null} if no input  streams are required.
 * @param outputSchemas Schemas of the output streams. May be {@code null} if no output streams are required.
 * @param params
 *            Parameters for the SPL Java Primitive operator, ignored if {@code null}.
 * @return List of {@code SPLStream} instances that represent the outputs of the operator.
 */
public static List<SPLStream> invokeJavaPrimitive(
        TopologyElement te,
        Class<? extends Operator> opClass,
        List<? extends SPLInput> inputs, List<StreamSchema> outputSchemas, Map<String, ? extends Object> params) {
    
    BOperatorInvocation op = te.builder().addOperator(
            getInvocationName(opClass),
            getKind(opClass), fixParameters(params));
    op.setModel(MODEL_SPL, LANGUAGE_JAVA);
    op._json().addProperty(KIND_CLASS, opClass.getCanonicalName());
    SourceInfo.setSourceInfo(op, JavaPrimitive.class);
    
    if (inputs != null && !inputs.isEmpty()) {
        for (SPLInput input : inputs)
            SPL.connectInputToOperator(input, op);
    }
    
    if (outputSchemas == null || outputSchemas.isEmpty())
        return Collections.emptyList();
    
    List<SPLStream> streams = new ArrayList<>(outputSchemas.size());
    for (StreamSchema outputSchema : outputSchemas)
        streams.add(newSPLStream(te, op, outputSchema, outputSchemas.size() == 1));
        
    return streams;
}
 

import com.ibm.streams.operator.StreamSchema; //导入依赖的package包/类
/**
 * Invocation of a Java primitive source operator to produce a SPL Stream.
 * 
 * @param te
 *            Reference to Topology the operator will be in.
 * @param opClass
 *            Class of the operator to be invoked.
 * @param schema
 *            Schema of the output port.
 * @param params
 *            Parameters for the SPL Java Primitive operator, ignored if {@code null}.
 * @return SPLStream the represents the output of the operator.
 */
public static SPLStream invokeJavaPrimitiveSource(TopologyElement te,
        Class<? extends Operator> opClass,
        StreamSchema schema, Map<String, ? extends Object> params) {

    
    BOperatorInvocation source = te.builder().addOperator(
            getInvocationName(opClass),
            getKind(opClass), fixParameters(params));
    source.setModel(MODEL_SPL, LANGUAGE_JAVA);
    source._json().addProperty(KIND_CLASS, opClass.getCanonicalName());
    SourceInfo.setSourceInfo(source, JavaPrimitive.class);
    return newSPLStream(te, source, schema, true);
}