<commit_before>/*
# Licensed Materials - Property of IBM
# Copyright IBM Corp. 2015  
 */
package com.ibm.streamsx.topology;

import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import com.ibm.streamsx.topology.builder.BInputPort;
import com.ibm.streamsx.topology.builder.BOperatorInvocation;
import com.ibm.streamsx.topology.builder.BOutput;
import com.ibm.streamsx.topology.function7.BiFunction;
import com.ibm.streamsx.topology.function7.Consumer;
import com.ibm.streamsx.topology.function7.Function;
import com.ibm.streamsx.topology.function7.Predicate;
import com.ibm.streamsx.topology.function7.UnaryOperator;
import com.ibm.streamsx.topology.spl.SPLStream;
import com.ibm.streamsx.topology.tuple.Keyable;

/**
 * A {@code TStream} is a declaration of a continuous sequence of tuples. A
 * connected topology of streams and functional transformations is built using
 * {@link Topology}. <BR>
 * Generic methods on this interface provide the ability to
 * {@link #filter(Predicate) filter}, {@link #transform(Function, Class)
 * transform} or {@link #sink(Consumer) sink} this declared stream using a
 * function. <BR>
 * Utility methods in the {@code com.ibm.streams.topology.streams} package
 * provide specific source streams, or transformations on streams with specific
 * types.
 * 
 * @param <T>
 *            Tuple type, any instance of {@code T} at runtime must be
 *            serializable.
 */
public interface TStream<T> extends TopologyElement {

    /**
     * Enumeration for routing tuples to parallel channels.
     * @see TStream#parallel(int, Routing)
     */
	public enum Routing {
	   /**
	    * Tuples will be routed to parallel channels such that an even 
	    * distribution is maintained.
	    */
	    ROUND_ROBIN, 
	    /**
	     * Tuples will be consistently routed to the same channel based upon 
	     * their value.
	     * <br>
	     * If the tuple implements the {@link Keyable} interface, then the tuple
	     * is routed to a parallel channel according to the 
	     * {@code hashCode()} of the object returned by {@link Keyable#getKey()}. 
	     * <BR>
	     * If the tuple does not 
	     * implement the {@link Keyable} interface, then the tuple is routed to a parallel 
	     * channel according to the {@code hashCode()} of the tuple object itself.
	     */
	    PARTITIONED};
	
    /**
     * Declare a new stream that filters tuples from this stream. Each tuple
     * {@code t} on this stream will appear in the returned     stream if
     * {@link Predicate#test(Object) filter.test(t)} returns {@code true}. If
     * {@code filter.test(t)} returns {@code false} then then {@code t} will not
     * appear in the returned stream.
     * <P>
     * Example of filtering out all empty strings from stream {@code s} of type
     * {@code String}
     * 
     * <pre>
     * <code>
     * Stream&lt;String> s = ...
     * Stream&lt;String> filtered = s.filter(new Predicate&lt;String>() {
     *             &#64;Override
     *             public boolean test(String t) {
     *                 return !t.isEmpty();
     *             }} );
     * </code>
     * </pre>
     * 
     * </P>
     * 
     * @param filter
     *            Filtering logic to be executed against each tuple.
     * @return Filtered stream
     */
    TStream<T> filter(Predicate<T> filter);

    /**
     * Declare a new stream that transforms each tuple from this stream into one
     * (or zero) tuple of a different type {@code U}. For each tuple {@code t}
     * on this stream, the returned stream will contain a tuple that is the
     * result of {@code transformer.apply(t)} when the return is not {@code null}.
     * If {@code transformer.apply(t)} returns {@code null} then no tuple
     * is submitted to the returned stream for {@code t}.
     * 
     * <P>
     * Example of transforming a stream containing numeric values as
     * {@code String} objects into a stream of {@code Double} values.
     * 
     * <pre>
     * <code>
     * TStream&lt;String> strings = ...
     * TStream&lt;Double> doubles = strings.transform(new Function<String, Double>() {
     *             &#64;Override
     *             public Double apply(String v) {
     *                 return Double.valueOf(v);
     *             }}, Double.class );
     * </code>
     * </pre>
     * 
     * </P>
     * 
     * @param transformer
     *            Transformation logic to be executed against each tuple.
     * @param tupleTypeClass
     *            Type {@code U} of the returned stream.
     * @return Stream that will contain tuples of type {@code U} transformed from this
     *         stream's tuples.
     */
    <U> TStream<U> transform(Function<T, U> transformer, Class<U> tupleTypeClass);

    TStream<T> modify(UnaryOperator<T> modifier);

