import com.ibm.streams.operator.Type; //导入依赖的package包/类
@Test
public void testSimple() throws Exception {

    Topology topology = new Topology("testSimple");

    TStream<String> hw = topology.strings("Hello", "World!", "Test!!");
    SPLStream hws = SPLStreams.stringToSPLStream(hw);

    Tester tester = topology.getTester();
    StreamCounter<Tuple> counter = tester.splHandler(hws,
            new StreamCounter<Tuple>());
    StreamCollector<LinkedList<Tuple>, Tuple> collector = tester
            .splHandler(hws, StreamCollector.newLinkedListCollector());

    StreamsContextFactory
            .getStreamsContext(StreamsContext.Type.EMBEDDED_TESTER)
            .submit(topology).get();

    assertEquals(3, counter.getTupleCount());
    assertEquals("Hello", collector.getTuples().get(0).getString(0));
    assertEquals("World!", collector.getTuples().get(1).getString(0));
    assertEquals("Test!!", collector.getTuples().get(2).getString(0));
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
@Test
public void testSimpleWithConditions() throws Exception {

    Topology topology = new Topology("testSimpleConditions");

    TStream<String> hw = topology.strings("Hello", "World!", "Test!!");

    Tester tester = topology.getTester();
    Condition<Long> expectedCount = tester.tupleCount(hw, 3);
    Condition<List<String>> expectedContents = tester.stringContents(hw,
            "Hello", "World!", "Test!!");

    StreamsContextFactory
            .getStreamsContext(StreamsContext.Type.EMBEDDED_TESTER)
            .submit(topology).get();

    assertTrue(expectedCount.valid());
    assertTrue(expectedContents.valid());
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
@Test
public void testIsSupported() throws Exception {

    Topology topology = new Topology("test");

    TStream<String> hw = topology.strings("Hello", "World!", "Test!!");
    TStream<String> hw2 = hw.transform(new AppendXform("(stream-2)"));
    // make sure "marker" ops are ok: union,parallel,unparallel
    hw
        .union(hw2)
        .parallel(2)
            .filter(nilFilter)
            .filter(nilFilter)
        .endParallel()
    .print();

    StreamsContext<?> sc = StreamsContextFactory
            .getStreamsContext(StreamsContext.Type.EMBEDDED);
    assertTrue(sc.isSupported(topology));

    sc = StreamsContextFactory
            .getStreamsContext(StreamsContext.Type.EMBEDDED_TESTER);
    assertTrue(sc.isSupported(topology));
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
public static SPLStream testTupleStream(Topology topology, boolean withSets) {
    TStream<Long> beacon = BeaconStreams.longBeacon(topology, TUPLE_COUNT);

    SPLStream tuples = SPLStreams.convertStream(beacon, new BiFunction<Long, OutputTuple, OutputTuple>() {
        private static final long serialVersionUID = 1L;
        
        private transient TupleType type;
        private transient Random rand;

        @Override
        public OutputTuple apply(Long v1, OutputTuple v2) {
        	if (type == null) {
        		type = Type.Factory.getTupleType(getPythonTypesSchema(withSets).getLanguageType());
        		rand = new Random();
        	}
        	Tuple randTuple = (Tuple) type.randomValue(rand);
        	v2.assign(randTuple);
            return v2;
        }
    }, getPythonTypesSchema(withSets));

    return tuples;
}
 

import com.ibm.streams.operator.Type; //导入依赖的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.Type; //导入依赖的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.Type; //导入依赖的package包/类
protected static boolean isList(Attribute attr,
		Class<?> clz) {
	Type type = attr.getType();
	if(type.getMetaType() == MetaType.LIST) {
		CollectionType cType = (CollectionType)type;
		if(cType.getElementType().getObjectType() == clz) {
			return true;
		}
	}
	return false;
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
private static Object listRStringParam(String[] values) {
    List<String> literals = new ArrayList<>(values.length);
    for (String v : values)
        literals.add("\"" + v + "\""); //$NON-NLS-1$ //$NON-NLS-2$

    return new Attribute() {

        @Override
        public int getIndex() {
            // TODO Auto-generated method stub
            return 0;
        }

        @Override
        public String getName() {
            return literals.toString();
        }

        @Override
        public Type getType() {
            // TODO Auto-generated method stub
            return null;
        }

        @Override
        public boolean same(Attribute arg0) {
            return false;
        }

    };
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
private IllegalStateException notSupported(JsonObject op) {
    
    String namespace = jstring(builder._json(), NAMESPACE);
    String name = jstring(builder._json(), NAME);
    
    return new IllegalStateException(
            "Topology '"+namespace+"."+name+"'"
            + " does not support "+StreamsContext.Type.EMBEDDED+" mode:"
            + " the topology contains non-Java operator:" +
            jstring(op, KIND));
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
@Test
public void testTwoStreams() throws Exception {

    Topology topology = new Topology("testTwoStreams");

    TStream<String> hw = topology.strings("Hello", "World!", "Test!!");
    SPLStream hws = SPLStreams.stringToSPLStream(hw);

    TStream<String> hw2 = StringStreams.contains(hw, "e");
    SPLStream hw2s = SPLStreams.stringToSPLStream(hw2);

    Tester tester = topology.getTester();
    StreamCounter<Tuple> counter = tester.splHandler(hws,
            new StreamCounter<Tuple>());
    StreamCollector<LinkedList<Tuple>, Tuple> collector = tester
            .splHandler(hws, StreamCollector.newLinkedListCollector());

    StreamCounter<Tuple> counter2 = tester.splHandler(hw2s,
            new StreamCounter<Tuple>());
    StreamCollector<LinkedList<Tuple>, Tuple> collector2 = tester
            .splHandler(hw2s, StreamCollector.newLinkedListCollector());

    StreamsContextFactory
            .getStreamsContext(StreamsContext.Type.EMBEDDED_TESTER)
            .submit(topology).get();

    assertEquals(3, counter.getTupleCount());
    assertEquals("Hello", collector.getTuples().get(0).getString(0));
    assertEquals("World!", collector.getTuples().get(1).getString(0));
    assertEquals("Test!!", collector.getTuples().get(2).getString(0));

    assertEquals(2, counter2.getTupleCount());
    assertEquals("Hello", collector2.getTuples().get(0).getString(0));
    assertEquals("Test!!", collector2.getTuples().get(1).getString(0));
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
@Before
public void runSpl() {
    assumeSPLOk();
    
    assumeTrue(getTesterContext().getType() == StreamsContext.Type.STANDALONE_TESTER
    		|| getTesterContext().getType() == StreamsContext.Type.DISTRIBUTED_TESTER);
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
@Test
public void testSourceWithClass() throws Exception {
    Topology topology = new Topology("testSourceWithClass");
    
    addTestToolkit(topology);
    
    StreamSchema outSchema = Type.Factory.getStreamSchema("tuple<int32 seq>");
    
    int count = new Random().nextInt(200) + 10;
    SPLStream pysrc = SPL.invokeSource(topology,
    		"com.ibm.streamsx.topology.pysamples.sources::Range",
    		Collections.singletonMap("count", count), outSchema);
    
    Tester tester = topology.getTester();
    Condition<Long> expectedCount = tester.tupleCount(pysrc, count);
    Condition<List<Tuple>> outTuples = tester.tupleContents(pysrc);
    
    // getConfig().put(ContextProperties.TRACING_LEVEL, TraceLevel.DEBUG);
            
    complete(tester, expectedCount, 20, TimeUnit.SECONDS);

    assertTrue(expectedCount.valid());
    
    List<Tuple> result = outTuples.getResult();
    
    assertEquals(count, result.size());
    for (int i = 0; i < count; i++)
        assertEquals(i, result.get(i).getInt("seq"));
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
@SuppressWarnings("serial")
private static SPLStream convertListToSPL(TStream<List<String>> s) {
    
    return SPLStreams.convertStream(s, new BiFunction<List<String>, OutputTuple, OutputTuple>() {

        @Override
        public OutputTuple apply(List<String> v1, OutputTuple v2) {
            v2.setList("lrs", v1);
            return v2;
        }
    }, Type.Factory.getStreamSchema("tuple<list<ustring> lrs>"));

}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
/**
 * Expected to raise an exception as triggerCount requires it
 * be in a operator driven consistent region. Somewhat
 * enforces that testConsistentOperatorDriven() is doing the correct thing.
 * @throws Exception
 */
@Test(expected=Exception.class)
public void testMissingOperatorDrivenConsistent() 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);
    
    Condition<Long> atLeast = topology.getTester().atLeastTupleCount(b, 300);
    complete(topology.getTester(), atLeast, 40, TimeUnit.SECONDS);
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
@ContextCheck(compile = true)
public static void checkOutputPort(OperatorContextChecker checker) {
	OperatorContext context = checker.getOperatorContext();
	if (context.getNumberOfStreamingOutputs() == 1) {
		StreamingOutput<OutputTuple> streamingOutputPort = context
				.getStreamingOutputs().get(0);
		if (streamingOutputPort.getStreamSchema().getAttributeCount() != 2) {
			checker.setInvalidContext(
					Messages.getString("HDFS_SINK_OUTPUT_PORT"), 
					null);

		} else {
			if (streamingOutputPort.getStreamSchema().getAttribute(0)
					.getType().getMetaType() != Type.MetaType.RSTRING) {
				checker.setInvalidContext(
						Messages.getString("HDFS_SINK_FIRST_OUTPUT_PORT"), 
						null);
			}
			if (streamingOutputPort.getStreamSchema().getAttribute(1)
					.getType().getMetaType() != Type.MetaType.UINT64) {
				checker.setInvalidContext(
						Messages.getString("HDFS_SINK_SECOND_OUTPUT_PORT"), 
						null);

			}

		}

	}
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
private static Object convertAttributeToAvro(String attributeName, Object tupleAttribute, Type tupleAttributeType,
		Schema avroSchema) {
	Object returnObject = null;
	MetaType metaType = tupleAttributeType.getMetaType();
	switch (metaType) {
	case BOOLEAN:
		returnObject = (Boolean) tupleAttribute;
		break;
	case FLOAT32:
		returnObject = (Float) tupleAttribute;
		break;
	case FLOAT64:
		returnObject = (Double) tupleAttribute;
		break;
	case INT32:
		returnObject = (Integer) tupleAttribute;
		break;
	case INT64:
		returnObject = (Long) tupleAttribute;
		break;
	case RSTRING:
		returnObject = ((RString) tupleAttribute).getString();
		break;
	case USTRING:
		returnObject = tupleAttribute.toString();
		break;
	case TUPLE:
		Tuple subTuple = (Tuple) tupleAttribute;
		StreamSchema subStreamSchema = subTuple.getStreamSchema();
		GenericRecord subDatum = convertTupleToAvro(subTuple, subStreamSchema, avroSchema);
		// Return the Avro record
		returnObject = subDatum;
		break;
	case LIST:
		@SuppressWarnings("unchecked")
		List<Object> subList = (List<Object>) tupleAttribute;
		// Obtain the type of the elements contained in the Streams list
		Type tupleElementType = ((CollectionType) tupleAttributeType).getElementType();
		// Obtain the type of the elements contained in the Avro array
		Schema avroArrayElementType = avroSchema.getElementType();
		// Now loop through all list elements and populate the associated
		// Avro array elements
		GenericArray<Object> subArray = new GenericData.Array<Object>(subList.size(), avroSchema);
		for (Object arrayElement : subList) {
			Object avroElement = convertAttributeToAvro(attributeName, arrayElement, tupleElementType,
					avroArrayElementType);
			subArray.add(avroElement);
		}
		// Return the Avro array
		returnObject = subArray;
		break;
	default:
		LOGGER.log(TraceLevel.WARN,
				"Ignoring attribute " + attributeName + " because of unsupported type " + metaType);
	}
	return returnObject;
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
Type.MetaType metaType() {
    return metaType;
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
private void windowInput(JsonObject input, final InputPortDeclaration port) {
    JsonObject window = input.getAsJsonObject("window");
    String wt = jstring(window, "type");
    if (wt == null)
        return;
    StreamWindow.Type wtype = StreamWindow.Type.valueOf(wt);
    
    switch (wtype) {
    case NOT_WINDOWED:
        return;
    case SLIDING:
        port.sliding();
        break;
    case TUMBLING:
        port.tumbling();
        break;
    }
    
    StreamWindow.Policy evictPolicy = StreamWindow.Policy.valueOf(jstring(window, "evictPolicy"));
    
    // Eviction
    switch (evictPolicy) {
    case COUNT:
        int ecount = window.get("evictConfig").getAsInt();
        port.evictCount(ecount);
        break;
    case TIME:
        long etime = window.get("evictConfig").getAsLong();
        TimeUnit eunit = TimeUnit.valueOf(window.get("evictTimeUnit").getAsString());
        port.evictTime(etime, eunit);
        break;
    default:
        throw new UnsupportedOperationException(evictPolicy.name());
    }
    
    String stp = jstring(window, "triggerPolicy");      
    if (stp != null) {
        StreamWindow.Policy triggerPolicy = StreamWindow.Policy.valueOf(stp);
        switch (triggerPolicy) {
        case NONE:
            break;
        case COUNT:
            int tcount = window.get("triggerConfig").getAsInt();
            port.triggerCount(tcount);
            break;
        case TIME:
            long ttime = window.get("triggerConfig").getAsLong();
            TimeUnit tunit = TimeUnit.valueOf(window.get("triggerTimeUnit").getAsString());
            port.triggerTime(ttime, tunit);
            break;
        default:
            throw new UnsupportedOperationException(evictPolicy.name());
        }
    }
    
    if (window.has("partitioned") && window.get("partitioned").getAsBoolean())
        port.partitioned();
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
@Test
public void testParallelSubmissionParamInner() throws Exception {
    checkUdpSupported();

    Topology topology = newTopology("testParallelSubmissionParamInner");
    final int count = new Random().nextInt(1000) + 37;
    String submissionWidthName = "width";
    Integer submissionWidth = 5;
    String submissionAppendName = "append";
    boolean submissionAppend = true;
    String submissionFlushName = "flush";
    Integer submissionFlush = 1;
    String submissionThresholdName = "threshold";
    String submissionDefaultedThresholdName = "defaultedTreshold";
    Integer submissionThreshold = -1;
    // getConfig().put(ContextProperties.KEEP_ARTIFACTS, true);

    Supplier<Integer> threshold = topology.createSubmissionParameter(submissionThresholdName, Integer.class);
    Supplier<Integer> defaultedThreshold = topology.createSubmissionParameter(submissionDefaultedThresholdName, submissionThreshold);
            
    TStream<BeaconTuple> fb = BeaconStreams.beacon(topology, count);
    TStream<BeaconTuple> pb = fb.parallel(
            topology.createSubmissionParameter(submissionWidthName, Integer.class));

    TStream<Integer> is = pb.transform(randomHashProducer());
    
    // submission param use within a parallel region
    StreamSchema schema = getStreamSchema("tuple<int32 i>");
    SPLStream splStream = SPLStreams.convertStream(is, cvtMsgFunc(), schema);
    File tmpFile = File.createTempFile("parallelTest", null);
    tmpFile.deleteOnExit();
    Map<String,Object> splParams = new HashMap<>();
    splParams.put("file", tmpFile.getAbsolutePath());
    splParams.put("append", topology.createSubmissionParameter(submissionAppendName, submissionAppend));
    splParams.put("flush", SPL.createSubmissionParameter(topology, submissionFlushName, SPL.createValue(0, Type.MetaType.UINT32), false));
    SPL.invokeSink("spl.adapter::FileSink", splStream, splParams);
    
    // use a submission parameter in "inner" functional logic
    is = is.filter(thresholdFilter(threshold));
    is = is.filter(thresholdFilter(defaultedThreshold));

    // avoid another parallel impl limitation noted in issue#173
    is = is.filter(passthru());
    
    TStream<Integer> joined = is.endParallel();
    TStream<String> numRegions = joined.transform(
            uniqueIdentifierMap(count));

    Tester tester = topology.getTester();
    Condition<Long> expectedCount = tester.tupleCount(numRegions, 1);
    Condition<List<String>> regionCount = tester.stringContents(numRegions, submissionWidth.toString());

    Map<String,Object> params = new HashMap<>();
    params.put(submissionWidthName, submissionWidth);
    params.put(submissionFlushName, submissionFlush);
    params.put(submissionThresholdName, submissionThreshold);
    getConfig().put(ContextProperties.SUBMISSION_PARAMS, params);
	
    complete(tester, allConditions(expectedCount, regionCount), 10, TimeUnit.SECONDS);

    assertTrue(expectedCount.valid());
    assertTrue(regionCount.valid());
}
 

import com.ibm.streams.operator.Type; //导入依赖的package包/类
private void checkUdpSupported() {
    assumeTrue(SC_OK);
    assumeTrue(getTesterType() == StreamsContext.Type.STANDALONE_TESTER ||
            getTesterType() == StreamsContext.Type.DISTRIBUTED_TESTER);
}