flink-libraries/flink-cep/src/main/java/org/apache/flink/cep/PatternStream.java - flink - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.flink.cep;

 import org.apache.flink.annotation.Internal;
 import org.apache.flink.api.common.ExecutionConfig;
 import org.apache.flink.api.common.typeinfo.TypeInformation;
 import org.apache.flink.api.java.typeutils.EitherTypeInfo;
 import org.apache.flink.api.java.typeutils.TypeExtractor;
 import org.apache.flink.cep.operator.CEPOperatorUtils;
 import org.apache.flink.cep.pattern.Pattern;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
 import org.apache.flink.streaming.api.functions.co.CoMapFunction;
 import org.apache.flink.types.Either;
 import org.apache.flink.util.OutputTag;

 import java.util.UUID;

 /**
  * Stream abstraction for CEP pattern detection. A pattern stream is a stream which emits detected
  * pattern sequences as a map of events associated with their names. The pattern is detected using a
  * {@link org.apache.flink.cep.nfa.NFA}. In order to process the detected sequences, the user
  * has to specify a {@link PatternSelectFunction} or a {@link PatternFlatSelectFunction}.
  *
  * <p>Additionally it allows to handle partially matched event patterns which have timed out. For this
  * the user has to specify a {@link PatternTimeoutFunction} or a {@link PatternFlatTimeoutFunction}.
  *
  * @param <T> Type of the events
  */
 public class PatternStream<T> {

 	// underlying data stream
 	private final DataStream<T> inputStream;

 	private final Pattern<T, ?> pattern;

 	// comparator to sort events
 	private final EventComparator<T> comparator;

 	/**
 	 * Side output {@code OutputTag} for late data. If no tag is set late data will be simply
 	 * dropped.
 	 */
 	private OutputTag<T> lateDataOutputTag;

 	PatternStream(final DataStream<T> inputStream, final Pattern<T, ?> pattern) {
 		this.inputStream = inputStream;
 		this.pattern = pattern;
 		this.comparator = null;
 	}

 	PatternStream(final DataStream<T> inputStream, final Pattern<T, ?> pattern, final EventComparator<T> comparator) {
 		this.inputStream = inputStream;
 		this.pattern = pattern;
 		this.comparator = comparator;
 	}

 	public Pattern<T, ?> getPattern() {
 		return pattern;
 	}

 	public DataStream<T> getInputStream() {
 		return inputStream;
 	}

 	public EventComparator<T> getComparator() {
 		return comparator;
 	}

 	/**
 	 * Applies a select function to the detected pattern sequence. For each pattern sequence the
 	 * provided {@link PatternSelectFunction} is called. The pattern select function can produce
 	 * exactly one resulting element.
 	 *
 	 * @param patternSelectFunction The pattern select function which is called for each detected
 	 *                              pattern sequence.
 	 * @param <R> Type of the resulting elements
 	 * @return {@link DataStream} which contains the resulting elements from the pattern select
 	 *         function.
 	 */
 	public <R> SingleOutputStreamOperator<R> select(final PatternSelectFunction<T, R> patternSelectFunction) {
 		// we have to extract the output type from the provided pattern selection function manually
 		// because the TypeExtractor cannot do that if the method is wrapped in a MapFunction

 		TypeInformation<R> returnType = TypeExtractor.getUnaryOperatorReturnType(
 			patternSelectFunction,
 			PatternSelectFunction.class,
 			0,
 			1,
 			new int[]{0, 1, 0},
 			new int[]{},
 			inputStream.getType(),
 			null,
 			false);

 		return select(patternSelectFunction, returnType);
 	}

 	/**
 	 * Invokes the {@link org.apache.flink.api.java.ClosureCleaner}
 	 * on the given function if closure cleaning is enabled in the {@link ExecutionConfig}.
 	 *
 	 * @return The cleaned Function
 	 */
 	private  <F> F clean(F f) {
 		return inputStream.getExecutionEnvironment().clean(f);
 	}

 	/**
 	 * Applies a select function to the detected pattern sequence. For each pattern sequence the
 	 * provided {@link PatternSelectFunction} is called. The pattern select function can produce
 	 * exactly one resulting element.
 	 *
 	 * @param patternSelectFunction The pattern select function which is called for each detected
 	 *                              pattern sequence.
 	 * @param <R> Type of the resulting elements
 	 * @param outTypeInfo Explicit specification of output type.
 	 * @return {@link DataStream} which contains the resulting elements from the pattern select
 	 *         function.
 	 */
 	public <R> SingleOutputStreamOperator<R> select(final PatternSelectFunction<T, R> patternSelectFunction, TypeInformation<R> outTypeInfo) {
 		return CEPOperatorUtils.createPatternStream(inputStream, pattern, comparator, clean(patternSelectFunction), outTypeInfo, lateDataOutputTag);
 	}

 	/**
 	 * Applies a select function to the detected pattern sequence. For each pattern sequence the
 	 * provided {@link PatternSelectFunction} is called. The pattern select function can produce
 	 * exactly one resulting element.
 	 *
 	 * <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
 	 * partial pattern sequence the provided {@link PatternTimeoutFunction} is called. The pattern
 	 * timeout function can produce exactly one resulting element.
 	 *
 	 * <p>You can get the stream of timed-out data resulting from the
 	 * {@link SingleOutputStreamOperator#getSideOutput(OutputTag)} on the
 	 * {@link SingleOutputStreamOperator} resulting from the select operation
 	 * with the same {@link OutputTag}.
 	 *
 	 * @param timeoutOutputTag {@link OutputTag} that identifies side output with timed out patterns
 	 * @param patternTimeoutFunction The pattern timeout function which is called for each partial
 	 *                               pattern sequence which has timed out.
 	 * @param patternSelectFunction The pattern select function which is called for each detected
 	 *                              pattern sequence.
 	 * @param <L> Type of the resulting timeout elements
 	 * @param <R> Type of the resulting elements
 	 * @return {@link DataStream} which contains the resulting elements with the resulting timeout
 	 * elements in a side output.
 	 */
 	public <L, R> SingleOutputStreamOperator<R> select(
 		final OutputTag<L> timeoutOutputTag,
 		final PatternTimeoutFunction<T, L> patternTimeoutFunction,
 		final PatternSelectFunction<T, R> patternSelectFunction) {

 		TypeInformation<R> rightTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
 			patternSelectFunction,
 			PatternSelectFunction.class,
 			0,
 			1,
 			new int[]{0, 1, 0},
 			new int[]{},
 			inputStream.getType(),
 			null,
 			false);

 		return select(
 			timeoutOutputTag,
 			patternTimeoutFunction,
 			rightTypeInfo,
 			patternSelectFunction);
 	}

 	/**
 	 * Applies a select function to the detected pattern sequence. For each pattern sequence the
 	 * provided {@link PatternSelectFunction} is called. The pattern select function can produce
 	 * exactly one resulting element.
 	 *
 	 * <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
 	 * partial pattern sequence the provided {@link PatternTimeoutFunction} is called. The pattern
 	 * timeout function can produce exactly one resulting element.
 	 *
 	 * <p>You can get the stream of timed-out data resulting from the
 	 * {@link SingleOutputStreamOperator#getSideOutput(OutputTag)} on the
 	 * {@link SingleOutputStreamOperator} resulting from the select operation
 	 * with the same {@link OutputTag}.
 	 *
 	 * @param timeoutOutputTag {@link OutputTag} that identifies side output with timed out patterns
 	 * @param patternTimeoutFunction The pattern timeout function which is called for each partial
 	 *                               pattern sequence which has timed out.
 	 * @param outTypeInfo Explicit specification of output type.
 	 * @param patternSelectFunction The pattern select function which is called for each detected
 	 *                              pattern sequence.
 	 * @param <L> Type of the resulting timeout elements
 	 * @param <R> Type of the resulting elements
 	 * @return {@link DataStream} which contains the resulting elements with the resulting timeout
 	 * elements in a side output.
 	 */
 	public <L, R> SingleOutputStreamOperator<R> select(
 			final OutputTag<L> timeoutOutputTag,
 			final PatternTimeoutFunction<T, L> patternTimeoutFunction,
 			final TypeInformation<R> outTypeInfo,
 			final PatternSelectFunction<T, R> patternSelectFunction) {
 		return CEPOperatorUtils.createTimeoutPatternStream(
 			inputStream,
 			pattern,
 			comparator,
 			clean(patternSelectFunction),
 			outTypeInfo,
 			timeoutOutputTag,
 			clean(patternTimeoutFunction),
 			lateDataOutputTag);
 	}

 	/**
 	 * Applies a select function to the detected pattern sequence. For each pattern sequence the
 	 * provided {@link PatternSelectFunction} is called. The pattern select function can produce
 	 * exactly one resulting element.
 	 *
 	 * <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
 	 * partial pattern sequence the provided {@link PatternTimeoutFunction} is called. The pattern
 	 * timeout function can produce exactly one resulting element.
 	 *
 	 * @param patternTimeoutFunction The pattern timeout function which is called for each partial
 	 *                               pattern sequence which has timed out.
 	 * @param patternSelectFunction The pattern select function which is called for each detected
 	 *                              pattern sequence.
 	 * @param <L> Type of the resulting timeout elements
 	 * @param <R> Type of the resulting elements
 	 *
 	 * @deprecated Use {@link PatternStream#select(OutputTag, PatternTimeoutFunction, PatternSelectFunction)}
 	 * that returns timed out events as a side-output
 	 *
 	 * @return {@link DataStream} which contains the resulting elements or the resulting timeout
 	 * elements wrapped in an {@link Either} type.
 	 */
 	@Deprecated
 	public <L, R> SingleOutputStreamOperator<Either<L, R>> select(
 		final PatternTimeoutFunction<T, L> patternTimeoutFunction,
 		final PatternSelectFunction<T, R> patternSelectFunction) {

 		TypeInformation<R> rightTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
 			patternSelectFunction,
 			PatternSelectFunction.class,
 			0,
 			1,
 			new int[]{0, 1, 0},
 			new int[]{},
 			inputStream.getType(),
 			null,
 			false);

 		TypeInformation<L> leftTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
 			patternTimeoutFunction,
 			PatternTimeoutFunction.class,
 			0,
 			1,
 			new int[]{0, 1, 0},
 			new int[]{},
 			inputStream.getType(),
 			null,
 			false);

 		final OutputTag<L> outputTag = new OutputTag<L>(UUID.randomUUID().toString(), leftTypeInfo);

 		final SingleOutputStreamOperator<R> mainStream = CEPOperatorUtils.createTimeoutPatternStream(
 			inputStream,
 			pattern,
 			comparator,
 			clean(patternSelectFunction),
 			rightTypeInfo,
 			outputTag,
 			clean(patternTimeoutFunction),
 			lateDataOutputTag);

 		final DataStream<L> timedOutStream = mainStream.getSideOutput(outputTag);

 		TypeInformation<Either<L, R>> outTypeInfo = new EitherTypeInfo<>(leftTypeInfo, rightTypeInfo);

 		return mainStream.connect(timedOutStream).map(new CoMapTimeout<>()).returns(outTypeInfo);
 	}

 	/**
 	 * Applies a flat select function to the detected pattern sequence. For each pattern sequence
 	 * the provided {@link PatternFlatSelectFunction} is called. The pattern flat select function
 	 * can produce an arbitrary number of resulting elements.
 	 *
 	 * @param patternFlatSelectFunction The pattern flat select function which is called for each
 	 *                                  detected pattern sequence.
 	 * @param <R> Type of the resulting elements
 	 * @return {@link DataStream} which contains the resulting elements from the pattern flat select
 	 *         function.
 	 */
 	public <R> SingleOutputStreamOperator<R> flatSelect(final PatternFlatSelectFunction<T, R> patternFlatSelectFunction) {
 		// we have to extract the output type from the provided pattern selection function manually
 		// because the TypeExtractor cannot do that if the method is wrapped in a MapFunction
 		TypeInformation<R> outTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
 			patternFlatSelectFunction,
 			PatternFlatSelectFunction.class,
 			0,
 			1,
 			new int[] {0, 1, 0},
 			new int[] {1, 0},
 			inputStream.getType(),
 			null,
 			false);

 		return flatSelect(patternFlatSelectFunction, outTypeInfo);
 	}

 	/**
 	 * Applies a flat select function to the detected pattern sequence. For each pattern sequence
 	 * the provided {@link PatternFlatSelectFunction} is called. The pattern flat select function
 	 * can produce an arbitrary number of resulting elements.
 	 *
 	 * @param patternFlatSelectFunction The pattern flat select function which is called for each
 	 *                                  detected pattern sequence.
 	 * @param <R> Type of the resulting elements
 	 * @param outTypeInfo Explicit specification of output type.
 	 * @return {@link DataStream} which contains the resulting elements from the pattern flat select
 	 *         function.
 	 */
 	public <R> SingleOutputStreamOperator<R> flatSelect(
 			final PatternFlatSelectFunction<T, R> patternFlatSelectFunction,
 			final TypeInformation<R> outTypeInfo) {
 		return CEPOperatorUtils.createPatternStream(
 			inputStream,
 			pattern,
 			comparator,
 			clean(patternFlatSelectFunction),
 			outTypeInfo,
 			lateDataOutputTag);
 	}

 	/**
 	 * Applies a flat select function to the detected pattern sequence. For each pattern sequence the
 	 * provided {@link PatternFlatSelectFunction} is called. The pattern select function can produce
 	 * exactly one resulting element.
 	 *
 	 * <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
 	 * partial pattern sequence the provided {@link PatternFlatTimeoutFunction} is called. The pattern
 	 * timeout function can produce exactly one resulting element.
 	 *
 	 * <p>You can get the stream of timed-out data resulting from the
 	 * {@link SingleOutputStreamOperator#getSideOutput(OutputTag)} on the
 	 * {@link SingleOutputStreamOperator} resulting from the select operation
 	 * with the same {@link OutputTag}.
 	 *
 	 * @param timeoutOutputTag {@link OutputTag} that identifies side output with timed out patterns
 	 * @param patternFlatTimeoutFunction The pattern timeout function which is called for each partial
 	 *                               pattern sequence which has timed out.
 	 * @param patternFlatSelectFunction The pattern select function which is called for each detected
 	 *                              pattern sequence.
 	 * @param <L> Type of the resulting timeout elements
 	 * @param <R> Type of the resulting elements
 	 * @return {@link DataStream} which contains the resulting elements with the resulting timeout
 	 * elements in a side output.
 	 */
 	public <L, R> SingleOutputStreamOperator<R> flatSelect(
 		final OutputTag<L> timeoutOutputTag,
 		final PatternFlatTimeoutFunction<T, L> patternFlatTimeoutFunction,
 		final PatternFlatSelectFunction<T, R> patternFlatSelectFunction) {

 		TypeInformation<R> rightTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
 			patternFlatSelectFunction,
 			PatternFlatSelectFunction.class,
 			0,
 			1,
 			new int[]{0, 1, 0},
 			new int[]{1, 0},
 			inputStream.getType(),
 			null,
 			false);

 		return flatSelect(timeoutOutputTag, patternFlatTimeoutFunction, rightTypeInfo, patternFlatSelectFunction);
 	}

 	/**
 	 * Applies a flat select function to the detected pattern sequence. For each pattern sequence the
 	 * provided {@link PatternFlatSelectFunction} is called. The pattern select function can produce
 	 * exactly one resulting element.
 	 *
 	 * <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
 	 * partial pattern sequence the provided {@link PatternFlatTimeoutFunction} is called. The pattern
 	 * timeout function can produce exactly one resulting element.
 	 *
 	 * <p>You can get the stream of timed-out data resulting from the
 	 * {@link SingleOutputStreamOperator#getSideOutput(OutputTag)} on the
 	 * {@link SingleOutputStreamOperator} resulting from the select operation
 	 * with the same {@link OutputTag}.
 	 *
 	 * @param timeoutOutputTag {@link OutputTag} that identifies side output with timed out patterns
 	 * @param patternFlatTimeoutFunction The pattern timeout function which is called for each partial
 	 *                               pattern sequence which has timed out.
 	 * @param patternFlatSelectFunction The pattern select function which is called for each detected
 	 *                              pattern sequence.
 	 * @param outTypeInfo Explicit specification of output type.
 	 * @param <L> Type of the resulting timeout elements
 	 * @param <R> Type of the resulting elements
 	 * @return {@link DataStream} which contains the resulting elements with the resulting timeout
 	 * elements in a side output.
 	 */
 	public <L, R> SingleOutputStreamOperator<R> flatSelect(
 		final OutputTag<L> timeoutOutputTag,
 		final PatternFlatTimeoutFunction<T, L> patternFlatTimeoutFunction,
 		final TypeInformation<R> outTypeInfo,
 		final PatternFlatSelectFunction<T, R> patternFlatSelectFunction) {

 		return CEPOperatorUtils.createTimeoutPatternStream(
 			inputStream,
 			pattern,
 			comparator,
 			clean(patternFlatSelectFunction),
 			outTypeInfo,
 			timeoutOutputTag,
 			clean(patternFlatTimeoutFunction),
 			lateDataOutputTag);
 	}

 	/**
 	 * Applies a flat select function to the detected pattern sequence. For each pattern sequence
 	 * the provided {@link PatternFlatSelectFunction} is called. The pattern flat select function
 	 * can produce an arbitrary number of resulting elements.
 	 *
 	 * <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
 	 * partial pattern sequence the provided {@link PatternFlatTimeoutFunction} is called. The
 	 * pattern timeout function can produce an arbitrary number of resulting elements.
 	 *
 	 * @param patternFlatTimeoutFunction The pattern flat timeout function which is called for each
 	 *                                   partial pattern sequence which has timed out.
 	 * @param patternFlatSelectFunction The pattern flat select function which is called for each
 	 *                                  detected pattern sequence.
 	 * @param <L> Type of the resulting timeout events
 	 * @param <R> Type of the resulting events
 	 *
 	 * @deprecated Use {@link PatternStream#flatSelect(OutputTag, PatternFlatTimeoutFunction, PatternFlatSelectFunction)}
 	 * that returns timed out events as a side-output
 	 *
 	 * @return {@link DataStream} which contains the resulting events from the pattern flat select
 	 * function or the resulting timeout events from the pattern flat timeout function wrapped in an
 	 * {@link Either} type.
 	 */
 	@Deprecated
 	public <L, R> SingleOutputStreamOperator<Either<L, R>> flatSelect(
 		final PatternFlatTimeoutFunction<T, L> patternFlatTimeoutFunction,
 		final PatternFlatSelectFunction<T, R> patternFlatSelectFunction) {

 		TypeInformation<L> leftTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
 			patternFlatTimeoutFunction,
 			PatternFlatTimeoutFunction.class,
 			0,
 			1,
 			new int[]{0, 1, 0},
 			new int[]{2, 0},
 			inputStream.getType(),
 			null,
 			false);

 		TypeInformation<R> rightTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
 			patternFlatSelectFunction,
 			PatternFlatSelectFunction.class,
 			0,
 			1,
 			new int[]{0, 1, 0},
 			new int[]{1, 0},
 			inputStream.getType(),
 			null,
 			false);

 		final OutputTag<L> outputTag = new OutputTag<L>(UUID.randomUUID().toString(), leftTypeInfo);

 		final SingleOutputStreamOperator<R> mainStream = CEPOperatorUtils.createTimeoutPatternStream(
 			inputStream,
 			pattern,
 			comparator,
 			clean(patternFlatSelectFunction),
 			rightTypeInfo,
 			outputTag,
 			clean(patternFlatTimeoutFunction),
 			lateDataOutputTag);

 		final DataStream<L> timedOutStream = mainStream.getSideOutput(outputTag);

 		TypeInformation<Either<L, R>> outTypeInfo = new EitherTypeInfo<>(leftTypeInfo, rightTypeInfo);

 		return mainStream.connect(timedOutStream).map(new CoMapTimeout<>()).returns(outTypeInfo);
 	}

 	public PatternStream<T> sideOutputLateData(OutputTag<T> outputTag) {
 		this.lateDataOutputTag = clean(outputTag);
 		return this;
 	}

 	/**
 	 * Used for joining results from timeout side-output for API backward compatibility.
 	 */
 	@Internal
 	public static class CoMapTimeout<R, L> implements CoMapFunction<R, L, Either<L, R>> {

 		private static final long serialVersionUID = 2059391566945212552L;

 		@Override
 		public Either<L, R> map1(R value) throws Exception {
 			return Either.Right(value);
 		}

 		@Override
 		public Either<L, R> map2(L value) throws Exception {
 			return Either.Left(value);
 		}
 	}
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.flink.cep;

	import org.apache.flink.annotation.Internal;
	import org.apache.flink.api.common.ExecutionConfig;
	import org.apache.flink.api.common.typeinfo.TypeInformation;
	import org.apache.flink.api.java.typeutils.EitherTypeInfo;
	import org.apache.flink.api.java.typeutils.TypeExtractor;
	import org.apache.flink.cep.operator.CEPOperatorUtils;
	import org.apache.flink.cep.pattern.Pattern;
	import org.apache.flink.streaming.api.datastream.DataStream;
	import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
	import org.apache.flink.streaming.api.functions.co.CoMapFunction;
	import org.apache.flink.types.Either;
	import org.apache.flink.util.OutputTag;

	import java.util.UUID;

	/**
	* Stream abstraction for CEP pattern detection. A pattern stream is a stream which emits detected
	* pattern sequences as a map of events associated with their names. The pattern is detected using a
	* {@link org.apache.flink.cep.nfa.NFA}. In order to process the detected sequences, the user
	* has to specify a {@link PatternSelectFunction} or a {@link PatternFlatSelectFunction}.
	*
	* <p>Additionally it allows to handle partially matched event patterns which have timed out. For this
	* the user has to specify a {@link PatternTimeoutFunction} or a {@link PatternFlatTimeoutFunction}.
	*
	* @param <T> Type of the events
	*/
	public class PatternStream<T> {

	// underlying data stream
	private final DataStream<T> inputStream;

	private final Pattern<T, ?> pattern;

	// comparator to sort events
	private final EventComparator<T> comparator;

	/**
	* Side output {@code OutputTag} for late data. If no tag is set late data will be simply
	* dropped.
	*/
	private OutputTag<T> lateDataOutputTag;

	PatternStream(final DataStream<T> inputStream, final Pattern<T, ?> pattern) {
	this.inputStream = inputStream;
	this.pattern = pattern;
	this.comparator = null;
	}

	PatternStream(final DataStream<T> inputStream, final Pattern<T, ?> pattern, final EventComparator<T> comparator) {
	this.inputStream = inputStream;
	this.pattern = pattern;
	this.comparator = comparator;
	}

	public Pattern<T, ?> getPattern() {
	return pattern;
	}

	public DataStream<T> getInputStream() {
	return inputStream;
	}

	public EventComparator<T> getComparator() {
	return comparator;
	}

	/**
	* Applies a select function to the detected pattern sequence. For each pattern sequence the
	* provided {@link PatternSelectFunction} is called. The pattern select function can produce
	* exactly one resulting element.
	*
	* @param patternSelectFunction The pattern select function which is called for each detected
	* pattern sequence.
	* @param <R> Type of the resulting elements
	* @return {@link DataStream} which contains the resulting elements from the pattern select
	* function.
	*/
	public <R> SingleOutputStreamOperator<R> select(final PatternSelectFunction<T, R> patternSelectFunction) {
	// we have to extract the output type from the provided pattern selection function manually
	// because the TypeExtractor cannot do that if the method is wrapped in a MapFunction

	TypeInformation<R> returnType = TypeExtractor.getUnaryOperatorReturnType(
	patternSelectFunction,
	PatternSelectFunction.class,
	0,
	1,
	new int[]{0, 1, 0},
	new int[]{},
	inputStream.getType(),
	null,
	false);

	return select(patternSelectFunction, returnType);
	}

	/**
	* Invokes the {@link org.apache.flink.api.java.ClosureCleaner}
	* on the given function if closure cleaning is enabled in the {@link ExecutionConfig}.
	*
	* @return The cleaned Function
	*/
	private <F> F clean(F f) {
	return inputStream.getExecutionEnvironment().clean(f);
	}

	/**
	* Applies a select function to the detected pattern sequence. For each pattern sequence the
	* provided {@link PatternSelectFunction} is called. The pattern select function can produce
	* exactly one resulting element.
	*
	* @param patternSelectFunction The pattern select function which is called for each detected
	* pattern sequence.
	* @param <R> Type of the resulting elements
	* @param outTypeInfo Explicit specification of output type.
	* @return {@link DataStream} which contains the resulting elements from the pattern select
	* function.
	*/
	public <R> SingleOutputStreamOperator<R> select(final PatternSelectFunction<T, R> patternSelectFunction, TypeInformation<R> outTypeInfo) {
	return CEPOperatorUtils.createPatternStream(inputStream, pattern, comparator, clean(patternSelectFunction), outTypeInfo, lateDataOutputTag);
	}

	/**
	* Applies a select function to the detected pattern sequence. For each pattern sequence the
	* provided {@link PatternSelectFunction} is called. The pattern select function can produce
	* exactly one resulting element.
	*
	* <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
	* partial pattern sequence the provided {@link PatternTimeoutFunction} is called. The pattern
	* timeout function can produce exactly one resulting element.
	*
	* <p>You can get the stream of timed-out data resulting from the
	* {@link SingleOutputStreamOperator#getSideOutput(OutputTag)} on the
	* {@link SingleOutputStreamOperator} resulting from the select operation
	* with the same {@link OutputTag}.
	*
	* @param timeoutOutputTag {@link OutputTag} that identifies side output with timed out patterns
	* @param patternTimeoutFunction The pattern timeout function which is called for each partial
	* pattern sequence which has timed out.
	* @param patternSelectFunction The pattern select function which is called for each detected
	* pattern sequence.
	* @param <L> Type of the resulting timeout elements
	* @param <R> Type of the resulting elements
	* @return {@link DataStream} which contains the resulting elements with the resulting timeout
	* elements in a side output.
	*/
	public <L, R> SingleOutputStreamOperator<R> select(
	final OutputTag<L> timeoutOutputTag,
	final PatternTimeoutFunction<T, L> patternTimeoutFunction,
	final PatternSelectFunction<T, R> patternSelectFunction) {

	TypeInformation<R> rightTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
	patternSelectFunction,
	PatternSelectFunction.class,
	0,
	1,
	new int[]{0, 1, 0},
	new int[]{},
	inputStream.getType(),
	null,
	false);

	return select(
	timeoutOutputTag,
	patternTimeoutFunction,
	rightTypeInfo,
	patternSelectFunction);
	}

	/**
	* Applies a select function to the detected pattern sequence. For each pattern sequence the
	* provided {@link PatternSelectFunction} is called. The pattern select function can produce
	* exactly one resulting element.
	*
	* <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
	* partial pattern sequence the provided {@link PatternTimeoutFunction} is called. The pattern
	* timeout function can produce exactly one resulting element.
	*
	* <p>You can get the stream of timed-out data resulting from the
	* {@link SingleOutputStreamOperator#getSideOutput(OutputTag)} on the
	* {@link SingleOutputStreamOperator} resulting from the select operation
	* with the same {@link OutputTag}.
	*
	* @param timeoutOutputTag {@link OutputTag} that identifies side output with timed out patterns
	* @param patternTimeoutFunction The pattern timeout function which is called for each partial
	* pattern sequence which has timed out.
	* @param outTypeInfo Explicit specification of output type.
	* @param patternSelectFunction The pattern select function which is called for each detected
	* pattern sequence.
	* @param <L> Type of the resulting timeout elements
	* @param <R> Type of the resulting elements
	* @return {@link DataStream} which contains the resulting elements with the resulting timeout
	* elements in a side output.
	*/
	public <L, R> SingleOutputStreamOperator<R> select(
	final OutputTag<L> timeoutOutputTag,
	final PatternTimeoutFunction<T, L> patternTimeoutFunction,
	final TypeInformation<R> outTypeInfo,
	final PatternSelectFunction<T, R> patternSelectFunction) {
	return CEPOperatorUtils.createTimeoutPatternStream(
	inputStream,
	pattern,
	comparator,
	clean(patternSelectFunction),
	outTypeInfo,
	timeoutOutputTag,
	clean(patternTimeoutFunction),
	lateDataOutputTag);
	}

	/**
	* Applies a select function to the detected pattern sequence. For each pattern sequence the
	* provided {@link PatternSelectFunction} is called. The pattern select function can produce
	* exactly one resulting element.
	*
	* <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
	* partial pattern sequence the provided {@link PatternTimeoutFunction} is called. The pattern
	* timeout function can produce exactly one resulting element.
	*
	* @param patternTimeoutFunction The pattern timeout function which is called for each partial
	* pattern sequence which has timed out.
	* @param patternSelectFunction The pattern select function which is called for each detected
	* pattern sequence.
	* @param <L> Type of the resulting timeout elements
	* @param <R> Type of the resulting elements
	*
	* @deprecated Use {@link PatternStream#select(OutputTag, PatternTimeoutFunction, PatternSelectFunction)}
	* that returns timed out events as a side-output
	*
	* @return {@link DataStream} which contains the resulting elements or the resulting timeout
	* elements wrapped in an {@link Either} type.
	*/
	@Deprecated
	public <L, R> SingleOutputStreamOperator<Either<L, R>> select(
	final PatternTimeoutFunction<T, L> patternTimeoutFunction,
	final PatternSelectFunction<T, R> patternSelectFunction) {

	TypeInformation<R> rightTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
	patternSelectFunction,
	PatternSelectFunction.class,
	0,
	1,
	new int[]{0, 1, 0},
	new int[]{},
	inputStream.getType(),
	null,
	false);

	TypeInformation<L> leftTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
	patternTimeoutFunction,
	PatternTimeoutFunction.class,
	0,
	1,
	new int[]{0, 1, 0},
	new int[]{},
	inputStream.getType(),
	null,
	false);

	final OutputTag<L> outputTag = new OutputTag<L>(UUID.randomUUID().toString(), leftTypeInfo);

	final SingleOutputStreamOperator<R> mainStream = CEPOperatorUtils.createTimeoutPatternStream(
	inputStream,
	pattern,
	comparator,
	clean(patternSelectFunction),
	rightTypeInfo,
	outputTag,
	clean(patternTimeoutFunction),
	lateDataOutputTag);

	final DataStream<L> timedOutStream = mainStream.getSideOutput(outputTag);

	TypeInformation<Either<L, R>> outTypeInfo = new EitherTypeInfo<>(leftTypeInfo, rightTypeInfo);

	return mainStream.connect(timedOutStream).map(new CoMapTimeout<>()).returns(outTypeInfo);
	}

	/**
	* Applies a flat select function to the detected pattern sequence. For each pattern sequence
	* the provided {@link PatternFlatSelectFunction} is called. The pattern flat select function
	* can produce an arbitrary number of resulting elements.
	*
	* @param patternFlatSelectFunction The pattern flat select function which is called for each
	* detected pattern sequence.
	* @param <R> Type of the resulting elements
	* @return {@link DataStream} which contains the resulting elements from the pattern flat select
	* function.
	*/
	public <R> SingleOutputStreamOperator<R> flatSelect(final PatternFlatSelectFunction<T, R> patternFlatSelectFunction) {
	// we have to extract the output type from the provided pattern selection function manually
	// because the TypeExtractor cannot do that if the method is wrapped in a MapFunction
	TypeInformation<R> outTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
	patternFlatSelectFunction,
	PatternFlatSelectFunction.class,
	0,
	1,
	new int[] {0, 1, 0},
	new int[] {1, 0},
	inputStream.getType(),
	null,
	false);

	return flatSelect(patternFlatSelectFunction, outTypeInfo);
	}

	/**
	* Applies a flat select function to the detected pattern sequence. For each pattern sequence
	* the provided {@link PatternFlatSelectFunction} is called. The pattern flat select function
	* can produce an arbitrary number of resulting elements.
	*
	* @param patternFlatSelectFunction The pattern flat select function which is called for each
	* detected pattern sequence.
	* @param <R> Type of the resulting elements
	* @param outTypeInfo Explicit specification of output type.
	* @return {@link DataStream} which contains the resulting elements from the pattern flat select
	* function.
	*/
	public <R> SingleOutputStreamOperator<R> flatSelect(
	final PatternFlatSelectFunction<T, R> patternFlatSelectFunction,
	final TypeInformation<R> outTypeInfo) {
	return CEPOperatorUtils.createPatternStream(
	inputStream,
	pattern,
	comparator,
	clean(patternFlatSelectFunction),
	outTypeInfo,
	lateDataOutputTag);
	}

	/**
	* Applies a flat select function to the detected pattern sequence. For each pattern sequence the
	* provided {@link PatternFlatSelectFunction} is called. The pattern select function can produce
	* exactly one resulting element.
	*
	* <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
	* partial pattern sequence the provided {@link PatternFlatTimeoutFunction} is called. The pattern
	* timeout function can produce exactly one resulting element.
	*
	* <p>You can get the stream of timed-out data resulting from the
	* {@link SingleOutputStreamOperator#getSideOutput(OutputTag)} on the
	* {@link SingleOutputStreamOperator} resulting from the select operation
	* with the same {@link OutputTag}.
	*
	* @param timeoutOutputTag {@link OutputTag} that identifies side output with timed out patterns
	* @param patternFlatTimeoutFunction The pattern timeout function which is called for each partial
	* pattern sequence which has timed out.
	* @param patternFlatSelectFunction The pattern select function which is called for each detected
	* pattern sequence.
	* @param <L> Type of the resulting timeout elements
	* @param <R> Type of the resulting elements
	* @return {@link DataStream} which contains the resulting elements with the resulting timeout
	* elements in a side output.
	*/
	public <L, R> SingleOutputStreamOperator<R> flatSelect(
	final OutputTag<L> timeoutOutputTag,
	final PatternFlatTimeoutFunction<T, L> patternFlatTimeoutFunction,
	final PatternFlatSelectFunction<T, R> patternFlatSelectFunction) {

	TypeInformation<R> rightTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
	patternFlatSelectFunction,
	PatternFlatSelectFunction.class,
	0,
	1,
	new int[]{0, 1, 0},
	new int[]{1, 0},
	inputStream.getType(),
	null,
	false);

	return flatSelect(timeoutOutputTag, patternFlatTimeoutFunction, rightTypeInfo, patternFlatSelectFunction);
	}

	/**
	* Applies a flat select function to the detected pattern sequence. For each pattern sequence the
	* provided {@link PatternFlatSelectFunction} is called. The pattern select function can produce
	* exactly one resulting element.
	*
	* <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
	* partial pattern sequence the provided {@link PatternFlatTimeoutFunction} is called. The pattern
	* timeout function can produce exactly one resulting element.
	*
	* <p>You can get the stream of timed-out data resulting from the
	* {@link SingleOutputStreamOperator#getSideOutput(OutputTag)} on the
	* {@link SingleOutputStreamOperator} resulting from the select operation
	* with the same {@link OutputTag}.
	*
	* @param timeoutOutputTag {@link OutputTag} that identifies side output with timed out patterns
	* @param patternFlatTimeoutFunction The pattern timeout function which is called for each partial
	* pattern sequence which has timed out.
	* @param patternFlatSelectFunction The pattern select function which is called for each detected
	* pattern sequence.
	* @param outTypeInfo Explicit specification of output type.
	* @param <L> Type of the resulting timeout elements
	* @param <R> Type of the resulting elements
	* @return {@link DataStream} which contains the resulting elements with the resulting timeout
	* elements in a side output.
	*/
	public <L, R> SingleOutputStreamOperator<R> flatSelect(
	final OutputTag<L> timeoutOutputTag,
	final PatternFlatTimeoutFunction<T, L> patternFlatTimeoutFunction,
	final TypeInformation<R> outTypeInfo,
	final PatternFlatSelectFunction<T, R> patternFlatSelectFunction) {

	return CEPOperatorUtils.createTimeoutPatternStream(
	inputStream,
	pattern,
	comparator,
	clean(patternFlatSelectFunction),
	outTypeInfo,
	timeoutOutputTag,
	clean(patternFlatTimeoutFunction),
	lateDataOutputTag);
	}

	/**
	* Applies a flat select function to the detected pattern sequence. For each pattern sequence
	* the provided {@link PatternFlatSelectFunction} is called. The pattern flat select function
	* can produce an arbitrary number of resulting elements.
	*
	* <p>Applies a timeout function to a partial pattern sequence which has timed out. For each
	* partial pattern sequence the provided {@link PatternFlatTimeoutFunction} is called. The
	* pattern timeout function can produce an arbitrary number of resulting elements.
	*
	* @param patternFlatTimeoutFunction The pattern flat timeout function which is called for each
	* partial pattern sequence which has timed out.
	* @param patternFlatSelectFunction The pattern flat select function which is called for each
	* detected pattern sequence.
	* @param <L> Type of the resulting timeout events
	* @param <R> Type of the resulting events
	*
	* @deprecated Use {@link PatternStream#flatSelect(OutputTag, PatternFlatTimeoutFunction, PatternFlatSelectFunction)}
	* that returns timed out events as a side-output
	*
	* @return {@link DataStream} which contains the resulting events from the pattern flat select
	* function or the resulting timeout events from the pattern flat timeout function wrapped in an
	* {@link Either} type.
	*/
	@Deprecated
	public <L, R> SingleOutputStreamOperator<Either<L, R>> flatSelect(
	final PatternFlatTimeoutFunction<T, L> patternFlatTimeoutFunction,
	final PatternFlatSelectFunction<T, R> patternFlatSelectFunction) {

	TypeInformation<L> leftTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
	patternFlatTimeoutFunction,
	PatternFlatTimeoutFunction.class,
	0,
	1,
	new int[]{0, 1, 0},
	new int[]{2, 0},
	inputStream.getType(),
	null,
	false);

	TypeInformation<R> rightTypeInfo = TypeExtractor.getUnaryOperatorReturnType(
	patternFlatSelectFunction,
	PatternFlatSelectFunction.class,
	0,
	1,
	new int[]{0, 1, 0},
	new int[]{1, 0},
	inputStream.getType(),
	null,
	false);

	final OutputTag<L> outputTag = new OutputTag<L>(UUID.randomUUID().toString(), leftTypeInfo);

	final SingleOutputStreamOperator<R> mainStream = CEPOperatorUtils.createTimeoutPatternStream(
	inputStream,
	pattern,
	comparator,
	clean(patternFlatSelectFunction),
	rightTypeInfo,
	outputTag,
	clean(patternFlatTimeoutFunction),
	lateDataOutputTag);

	final DataStream<L> timedOutStream = mainStream.getSideOutput(outputTag);

	TypeInformation<Either<L, R>> outTypeInfo = new EitherTypeInfo<>(leftTypeInfo, rightTypeInfo);

	return mainStream.connect(timedOutStream).map(new CoMapTimeout<>()).returns(outTypeInfo);
	}

	public PatternStream<T> sideOutputLateData(OutputTag<T> outputTag) {
	this.lateDataOutputTag = clean(outputTag);
	return this;
	}

	/**
	* Used for joining results from timeout side-output for API backward compatibility.
	*/
	@Internal
	public static class CoMapTimeout<R, L> implements CoMapFunction<R, L, Either<L, R>> {

	private static final long serialVersionUID = 2059391566945212552L;

	@Override
	public Either<L, R> map1(R value) throws Exception {
	return Either.Right(value);
	}

	@Override
	public Either<L, R> map2(L value) throws Exception {
	return Either.Left(value);
	}
	}
	}