com.ibm.streamsx.topology.spl

Class SPLStreams

java.lang.Object

com.ibm.streamsx.topology.spl.SPLStreams

public class SPLStreams
extends java.lang.Object

Utilities for SPL attribute schema streams.

Constructor Summary

Constructors

Constructor and Description
SPLStreams()

Method Summary

Modifier and Type	Method and Description
static <T> SPLStream	convertStream(TStream<T> stream, BiFunction<T,com.ibm.streams.operator.OutputTuple,com.ibm.streams.operator.OutputTuple> converter, com.ibm.streams.operator.StreamSchema schema) Convert a Stream to an SPLStream.
static SPLStream	stringToSPLStream(TStream<java.lang.String> stream) Represent stream as an SPLStream with schema SPLSchemas.STRING.
static SPLStream	subscribe(TopologyElement te, java.lang.String topic, com.ibm.streams.operator.StreamSchema schema) Subscribe to an SPLStream published by topic.
static SPLStream	subscribe(TopologyElement te, Supplier<java.lang.String> topic, com.ibm.streams.operator.StreamSchema schema) Subscribe to an SPLStream published by topic.
static TStream<java.lang.String>	toStringStream(SPLStream stream, java.lang.String attributeName) Convert an SPLStream to a TStream<String> by taking a specific attribute.
static TStream<java.lang.String>	toStringStream(SPLStream stream) Convert an SPLStream to a TStream<String> by taking its first attribute.
static SPLWindow	triggerCount(TWindow<com.ibm.streams.operator.Tuple,?> window, int count) Convert window to an SPL window with a count based trigger policy.
static SPLWindow	triggerTime(TWindow<com.ibm.streams.operator.Tuple,?> window, long time, java.util.concurrent.TimeUnit unit) Convert window to an SPL window with a time based trigger policy.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

SPLStreams

public SPLStreams()

Method Detail

subscribe

public static SPLStream subscribe(TopologyElement te,
                                  java.lang.String topic,
                                  com.ibm.streams.operator.StreamSchema schema)

Subscribe to an SPLStream published by topic.

Parameters:

te - Topology the stream will be contained in.

topic - Topic to subscribe to.

schema - SPL Schema of the published stream.

Returns:

Stream containing tuples for the published topic.

subscribe

public static SPLStream subscribe(TopologyElement te,
                                  Supplier<java.lang.String> topic,
                                  com.ibm.streams.operator.StreamSchema schema)

Subscribe to an SPLStream published by topic. Supports topic as a submission time parameter, for example using the topic defined by the submission parameter eventTopic.:

 
 Supplier topicParam = topology.createSubmissionParameter("eventTopic", String.class);
 SPLStream events = SPLStreams.subscribe(topology, topicParam, schema);
 
 

Parameters:

te - Topology the stream will be contained in.

topic - Topic to subscribe to.

schema - SPL Schema of the published stream.

Returns:

Stream containing tuples for the published topic.

Since:

1.8

See Also:

Topology.createSubmissionParameter(String, Class), Topology.createSubmissionParameter(String, Object)

convertStream

public static <T> SPLStream convertStream(TStream<T> stream,
                                          BiFunction<T,com.ibm.streams.operator.OutputTuple,com.ibm.streams.operator.OutputTuple> converter,
                                          com.ibm.streams.operator.StreamSchema schema)

Convert a Stream to an SPLStream. For each tuple t on stream, the returned stream will contain a tuple that is the result of converter.apply(t, outTuple) when the return is not null. outTuple is a newly created, empty, OutputTuple, the converter.apply() method populates outTuple from t.

Example of converting a stream containing a Sensor object to an SPL schema of tuple<rstring id, float64 reading>.

 
 Stream<Sensor> sensors = ...
 StreamSchema schema = Type.Factory.getStreamSchema("tuple<rstring id, float64 reading>");
 SPLStream splSensors = SPLStreams.convertStream(sensors,
   new BiFunction<Sensor, OutputTuple, OutputTuple>() {
             @Override
             public OutputTuple apply(Sensor sensor, OutputTuple outTuple) {
                 outTuple.setString("id", sensor.getId());
                 outTuple.setDouble("reading", sensor.getReading());
                 return outTuple;
             }}, schema);
 
 

Parameters:

stream - Stream to be converted.

converter - Converter used to populate the SPL tuple.

schema - Schema of returned SPLStream.

Returns:

SPLStream containing the converted tuples.

See Also:

SPLStream.convert(com.ibm.streamsx.topology.function.Function)

toStringStream

public static TStream<java.lang.String> toStringStream(SPLStream stream)

Convert an SPLStream to a TStream<String> by taking its first attribute. The returned stream will contain a String tuple for each tuple T on stream, the value of the String tuple is T.getString(0). A runtime error will occur if the first attribute (index 0) can not be converted using Tuple.getString(0).

Parameters:

stream - Stream to be converted to a TStream<String>.

Returns:

Stream that will contain the first attribute of tuples from stream

toStringStream

public static TStream<java.lang.String> toStringStream(SPLStream stream,
                                                       java.lang.String attributeName)

Convert an SPLStream to a TStream<String> by taking a specific attribute. The returned stream will contain a String tuple for each tuple T on stream, the value of the String tuple is T.getString(attributeName). A runtime error will occur if the attribute can not be converted using Tuple.getString(attributeName).

Parameters:

stream - Stream to be converted to a TStream<String>.

Returns:

Stream that will contain a single attribute of tuples from stream

stringToSPLStream

public static SPLStream stringToSPLStream(TStream<java.lang.String> stream)

Represent stream as an SPLStream with schema SPLSchemas.STRING.

Parameters:

stream - Stream to be represented as an SPLStream.

Returns:

SPLStream representation of stream.

triggerCount

public static SPLWindow triggerCount(TWindow<com.ibm.streams.operator.Tuple,?> window,
                                     int count)

Convert window to an SPL window with a count based trigger policy.

Parameters:

window - Window to be converted.

count - Count trigger policy value

Returns:

SPL window with that will trigger every count tuples.

triggerTime

public static SPLWindow triggerTime(TWindow<com.ibm.streams.operator.Tuple,?> window,
                                    long time,
                                    java.util.concurrent.TimeUnit unit)

Convert window to an SPL window with a time based trigger policy.

Parameters:

window - Window to be converted.

time - TIme trigger policy value.

unit - Unit for time.

Returns:

SPL window with that will trigger periodically according to time.