    /**
     * Declare a new stream that transforms tuples from this stream into one or
     * more (or zero) tuples of a different type {@code U}. For each tuple
     * {@code t} on this stream, the returned stream will contain all tuples in
     * the {@code Iterator<U>} that is the result of {@code transformer.apply(t)}.
     * Tuples will be added to the returned stream in the order the iterator
     * returns them.
     * 
     * <BR>
     * If the return is null or an empty iterator then no tuples are added to
     * the returned stream for input tuple {@code t}.
     * <P>
     * Example of transforming a stream containing lines of text into a stream
     * of words split out from each line. The order of the words in the stream
     * will match the order of the words in the lines.
     * 
     * <pre>
     * <code>
     * TStream&lt;String> lines = ...
     * TStream&lt;String> words = lines.multiTransform(new Function<String, Iterable<String>>() {
     *             &#64;Override
     *             public Iterable<String> apply(String t) {
     *                 return Arrays.asList(t.split(" "));
     *             }}, String.class);
     * </code>
     * </pre>
     * 
     * </P>
     * 
     * @param transformer
     *            Transformation logic to be executed against each tuple.
     * @param tupleTypeClass
     *            Type {@code U} of the returned stream.
     * @return Stream that will contain tuples of type {@code U} transformed from this
     *         stream's tuples.
     */
    <U> TStream<U> multiTransform(Function<T, Iterable<U>> transformer,
            Class<U> tupleTypeClass);

    /**
     * Sink (terminate) this stream. For each tuple {@code t} on this stream
     * {@link Consumer#accept(Object) sinker.accept(t)} will be called. This is
     * typically used to send information to external systems, such as databases
     * or dashboards.
     * <P>
     * Example of terminating a stream of {@code String} tuples by printing them
     * to {@code System.out}.
     * 
     * <pre>
     * <code>
     * Stream&lt;String> values = ...
     * values.sink(new FunctionV<String>() {
     *             
     *             &#64;Override
     *             public void accept(String v) {
     *                 System.out.println(v);
     *                 
     *             }
     *         });
     * </code>
     * </pre>
     * 
     * </P>
     * 
     * @param sinker
     *            Logic to be executed against each tuple on this stream.
     */<commit_msg>Cleanup Javadoc for sink & filter<commit_after>/*
# Licensed Materials - Property of IBM
# Copyright IBM Corp. 2015  
 */
package com.ibm.streamsx.topology;

import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import com.ibm.streamsx.topology.builder.BInputPort;
import com.ibm.streamsx.topology.builder.BOperatorInvocation;
import com.ibm.streamsx.topology.builder.BOutput;
import com.ibm.streamsx.topology.function7.BiFunction;
import com.ibm.streamsx.topology.function7.Consumer;
import com.ibm.streamsx.topology.function7.Function;
import com.ibm.streamsx.topology.function7.Predicate;
import com.ibm.streamsx.topology.function7.UnaryOperator;
import com.ibm.streamsx.topology.spl.SPLStream;
import com.ibm.streamsx.topology.tuple.Keyable;

/**
 * A {@code TStream} is a declaration of a continuous sequence of tuples. A
 * connected topology of streams and functional transformations is built using
 * {@link Topology}. <BR>
 * Generic methods on this interface provide the ability to
 * {@link #filter(Predicate) filter}, {@link #transform(Function, Class)
 * transform} or {@link #sink(Consumer) sink} this declared stream using a
 * function. <BR>
 * Utility methods in the {@code com.ibm.streams.topology.streams} package
 * provide specific source streams, or transformations on streams with specific
 * types.
 * 
 * @param <T>
 *            Tuple type, any instance of {@code T} at runtime must be
 *            serializable.
 */
public interface TStream<T> extends TopologyElement {

    /**
     * Enumeration for routing tuples to parallel channels.
     * @see TStream#parallel(int, Routing)
     */
	public enum Routing {
	   /**
	    * Tuples will be routed to parallel channels such that an even 
	    * distribution is maintained.
	    */
	    ROUND_ROBIN, 
	    /**
	     * Tuples will be consistently routed to the same channel based upon 
	     * their value.
	     * <br>
	     * If the tuple implements the {@link Keyable} interface, then the tuple
	     * is routed to a parallel channel according to the 
	     * {@code hashCode()} of the object returned by {@link Keyable#getKey()}. 
	     * <BR>
	     * If the tuple does not 
	     * implement the {@link Keyable} interface, then the tuple is routed to a parallel 
	     * channel according to the {@code hashCode()} of the tuple object itself.
	     */
	    PARTITIONED};
	
    /**
     * Declare a new stream that filters tuples from this stream. Each tuple
     * {@code t} on this stream will appear in the returned     stream if
     * {@link Predicate#test(Object) filter.test(t)} returns {@code true}. If
     * {@code filter.test(t)} returns {@code false} then then {@code t} will not
     * appear in the returned stream.
     * <P>
     * Example of filtering out all empty strings from stream {@code s} of type
     * {@code String}
     * 
     * <pre>
     * <code>
     * TStream&lt;String> s = ...
     * TStream&lt;String> filtered = s.filter(new Predicate&lt;String>() {
     *             &#64;Override
     *             public boolean test(String t) {
     *                 return !t.isEmpty();
     *             }} );
     * </code>
     * </pre>
     * 
     * </P>
     * 
     * @param filter
     *            Filtering logic to be executed against each tuple.
     * @return Filtered stream
     */
    TStream<T> filter(Predicate<T> filter);

    /**
     * Declare a new stream that transforms each tuple from this stream into one
     * (or zero) tuple of a different type {@code U}. For each tuple {@code t}
     * on this stream, the returned stream will contain a tuple that is the
     * result of {@code transformer.apply(t)} when the return is not {@code null}.
     * If {@code transformer.apply(t)} returns {@code null} then no tuple
     * is submitted to the returned stream for {@code t}.
     * 
     * <P>
     * Example of transforming a stream containing numeric values as
     * {@code String} objects into a stream of {@code Double} values.
     * 
     * <pre>
     * <code>
     * TStream&lt;String> strings = ...
     * TStream&lt;Double> doubles = strings.transform(new Function<String, Double>() {
     *             &#64;Override
     *             public Double apply(String v) {
     *                 return Double.valueOf(v);
     *             }}, Double.class );
     * </code>
     * </pre>
     * 
     * </P>
     * 
     * @param transformer
     *            Transformation logic to be executed against each tuple.
     * @param tupleTypeClass
     *            Type {@code U} of the returned stream.
     * @return Stream that will contain tuples of type {@code U} transformed from this
     *         stream's tuples.
     */
    <U> TStream<U> transform(Function<T, U> transformer, Class<U> tupleTypeClass);

    TStream<T> modify(UnaryOperator<T> modifier);

    /**
     * Declare a new stream that transforms tuples from this stream into one or
     * more (or zero) tuples of a different type {@code U}. For each tuple
     * {@code t} on this stream, the returned stream will contain all tuples in
     * the {@code Iterator<U>} that is the result of {@code transformer.apply(t)}.
     * Tuples will be added to the returned stream in the order the iterator
     * returns them.
     * 
     * <BR>
     * If the return is null or an empty iterator then no tuples are added to
     * the returned stream for input tuple {@code t}.
     * <P>
     * Example of transforming a stream containing lines of text into a stream
     * of words split out from each line. The order of the words in the stream
     * will match the order of the words in the lines.
     * 
     * <pre>
     * <code>
     * TStream&lt;String> lines = ...
     * TStream&lt;String> words = lines.multiTransform(new Function<String, Iterable<String>>() {
     *             &#64;Override
     *             public Iterable<String> apply(String t) {
     *                 return Arrays.asList(t.split(" "));
     *             }}, String.class);
     * </code>
     * </pre>
     * 
     * </P>
     * 
     * @param transformer
     *            Transformation logic to be executed against each tuple.
     * @param tupleTypeClass
     *            Type {@code U} of the returned stream.
     * @return Stream that will contain tuples of type {@code U} transformed from this
     *         stream's tuples.
     */
    <U> TStream<U> multiTransform(Function<T, Iterable<U>> transformer,
            Class<U> tupleTypeClass);

    /**
     * Sink (terminate) this stream. For each tuple {@code t} on this stream
     * {@link Consumer#accept(Object) sinker.accept(t)} will be called. This is
     * typically used to send information to external systems, such as databases
     * or dashboards.
     * <P>
     * Example of terminating a stream of {@code String} tuples by printing them
     * to {@code System.out}.
     * 
     * <pre>
     * <code>
     * TStream&lt;String> values = ...
     * values.sink(new Consumer<String>() {
     *             
     *             &#64;Override
     *             public void accept(String v) {
     *                 System.out.println(v);
     *                 
     *             }
     *         });
     * </code>
     * </pre>
     * 
     * </P>
     * 
     * @param sinker
     *            Logic to be executed against each tuple on this stream.
     */
{'fraction_non_alphanumeric': 0.10493743199129489, 'fraction_numerical': 0.002312295973884657, 'mean_word_length': 2.250442086648983, 'pattern_counts': {'":': 0, '<': 103, '<?xml version=': 0, '>': 119, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7651868', 'n_tokens_mistral': 4285, 'n_tokens_neox': 4110, 'n_words': 1490}