>asRecord());
+ }
+ return priorityQueue;
+ }
+}
diff --git a/src/main/java/org/apache/flink/contrib/siddhi/package-info.java b/src/main/java/org/apache/flink/contrib/siddhi/package-info.java
new file mode 100644
index 0000000..91b13d9
--- /dev/null
+++ b/src/main/java/org/apache/flink/contrib/siddhi/package-info.java
@@ -0,0 +1,78 @@
+/*
+ * 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.
+ */
+
+/**
+ * Features
+ *
+ * -
+ * Integrate Siddhi CEP as an stream operator (i.e. `TupleStreamSiddhiOperator`), supporting rich CEP features like
+ *
+ * - Filter
+ * - Join
+ * - Aggregation
+ * - Group by
+ * - Having
+ * - Window
+ * - Conditions and Expressions
+ * - Pattern processing
+ * - Sequence processing
+ * - Event Tables
+ * - ...
+ *
+ *
+ * -
+ * Provide easy-to-use Siddhi CEP API to integrate Flink DataStream API (See `SiddhiCEP` and `SiddhiStream`)
+ *
+ * - Register Flink DataStream associating native type information with Siddhi Stream Schema, supporting POJO,Tuple, Primitive Type, etc.
+ * - Connect with single or multiple Flink DataStreams with Siddhi CEP Execution Plan
+ * - Return output stream as DataStream with type intelligently inferred from Siddhi Stream Schema
+ *
+ *
+ * -
+ * Integrate siddhi runtime state management with Flink state (See `AbstractSiddhiOperator`)
+ *
+ * -
+ * Support siddhi plugin management to extend CEP functions. (See `SiddhiCEP#registerExtension`)
+ *
+ *
+ *
+ * Example
+ *
+ * StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ * SiddhiCEP cep = SiddhiCEP.getSiddhiEnvironment(env);
+ *
+ * cep.registerExtension("custom:plus",CustomPlusFunctionExtension.class);
+ *
+ * cep.registerStream("inputStream1", input1, "id", "name", "price","timestamp");
+ * cep.registerStream("inputStream2", input2, "id", "name", "price","timestamp");
+ *
+ * DataStream<Tuple4<Integer,String,Integer,String>> output = cep
+ * .from("inputStream1").union("inputStream2")
+ * .sql(
+ * "from every s1 = inputStream1[id == 2] "
+ * + " -> s2 = inputStream2[id == 3] "
+ * + "select s1.id as id_1, s1.name as name_1, s2.id as id_2, s2.name as name_2 "
+ * + "insert into outputStream"
+ * )
+ * .returns("outputStream");
+ *
+ * env.execute();
+ *
+ *
+ * @see /~https://github.com/wso2/siddhi
+ */
+package org.apache.flink.contrib.siddhi;
diff --git a/src/main/java/org/apache/flink/contrib/siddhi/schema/SiddhiStreamSchema.java b/src/main/java/org/apache/flink/contrib/siddhi/schema/SiddhiStreamSchema.java
new file mode 100644
index 0000000..c81d2bc
--- /dev/null
+++ b/src/main/java/org/apache/flink/contrib/siddhi/schema/SiddhiStreamSchema.java
@@ -0,0 +1,67 @@
+/*
+ * 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.contrib.siddhi.schema;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.contrib.siddhi.utils.SiddhiTypeFactory;
+import org.apache.flink.util.Preconditions;
+import org.wso2.siddhi.query.api.definition.Attribute;
+import org.wso2.siddhi.query.api.definition.StreamDefinition;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class SiddhiStreamSchema extends StreamSchema {
+ private static final String DEFINE_STREAM_TEMPLATE = "define stream %s (%s);";
+
+ public SiddhiStreamSchema(TypeInformation typeInfo, String... fieldNames) {
+ super(typeInfo, fieldNames);
+ }
+
+ public SiddhiStreamSchema(TypeInformation typeInfo, int[] fieldIndexes, String[] fieldNames) {
+ super(typeInfo, fieldIndexes, fieldNames);
+ }
+
+ public StreamDefinition getStreamDefinition(String streamId) {
+ StreamDefinition streamDefinition = StreamDefinition.id(streamId);
+ for (int i = 0; i < getFieldNames().length; i++) {
+ streamDefinition.attribute(getFieldNames()[i], SiddhiTypeFactory.getAttributeType(getFieldTypes()[i]));
+ }
+ return streamDefinition;
+ }
+
+ public String getStreamDefinitionExpression(StreamDefinition streamDefinition) {
+ List columns = new ArrayList<>();
+ Preconditions.checkNotNull(streamDefinition, "StreamDefinition is null");
+ for (Attribute attribute : streamDefinition.getAttributeList()) {
+ columns.add(String.format("%s %s", attribute.getName(), attribute.getType().toString().toLowerCase()));
+ }
+ return String.format(DEFINE_STREAM_TEMPLATE, streamDefinition.getId(), StringUtils.join(columns, ","));
+ }
+
+ public String getStreamDefinitionExpression(String streamId) {
+ StreamDefinition streamDefinition = getStreamDefinition(streamId);
+ List columns = new ArrayList<>();
+ Preconditions.checkNotNull(streamDefinition, "StreamDefinition is null");
+ for (Attribute attribute : streamDefinition.getAttributeList()) {
+ columns.add(String.format("%s %s", attribute.getName(), attribute.getType().toString().toLowerCase()));
+ }
+ return String.format(DEFINE_STREAM_TEMPLATE, streamDefinition.getId(), StringUtils.join(columns, ","));
+ }
+}
diff --git a/src/main/java/org/apache/flink/contrib/siddhi/schema/StreamSchema.java b/src/main/java/org/apache/flink/contrib/siddhi/schema/StreamSchema.java
new file mode 100644
index 0000000..145a03b
--- /dev/null
+++ b/src/main/java/org/apache/flink/contrib/siddhi/schema/StreamSchema.java
@@ -0,0 +1,168 @@
+/*
+ * 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.contrib.siddhi.schema;
+
+import org.apache.flink.api.common.typeinfo.AtomicType;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.CompositeType;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.TupleTypeInfo;
+import org.apache.flink.api.scala.typeutils.CaseClassTypeInfo;
+import org.apache.flink.util.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Serializable;
+
+public class StreamSchema implements Serializable {
+ private static final Logger LOGGER = LoggerFactory.getLogger(StreamSchema.class);
+ private final TypeInformation typeInfo;
+ private final int[] fieldIndexes;
+ private final String[] fieldNames;
+ private TypeInformation[] fieldTypes;
+ private final StreamSerializer streamSerializer;
+ private TypeSerializer typeSerializer;
+
+ public StreamSchema(TypeInformation typeInfo, String... fieldNames) {
+ Preconditions.checkNotNull(fieldNames, "Field name is required");
+ this.typeInfo = typeInfo;
+ this.fieldNames = fieldNames;
+ this.fieldIndexes = getFieldIndexes(typeInfo, fieldNames);
+ this.fieldTypes = getFieldTypes(typeInfo, fieldIndexes, fieldNames);
+ this.streamSerializer = new StreamSerializer<>(this);
+ }
+
+ public StreamSchema(TypeInformation typeInfo, int[] fieldIndexes, String[] fieldNames) {
+ this.typeInfo = typeInfo;
+ this.fieldIndexes = fieldIndexes;
+ this.fieldNames = fieldNames;
+ this.fieldTypes = getFieldTypes(typeInfo, fieldIndexes, fieldNames);
+ this.streamSerializer = new StreamSerializer<>(this);
+ }
+
+ public boolean isAtomicType() {
+ return typeInfo instanceof AtomicType;
+ }
+
+ public boolean isTupleType() {
+ return typeInfo instanceof TupleTypeInfo;
+ }
+
+ public boolean isPojoType() {
+ return typeInfo instanceof PojoTypeInfo;
+ }
+
+ public boolean isCaseClassType() {
+ return typeInfo instanceof CaseClassTypeInfo;
+ }
+
+ public boolean isCompositeType() {
+ return typeInfo instanceof CompositeType;
+ }
+
+ private int[] getFieldIndexes(TypeInformation typeInfo, String... fieldNames) {
+ int[] result;
+ if (isAtomicType()) {
+ result = new int[]{0};
+ } else if (isTupleType()) {
+ result = new int[fieldNames.length];
+ for (int i = 0; i < fieldNames.length; i++) {
+ result[i] = i;
+ }
+ } else if (isPojoType()) {
+ result = new int[fieldNames.length];
+ for (int i = 0; i < fieldNames.length; i++) {
+ int index = ((PojoTypeInfo) typeInfo).getFieldIndex(fieldNames[i]);
+ if (index < 0) {
+ throw new IllegalArgumentException(fieldNames[i] + " is not a field of type " + typeInfo);
+ }
+ result[i] = index;
+ }
+ } else if (isCaseClassType()) {
+ result = new int[fieldNames.length];
+ for (int i = 0; i < fieldNames.length; i++) {
+ int index = ((CaseClassTypeInfo) typeInfo).getFieldIndex(fieldNames[i]);
+ if (index < 0) {
+ throw new IllegalArgumentException(fieldNames[i] + " is not a field of type " + typeInfo);
+ }
+ result[i] = index;
+ }
+ } else {
+ throw new IllegalArgumentException("Failed to get field index from " + typeInfo);
+ }
+ return result;
+ }
+
+
+ private TypeInformation[] getFieldTypes(TypeInformation typeInfo, int[] fieldIndexes, String[] fieldNames) {
+ TypeInformation[] fieldTypes;
+ if (isCompositeType()) {
+ CompositeType cType = (CompositeType) typeInfo;
+ if (fieldNames.length != cType.getArity()) {
+// throw new IllegalArgumentException("Arity of type (" + cType.getFieldNames().length+ ") " +
+// "not equal to number of field names " + fieldNames.length + ".");
+ LOGGER.warn("Arity of type (" + cType.getFieldNames().length + ") " +
+ "not equal to number of field names " + fieldNames.length + ".");
+ }
+ fieldTypes = new TypeInformation[fieldIndexes.length];
+ for (int i = 0; i < fieldIndexes.length; i++) {
+ fieldTypes[i] = cType.getTypeAt(fieldIndexes[i]);
+ }
+ } else if (isAtomicType()) {
+ if (fieldIndexes.length != 1 || fieldIndexes[0] != 0) {
+ throw new IllegalArgumentException(
+ "Non-composite input type may have only a single field and its index must be 0.");
+ }
+ fieldTypes = new TypeInformation[]{typeInfo};
+ } else {
+ throw new IllegalArgumentException(
+ "Illegal input type info"
+ );
+ }
+ return fieldTypes;
+ }
+
+ public TypeInformation getTypeInfo() {
+ return typeInfo;
+ }
+
+ public int[] getFieldIndexes() {
+ return fieldIndexes;
+ }
+
+ public String[] getFieldNames() {
+ return fieldNames;
+ }
+
+ public TypeInformation[] getFieldTypes() {
+ return fieldTypes;
+ }
+
+ public StreamSerializer getStreamSerializer() {
+ return streamSerializer;
+ }
+
+ public TypeSerializer getTypeSerializer() {
+ return typeSerializer;
+ }
+
+ public void setTypeSerializer(TypeSerializer typeSerializer) {
+ this.typeSerializer = typeSerializer;
+ }
+}
diff --git a/src/main/java/org/apache/flink/contrib/siddhi/schema/StreamSerializer.java b/src/main/java/org/apache/flink/contrib/siddhi/schema/StreamSerializer.java
new file mode 100644
index 0000000..9ac4b51
--- /dev/null
+++ b/src/main/java/org/apache/flink/contrib/siddhi/schema/StreamSerializer.java
@@ -0,0 +1,73 @@
+/*
+ * 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.contrib.siddhi.schema;
+
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.util.Preconditions;
+
+import java.io.Serializable;
+import java.lang.reflect.Field;
+
+public class StreamSerializer implements Serializable {
+ private final StreamSchema schema;
+
+ public StreamSerializer(StreamSchema schema) {
+ this.schema = schema;
+ }
+
+ public Object[] getRow(T input) {
+ Preconditions.checkArgument(input.getClass() == schema.getTypeInfo().getTypeClass()
+ , "Invalid input type: " + input + ", expected: " + schema.getTypeInfo());
+
+ Object[] data;
+ if (schema.isAtomicType()) {
+ data = new Object[]{input};
+ } else if (schema.isTupleType()) {
+ Tuple tuple = (Tuple) input;
+ data = new Object[schema.getFieldIndexes().length];
+ for (int i = 0; i < schema.getFieldIndexes().length; i++) {
+ data[i] = tuple.getField(schema.getFieldIndexes()[i]);
+ }
+ } else if (schema.isPojoType() || schema.isCaseClassType()) {
+ data = new Object[schema.getFieldIndexes().length];
+ for (int i = 0; i < schema.getFieldNames().length; i++) {
+ data[i] = getFieldValue(schema.getFieldNames()[i], input);
+ }
+ } else {
+ throw new IllegalArgumentException("Failed to get field values from " + schema.getTypeInfo());
+ }
+ return data;
+ }
+
+ private Object getFieldValue(String fieldName, T input) {
+ // TODO: Cache Field Accessor
+ Field field = TypeExtractor.getDeclaredField(schema.getTypeInfo().getTypeClass(), fieldName);
+ if (field == null) {
+ throw new IllegalArgumentException(fieldName + " is not found in " + schema.getTypeInfo());
+ }
+ if (!field.isAccessible()) {
+ field.setAccessible(true);
+ }
+ try {
+ return field.get(input);
+ } catch (IllegalAccessException e) {
+ throw new IllegalStateException(e.getMessage(), e);
+ }
+ }
+}
diff --git a/src/main/java/org/apache/flink/contrib/siddhi/utils/SiddhiStreamFactory.java b/src/main/java/org/apache/flink/contrib/siddhi/utils/SiddhiStreamFactory.java
new file mode 100644
index 0000000..d7431b8
--- /dev/null
+++ b/src/main/java/org/apache/flink/contrib/siddhi/utils/SiddhiStreamFactory.java
@@ -0,0 +1,33 @@
+/*
+ * 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.contrib.siddhi.utils;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.contrib.siddhi.operator.SiddhiOperatorContext;
+import org.apache.flink.contrib.siddhi.operator.TupleStreamSiddhiOperator;
+import org.apache.flink.streaming.api.datastream.DataStream;
+
+/**
+ * Convert SiddhiCEPExecutionPlan to SiddhiCEP Operator and build output DataStream
+ */
+public class SiddhiStreamFactory {
+ @SuppressWarnings("unchecked")
+ public static DataStream createDataStream(SiddhiOperatorContext context, DataStream> namedStream) {
+ return namedStream.transform(context.getName(), context.getOutputStreamType(), new TupleStreamSiddhiOperator(context));
+ }
+}
diff --git a/src/main/java/org/apache/flink/contrib/siddhi/utils/SiddhiTupleFactory.java b/src/main/java/org/apache/flink/contrib/siddhi/utils/SiddhiTupleFactory.java
new file mode 100644
index 0000000..28ff99a
--- /dev/null
+++ b/src/main/java/org/apache/flink/contrib/siddhi/utils/SiddhiTupleFactory.java
@@ -0,0 +1,123 @@
+/*
+ * 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.contrib.siddhi.utils;
+
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple0;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.tuple.Tuple5;
+import org.apache.flink.api.java.tuple.Tuple6;
+import org.apache.flink.api.java.tuple.Tuple7;
+import org.apache.flink.api.java.tuple.Tuple8;
+import org.apache.flink.api.java.tuple.Tuple9;
+import org.apache.flink.api.java.tuple.Tuple10;
+import org.apache.flink.api.java.tuple.Tuple11;
+import org.apache.flink.api.java.tuple.Tuple12;
+import org.apache.flink.api.java.tuple.Tuple13;
+import org.apache.flink.api.java.tuple.Tuple14;
+import org.apache.flink.api.java.tuple.Tuple15;
+import org.apache.flink.api.java.tuple.Tuple16;
+import org.apache.flink.api.java.tuple.Tuple17;
+import org.apache.flink.api.java.tuple.Tuple18;
+import org.apache.flink.api.java.tuple.Tuple19;
+import org.apache.flink.api.java.tuple.Tuple20;
+import org.apache.flink.api.java.tuple.Tuple21;
+import org.apache.flink.api.java.tuple.Tuple22;
+import org.apache.flink.api.java.tuple.Tuple23;
+import org.apache.flink.api.java.tuple.Tuple24;
+import org.apache.flink.api.java.tuple.Tuple25;
+import org.apache.flink.util.Preconditions;
+
+public class SiddhiTupleFactory {
+ /**
+ * Convert object array to type of Tuple{N} where N is between 0 to 25.
+ */
+ public static T newTuple(Object[] row) {
+ Preconditions.checkNotNull(row, "Tuple row is null");
+ switch (row.length) {
+ case 0:
+ return setTupleValue(new Tuple0(), row);
+ case 1:
+ return setTupleValue(new Tuple1(), row);
+ case 2:
+ return setTupleValue(new Tuple2(), row);
+ case 3:
+ return setTupleValue(new Tuple3(), row);
+ case 4:
+ return setTupleValue(new Tuple4(), row);
+ case 5:
+ return setTupleValue(new Tuple5(), row);
+ case 6:
+ return setTupleValue(new Tuple6(), row);
+ case 7:
+ return setTupleValue(new Tuple7(), row);
+ case 8:
+ return setTupleValue(new Tuple8(), row);
+ case 9:
+ return setTupleValue(new Tuple9(), row);
+ case 10:
+ return setTupleValue(new Tuple10(), row);
+ case 11:
+ return setTupleValue(new Tuple11(), row);
+ case 12:
+ return setTupleValue(new Tuple12(), row);
+ case 13:
+ return setTupleValue(new Tuple13(), row);
+ case 14:
+ return setTupleValue(new Tuple14(), row);
+ case 15:
+ return setTupleValue(new Tuple15(), row);
+ case 16:
+ return setTupleValue(new Tuple16(), row);
+ case 17:
+ return setTupleValue(new Tuple17(), row);
+ case 18:
+ return setTupleValue(new Tuple18(), row);
+ case 19:
+ return setTupleValue(new Tuple19(), row);
+ case 20:
+ return setTupleValue(new Tuple20(), row);
+ case 21:
+ return setTupleValue(new Tuple21(), row);
+ case 22:
+ return setTupleValue(new Tuple22(), row);
+ case 23:
+ return setTupleValue(new Tuple23(), row);
+ case 24:
+ return setTupleValue(new Tuple24(), row);
+ case 25:
+ return setTupleValue(new Tuple25(), row);
+ default:
+ throw new IllegalArgumentException("Too long row: " + row.length + ", unable to convert to Tuple");
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ public static T setTupleValue(Tuple tuple, Object[] row) {
+ if (row.length != tuple.getArity()) {
+ throw new IllegalArgumentException("Row length" + row.length + " is not equal with tuple's arity: " + tuple.getArity());
+ }
+ for (int i = 0; i < row.length; i++) {
+ tuple.setField(row[i], i);
+ }
+ return (T) tuple;
+ }
+}
diff --git a/src/main/java/org/apache/flink/contrib/siddhi/utils/SiddhiTypeFactory.java b/src/main/java/org/apache/flink/contrib/siddhi/utils/SiddhiTypeFactory.java
new file mode 100644
index 0000000..0a2d3f0
--- /dev/null
+++ b/src/main/java/org/apache/flink/contrib/siddhi/utils/SiddhiTypeFactory.java
@@ -0,0 +1,134 @@
+/*
+ * 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.contrib.siddhi.utils;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.wso2.siddhi.core.ExecutionPlanRuntime;
+import org.wso2.siddhi.core.SiddhiManager;
+import org.wso2.siddhi.query.api.definition.AbstractDefinition;
+import org.wso2.siddhi.query.api.definition.Attribute;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Siddhi Type Utils for conversion between Java Type, Siddhi Field Type, Stream Definition, and Flink Type Information.
+ */
+public class SiddhiTypeFactory {
+ private final static Map, Attribute.Type> JAVA_TO_SIDDHI_TYPE = new HashMap<>();
+ private final static Map> SIDDHI_TO_JAVA_TYPE = new HashMap<>();
+
+ static {
+ registerType(String.class, Attribute.Type.STRING);
+ registerType(Integer.class, Attribute.Type.INT);
+ registerType(int.class, Attribute.Type.INT);
+ registerType(Long.class, Attribute.Type.LONG);
+ registerType(long.class, Attribute.Type.LONG);
+ registerType(Float.class, Attribute.Type.FLOAT);
+ registerType(float.class, Attribute.Type.FLOAT);
+ registerType(Double.class, Attribute.Type.DOUBLE);
+ registerType(double.class, Attribute.Type.DOUBLE);
+ registerType(Boolean.class, Attribute.Type.BOOL);
+ registerType(boolean.class, Attribute.Type.BOOL);
+ }
+
+ public static void registerType(Class javaType, Attribute.Type siddhiType) {
+ if (JAVA_TO_SIDDHI_TYPE.containsKey(javaType)) {
+ throw new IllegalArgumentException("Java type: " + javaType + " or siddhi type: " + siddhiType + " were already registered");
+ }
+ JAVA_TO_SIDDHI_TYPE.put(javaType, siddhiType);
+ SIDDHI_TO_JAVA_TYPE.put(siddhiType, javaType);
+ }
+
+ public static AbstractDefinition getStreamDefinition(String executionPlan, String streamId) {
+ SiddhiManager siddhiManager = null;
+ ExecutionPlanRuntime runtime = null;
+ try {
+ siddhiManager = new SiddhiManager();
+ runtime = siddhiManager.createExecutionPlanRuntime(executionPlan);
+ Map definitionMap = runtime.getStreamDefinitionMap();
+ if (definitionMap.containsKey(streamId)) {
+ return definitionMap.get(streamId);
+ } else {
+ throw new IllegalArgumentException("Unknown stream id" + streamId);
+ }
+ } finally {
+ if (runtime != null) {
+ runtime.shutdown();
+ }
+ if (siddhiManager != null) {
+ siddhiManager.shutdown();
+ }
+ }
+ }
+
+ public static TypeInformation getTupleTypeInformation(AbstractDefinition definition) {
+ int tupleSize = definition.getAttributeList().size();
+ StringBuilder stringBuilder = new StringBuilder();
+ stringBuilder.append("Tuple").append(tupleSize);
+ stringBuilder.append("<");
+ List attributeTypes = new ArrayList<>();
+ for (Attribute attribute : definition.getAttributeList()) {
+ attributeTypes.add(getJavaType(attribute.getType()).getName());
+ }
+ stringBuilder.append(StringUtils.join(attributeTypes, ","));
+ stringBuilder.append(">");
+ try {
+ return TypeInfoParser.parse(stringBuilder.toString());
+ } catch (IllegalArgumentException ex) {
+ throw new IllegalArgumentException("Unable to parse " + stringBuilder.toString(), ex);
+ }
+ }
+
+ public static TypeInformation getTupleTypeInformation(String executionPlan, String streamId) {
+ return getTupleTypeInformation(getStreamDefinition(executionPlan, streamId));
+ }
+
+ @SuppressWarnings("unchecked")
+ private static final TypeInformation MAP_TYPE_INFORMATION = TypeExtractor.createTypeInfo(new HashMap().getClass());
+
+ public static TypeInformation> getMapTypeInformation() {
+ return (TypeInformation>) MAP_TYPE_INFORMATION;
+ }
+
+ public static Attribute.Type getAttributeType(TypeInformation fieldType) {
+ if (JAVA_TO_SIDDHI_TYPE.containsKey(fieldType.getTypeClass())) {
+ return JAVA_TO_SIDDHI_TYPE.get(fieldType.getTypeClass());
+ } else {
+ return Attribute.Type.OBJECT;
+ }
+ }
+
+ public static Class getJavaType(Attribute.Type attributeType) {
+ if (!SIDDHI_TO_JAVA_TYPE.containsKey(attributeType)) {
+ throw new IllegalArgumentException("Unable to get java type for siddhi attribute type: " + attributeType);
+ }
+ return SIDDHI_TO_JAVA_TYPE.get(attributeType);
+ }
+
+ public static TypeInformation> getStreamTupleTypeInformation(TypeInformation typeInformation) {
+ return TypeInfoParser.parse("Tuple2");
+ }
+}
diff --git a/src/test/java/org/apache/flink/contrib/siddhi/SiddhiCEPITCase.java b/src/test/java/org/apache/flink/contrib/siddhi/SiddhiCEPITCase.java
new file mode 100644
index 0000000..a4b3043
--- /dev/null
+++ b/src/test/java/org/apache/flink/contrib/siddhi/SiddhiCEPITCase.java
@@ -0,0 +1,375 @@
+/*
+ * 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.contrib.siddhi;
+
+import org.apache.flink.api.common.functions.InvalidTypesException;
+import org.apache.flink.api.common.functions.MapFunction;
+import org.apache.flink.api.java.tuple.Tuple1;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.tuple.Tuple5;
+import org.apache.flink.contrib.siddhi.exception.UndefinedStreamException;
+import org.apache.flink.contrib.siddhi.extension.CustomPlusFunctionExtension;
+import org.apache.flink.contrib.siddhi.source.Event;
+import org.apache.flink.contrib.siddhi.source.RandomEventSource;
+import org.apache.flink.contrib.siddhi.source.RandomTupleSource;
+import org.apache.flink.contrib.siddhi.source.RandomWordSource;
+import org.apache.flink.core.fs.FileSystem;
+import org.apache.flink.streaming.api.TimeCharacteristic;
+import org.apache.flink.streaming.api.datastream.DataStream;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
+import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.IOException;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ * flink-siddhi integration test cases
+ */
+public class SiddhiCEPITCase extends StreamingMultipleProgramsTestBase {
+
+ @Rule
+ public TemporaryFolder tempFolder = new TemporaryFolder();
+
+ @Test
+ public void testSimplePojoStreamAndReturnPojo() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input = env.fromElements(
+ Event.of(1, "start", 1.0),
+ Event.of(2, "middle", 2.0),
+ Event.of(3, "end", 3.0),
+ Event.of(4, "start", 4.0),
+ Event.of(5, "middle", 5.0),
+ Event.of(6, "end", 6.0)
+ );
+
+ DataStream output = SiddhiCEP
+ .define("inputStream", input, "id", "name", "price")
+ .sql("from inputStream insert into outputStream")
+ .returns("outputStream", Event.class);
+ String path = tempFolder.newFile().toURI().toString();
+ output.writeAsText(path);
+ env.execute();
+ }
+
+ @Test
+ public void testUnboundedPojoSourceAndReturnTuple() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input = env.addSource(new RandomEventSource(5));
+
+ DataStream> output = SiddhiCEP
+ .define("inputStream", input, "id", "name", "price", "timestamp")
+ .sql("from inputStream select timestamp, id, name, price insert into outputStream")
+ .returns("outputStream");
+
+ DataStream following = output.map(new MapFunction, Integer>() {
+ @Override
+ public Integer map(Tuple4 value) throws Exception {
+ return value.f1;
+ }
+ });
+ String resultPath = tempFolder.newFile().toURI().toString();
+ following.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(5, getLineCount(resultPath));
+ }
+
+ @Test
+ public void testUnboundedTupleSourceAndReturnTuple() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream> input = env.addSource(new RandomTupleSource(5).closeDelay(1500));
+
+ DataStream> output = SiddhiCEP
+ .define("inputStream", input, "id", "name", "price", "timestamp")
+ .sql("from inputStream select timestamp, id, name, price insert into outputStream")
+ .returns("outputStream");
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(5, getLineCount(resultPath));
+ }
+
+ @Test
+ public void testUnboundedPrimitiveTypeSourceAndReturnTuple() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input = env.addSource(new RandomWordSource(5).closeDelay(1500));
+
+ DataStream> output = SiddhiCEP
+ .define("wordStream", input, "words")
+ .sql("from wordStream select words insert into outputStream")
+ .returns("outputStream");
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(5, getLineCount(resultPath));
+ }
+
+ @Test(expected = InvalidTypesException.class)
+ public void testUnboundedPojoSourceButReturnInvalidTupleType() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input = env.addSource(new RandomEventSource(5).closeDelay(1500));
+
+ DataStream> output = SiddhiCEP
+ .define("inputStream", input, "id", "name", "price", "timestamp")
+ .sql("from inputStream select timestamp, id, name, price insert into outputStream")
+ .returns("outputStream");
+
+ DataStream following = output.map(new MapFunction, Long>() {
+ @Override
+ public Long map(Tuple5 value) throws Exception {
+ return value.f0;
+ }
+ });
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ following.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(5, getLineCount(resultPath));
+ env.execute();
+ }
+
+ @Test
+ public void testUnboundedPojoStreamAndReturnMap() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setParallelism(1);
+ env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
+ DataStream input = env.addSource(new RandomEventSource(5));
+
+ DataStream> output = SiddhiCEP
+ .define("inputStream", input, "id", "name", "price", "timestamp")
+ .sql("from inputStream select timestamp, id, name, price insert into outputStream")
+ .returnAsMap("outputStream");
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(5, getLineCount(resultPath));
+ }
+
+ @Test
+ public void testUnboundedPojoStreamAndReturnPojo() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input = env.addSource(new RandomEventSource(5));
+ input.assignTimestampsAndWatermarks(new AscendingTimestampExtractor() {
+ @Override
+ public long extractAscendingTimestamp(Event element) {
+ return element.getTimestamp();
+ }
+ });
+
+ DataStream output = SiddhiCEP
+ .define("inputStream", input, "id", "name", "price", "timestamp")
+ .sql("from inputStream select timestamp, id, name, price insert into outputStream")
+ .returns("outputStream", Event.class);
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(5, getLineCount(resultPath));
+ }
+
+
+ @Test
+ public void testMultipleUnboundedPojoStreamSimpleUnion() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input1 = env.addSource(new RandomEventSource(2), "input1");
+ DataStream input2 = env.addSource(new RandomEventSource(2), "input2");
+ DataStream input3 = env.addSource(new RandomEventSource(2), "input2");
+ DataStream output = SiddhiCEP
+ .define("inputStream1", input1, "id", "name", "price", "timestamp")
+ .union("inputStream2", input2, "id", "name", "price", "timestamp")
+ .union("inputStream3", input3, "id", "name", "price", "timestamp")
+ .sql(
+ "from inputStream1 select timestamp, id, name, price insert into outputStream;"
+ + "from inputStream2 select timestamp, id, name, price insert into outputStream;"
+ + "from inputStream3 select timestamp, id, name, price insert into outputStream;"
+ )
+ .returns("outputStream", Event.class);
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(6, getLineCount(resultPath));
+ }
+
+ /**
+ * @see https://docs.wso2.com/display/CEP300/Joins
+ */
+ @Test
+ public void testMultipleUnboundedPojoStreamUnionAndJoinWithWindow() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input1 = env.addSource(new RandomEventSource(5), "input1");
+ DataStream input2 = env.addSource(new RandomEventSource(5), "input2");
+
+ DataStream output = SiddhiCEP
+ .define("inputStream1", input1.keyBy("id"), "id", "name", "price", "timestamp")
+ .union("inputStream2", input2.keyBy("id"), "id", "name", "price", "timestamp")
+ .sql(
+ "from inputStream1#window.length(5) as s1 "
+ + "join inputStream2#window.time(500) as s2 "
+ + "on s1.id == s2.id "
+ + "select s1.timestamp as t, s1.name as n, s1.price as p1, s2.price as p2 "
+ + "insert into JoinStream;"
+ )
+ .returnAsMap("JoinStream");
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(5, getLineCount(resultPath));
+ }
+
+ /**
+ * @see https://docs.wso2.com/display/CEP300/Patterns
+ */
+ @Test
+ public void testUnboundedPojoStreamSimplePatternMatch() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input1 = env.addSource(new RandomEventSource(5).closeDelay(1500), "input1");
+ DataStream input2 = env.addSource(new RandomEventSource(5).closeDelay(1500), "input2");
+
+ DataStream> output = SiddhiCEP
+ .define("inputStream1", input1.keyBy("name"), "id", "name", "price", "timestamp")
+ .union("inputStream2", input2.keyBy("name"), "id", "name", "price", "timestamp")
+ .sql(
+ "from every s1 = inputStream1[id == 2] "
+ + " -> s2 = inputStream2[id == 3] "
+ + "select s1.id as id_1, s1.name as name_1, s2.id as id_2, s2.name as name_2 "
+ + "insert into outputStream"
+ )
+ .returnAsMap("outputStream");
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(1, getLineCount(resultPath));
+ compareResultsByLinesInMemory("{id_1=2, name_1=test_event, id_2=3, name_2=test_event}", resultPath);
+ }
+
+ /**
+ * @see https://docs.wso2.com/display/CEP300/Sequences
+ */
+ @Test
+ public void testUnboundedPojoStreamSimpleSequences() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input1 = env.addSource(new RandomEventSource(5).closeDelay(1500), "input1");
+ DataStream> output = SiddhiCEP
+ .define("inputStream1", input1.keyBy("name"), "id", "name", "price", "timestamp")
+ .union("inputStream2", input1.keyBy("name"), "id", "name", "price", "timestamp")
+ .sql(
+ "from every s1 = inputStream1[id == 2]+ , "
+ + "s2 = inputStream2[id == 3]? "
+ + "within 1000 second "
+ + "select s1[0].name as n1, s2.name as n2 "
+ + "insert into outputStream"
+ )
+ .returnAsMap("outputStream");
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(1, getLineCount(resultPath));
+ }
+
+ private static int getLineCount(String resPath) throws IOException {
+ List result = new LinkedList<>();
+ readAllResultLines(result, resPath);
+ return result.size();
+ }
+
+ @Test
+ public void testCustomizeSiddhiFunctionExtension() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input = env.addSource(new RandomEventSource(5));
+
+ SiddhiCEP cep = SiddhiCEP.getSiddhiEnvironment(env);
+ cep.registerExtension("custom:plus", CustomPlusFunctionExtension.class);
+
+ DataStream> output = cep
+ .from("inputStream", input, "id", "name", "price", "timestamp")
+ .sql("from inputStream select timestamp, id, name, custom:plus(price,price) as doubled_price insert into outputStream")
+ .returnAsMap("outputStream");
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(5, getLineCount(resultPath));
+ }
+
+ @Test
+ public void testRegisterStreamAndExtensionWithSiddhiCEPEnvironment() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input1 = env.addSource(new RandomEventSource(5), "input1");
+ DataStream input2 = env.addSource(new RandomEventSource(5), "input2");
+
+ SiddhiCEP cep = SiddhiCEP.getSiddhiEnvironment(env);
+ cep.registerExtension("custom:plus", CustomPlusFunctionExtension.class);
+
+ cep.registerStream("inputStream1", input1.keyBy("id"), "id", "name", "price", "timestamp");
+ cep.registerStream("inputStream2", input2.keyBy("id"), "id", "name", "price", "timestamp");
+
+ DataStream> output = cep
+ .from("inputStream1").union("inputStream2")
+ .sql(
+ "from inputStream1#window.length(5) as s1 "
+ + "join inputStream2#window.time(500) as s2 "
+ + "on s1.id == s2.id "
+ + "select s1.timestamp as t, s1.name as n, s1.price as p1, s2.price as p2 "
+ + "insert into JoinStream;"
+ )
+ .returns("JoinStream");
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ assertEquals(5, getLineCount(resultPath));
+ }
+
+ @Test(expected = UndefinedStreamException.class)
+ public void testTriggerUndefinedStreamException() throws Exception {
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ DataStream input1 = env.addSource(new RandomEventSource(5), "input1");
+
+ SiddhiCEP cep = SiddhiCEP.getSiddhiEnvironment(env);
+ cep.registerStream("inputStream1", input1.keyBy("id"), "id", "name", "price", "timestamp");
+
+ DataStream> output = cep
+ .from("inputStream1").union("inputStream2")
+ .sql(
+ "from inputStream1#window.length(5) as s1 "
+ + "join inputStream2#window.time(500) as s2 "
+ + "on s1.id == s2.id "
+ + "select s1.timestamp as t, s1.name as n, s1.price as p1, s2.price as p2 "
+ + "insert into JoinStream;"
+ )
+ .returnAsMap("JoinStream");
+
+ String resultPath = tempFolder.newFile().toURI().toString();
+ output.writeAsText(resultPath, FileSystem.WriteMode.OVERWRITE);
+ env.execute();
+ }
+}
diff --git a/src/test/java/org/apache/flink/contrib/siddhi/extension/CustomPlusFunctionExtension.java b/src/test/java/org/apache/flink/contrib/siddhi/extension/CustomPlusFunctionExtension.java
new file mode 100644
index 0000000..6e2746e
--- /dev/null
+++ b/src/test/java/org/apache/flink/contrib/siddhi/extension/CustomPlusFunctionExtension.java
@@ -0,0 +1,142 @@
+/*
+ * 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.contrib.siddhi.extension;
+
+import org.wso2.siddhi.core.config.ExecutionPlanContext;
+import org.wso2.siddhi.core.exception.ExecutionPlanCreationException;
+import org.wso2.siddhi.core.executor.ExpressionExecutor;
+import org.wso2.siddhi.core.executor.function.FunctionExecutor;
+import org.wso2.siddhi.query.api.definition.Attribute;
+
+public class CustomPlusFunctionExtension extends FunctionExecutor {
+
+ private Attribute.Type returnType;
+
+ /**
+ * The initialization method for FunctionExecutor, this method will be called before the other methods
+ *
+ * @param attributeExpressionExecutors are the executors of each function parameters
+ * @param executionPlanContext the context of the execution plan
+ */
+ @Override
+ protected void init(ExpressionExecutor[] attributeExpressionExecutors, ExecutionPlanContext executionPlanContext) {
+ for (ExpressionExecutor expressionExecutor : attributeExpressionExecutors) {
+ Attribute.Type attributeType = expressionExecutor.getReturnType();
+ if (attributeType == Attribute.Type.DOUBLE) {
+ returnType = attributeType;
+
+ } else if ((attributeType == Attribute.Type.STRING) || (attributeType == Attribute.Type.BOOL)) {
+ throw new ExecutionPlanCreationException("Plus cannot have parameters with types String or Bool");
+ } else {
+ returnType = Attribute.Type.LONG;
+ }
+ }
+ }
+
+ /**
+ * The main execution method which will be called upon event arrival
+ * when there are more then one function parameter
+ *
+ * @param data the runtime values of function parameters
+ * @return the function result
+ */
+ @Override
+ protected Object execute(Object[] data) {
+ if (returnType == Attribute.Type.DOUBLE) {
+ double total = 0;
+ for (Object aObj : data) {
+ total += Double.parseDouble(String.valueOf(aObj));
+ }
+
+ return total;
+ } else {
+ long total = 0;
+ for (Object aObj : data) {
+ total += Long.parseLong(String.valueOf(aObj));
+ }
+ return total;
+ }
+ }
+
+ /**
+ * The main execution method which will be called upon event arrival
+ * when there are zero or one function parameter
+ *
+ * @param data null if the function parameter count is zero or
+ * runtime data value of the function parameter
+ * @return the function result
+ */
+ @Override
+ protected Object execute(Object data) {
+ if (returnType == Attribute.Type.DOUBLE) {
+ return Double.parseDouble(String.valueOf(data));
+ } else {
+ return Long.parseLong(String.valueOf(data));
+ }
+ }
+
+ /**
+ * This will be called only once and this can be used to acquire
+ * required resources for the processing element.
+ * This will be called after initializing the system and before
+ * starting to process the events.
+ */
+ @Override
+ public void start() {
+
+ }
+
+ /**
+ * This will be called only once and this can be used to release
+ * the acquired resources for processing.
+ * This will be called before shutting down the system.
+ */
+ @Override
+ public void stop() {
+
+ }
+
+ @Override
+ public Attribute.Type getReturnType() {
+ return returnType;
+ }
+
+ /**
+ * Used to collect the serializable state of the processing element, that need to be
+ * persisted for the reconstructing the element to the same state on a different point of time
+ *
+ * @return stateful objects of the processing element as an array
+ */
+ @Override
+ public Object[] currentState() {
+ return new Object[0];
+ }
+
+ /**
+ * Used to restore serialized state of the processing element, for reconstructing
+ * the element to the same state as if was on a previous point of time.
+ *
+ * @param state the stateful objects of the element as an array on
+ * the same order provided by currentState().
+ */
+ @Override
+ public void restoreState(Object[] state) {
+
+ }
+
+}
diff --git a/src/test/java/org/apache/flink/contrib/siddhi/operator/SiddhiSyntaxTest.java b/src/test/java/org/apache/flink/contrib/siddhi/operator/SiddhiSyntaxTest.java
new file mode 100644
index 0000000..d70107d
--- /dev/null
+++ b/src/test/java/org/apache/flink/contrib/siddhi/operator/SiddhiSyntaxTest.java
@@ -0,0 +1,83 @@
+/*
+ * 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.contrib.siddhi.operator;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.wso2.siddhi.core.ExecutionPlanRuntime;
+import org.wso2.siddhi.core.SiddhiManager;
+import org.wso2.siddhi.core.event.Event;
+import org.wso2.siddhi.core.stream.input.InputHandler;
+import org.wso2.siddhi.core.stream.output.StreamCallback;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class SiddhiSyntaxTest {
+
+ private SiddhiManager siddhiManager;
+
+ @Before
+ public void setUp() {
+ siddhiManager = new SiddhiManager();
+ }
+
+ @After
+ public void after() {
+ siddhiManager = new SiddhiManager();
+ }
+
+ @Test
+ public void testSimplePlan() throws InterruptedException {
+ ExecutionPlanRuntime runtime = siddhiManager.createExecutionPlanRuntime(
+ "define stream inStream (name string, value double);"
+ + "from inStream insert into outStream");
+ runtime.start();
+
+ final List received = new ArrayList<>(3);
+ InputHandler inputHandler = runtime.getInputHandler("inStream");
+ Assert.assertNotNull(inputHandler);
+
+ try {
+ runtime.getInputHandler("unknownStream");
+ Assert.fail("Should throw exception for getting input handler for unknown streamId.");
+ } catch (Exception ex) {
+ // Expected exception for getting input handler for illegal streamId.
+ }
+
+ runtime.addCallback("outStream", new StreamCallback() {
+ @Override
+ public void receive(Event[] events) {
+ for (Event event : events) {
+ received.add(event.getData());
+ }
+ }
+ });
+
+ inputHandler.send(new Object[]{"a", 1.1});
+ inputHandler.send(new Object[]{"b", 1.2});
+ inputHandler.send(new Object[]{"c", 1.3});
+ Thread.sleep(100);
+ Assert.assertEquals(3, received.size());
+ Assert.assertArrayEquals(received.get(0), new Object[]{"a", 1.1});
+ Assert.assertArrayEquals(received.get(1), new Object[]{"b", 1.2});
+ Assert.assertArrayEquals(received.get(2), new Object[]{"c", 1.3});
+ }
+}
diff --git a/src/test/java/org/apache/flink/contrib/siddhi/schema/SiddhiExecutionPlanSchemaTest.java b/src/test/java/org/apache/flink/contrib/siddhi/schema/SiddhiExecutionPlanSchemaTest.java
new file mode 100644
index 0000000..8af3cc8
--- /dev/null
+++ b/src/test/java/org/apache/flink/contrib/siddhi/schema/SiddhiExecutionPlanSchemaTest.java
@@ -0,0 +1,49 @@
+/*
+ * 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.contrib.siddhi.schema;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.contrib.siddhi.source.Event;
+import org.junit.Test;
+import org.wso2.siddhi.query.api.definition.Attribute;
+import org.wso2.siddhi.query.api.definition.StreamDefinition;
+
+import static org.junit.Assert.*;
+
+public class SiddhiExecutionPlanSchemaTest {
+ @Test
+ public void testStreamSchemaWithPojo() {
+ TypeInformation typeInfo = TypeExtractor.createTypeInfo(Event.class);
+ assertTrue("Type information should be PojoTypeInfo", typeInfo instanceof PojoTypeInfo);
+
+ SiddhiStreamSchema schema = new SiddhiStreamSchema<>(typeInfo, "id", "timestamp", "name", "price");
+ assertEquals(4, schema.getFieldIndexes().length);
+
+ StreamDefinition streamDefinition = schema.getStreamDefinition("test_stream");
+ assertArrayEquals(new String[]{"id", "timestamp", "name", "price"}, streamDefinition.getAttributeNameArray());
+
+ assertEquals(Attribute.Type.INT, streamDefinition.getAttributeType("id"));
+ assertEquals(Attribute.Type.LONG, streamDefinition.getAttributeType("timestamp"));
+ assertEquals(Attribute.Type.STRING, streamDefinition.getAttributeType("name"));
+ assertEquals(Attribute.Type.DOUBLE, streamDefinition.getAttributeType("price"));
+
+ assertEquals("define stream test_stream (id int,timestamp long,name string,price double);", schema.getStreamDefinitionExpression("test_stream"));
+ }
+}
diff --git a/src/test/java/org/apache/flink/contrib/siddhi/schema/StreamSchemaTest.java b/src/test/java/org/apache/flink/contrib/siddhi/schema/StreamSchemaTest.java
new file mode 100644
index 0000000..c69418a
--- /dev/null
+++ b/src/test/java/org/apache/flink/contrib/siddhi/schema/StreamSchemaTest.java
@@ -0,0 +1,94 @@
+/*
+ * 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.contrib.siddhi.schema;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.typeutils.PojoTypeInfo;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.api.java.typeutils.TypeInfoParser;
+import org.apache.flink.contrib.siddhi.source.Event;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class StreamSchemaTest {
+ @Test
+ public void testStreamSchemaWithPojo() {
+ TypeInformation typeInfo = TypeExtractor.createTypeInfo(Event.class);
+ assertTrue("Type information should be PojoTypeInfo", typeInfo instanceof PojoTypeInfo);
+ StreamSchema schema = new StreamSchema<>(typeInfo, "id", "timestamp", "name", "price");
+ assertEquals(4, schema.getFieldIndexes().length);
+ assertEquals(Event.class, schema.getTypeInfo().getTypeClass());
+ }
+
+ @Test
+ public void testStreamSchemaWithTuple() {
+ TypeInformation typeInfo = TypeInfoParser.parse("Tuple4");
+ StreamSchema schema = new StreamSchema<>(typeInfo, "id", "timestamp", "name", "price");
+ assertEquals(Tuple4.class, schema.getTypeInfo().getTypeClass());
+ assertEquals(4, schema.getFieldIndexes().length);
+ assertEquals(Tuple4.class, schema.getTypeInfo().getTypeClass());
+ }
+
+ @Test
+ public void testStreamSchemaWithPrimitive() {
+ TypeInformation typeInfo = TypeInfoParser.parse("String");
+ StreamSchema schema = new StreamSchema<>(typeInfo, "words");
+ assertEquals(String.class, schema.getTypeInfo().getTypeClass());
+ assertEquals(1, schema.getFieldIndexes().length);
+ assertEquals(String.class, schema.getTypeInfo().getTypeClass());
+ }
+
+ @Test(expected = IllegalArgumentException.class)
+ public void testStreamSchemaWithPojoAndUnknownField() {
+ TypeInformation typeInfo = TypeExtractor.createTypeInfo(Event.class);
+ new StreamSchema<>(typeInfo, "id", "timestamp", "name", "price", "unknown");
+ }
+
+ @Test
+ public void testStreamTupleSerializerWithPojo() {
+ TypeInformation typeInfo = TypeExtractor.createTypeInfo(Event.class);
+ assertTrue("Type information should be PojoTypeInfo", typeInfo instanceof PojoTypeInfo);
+ StreamSchema schema = new StreamSchema<>(typeInfo, "id", "timestamp", "name", "price");
+ assertEquals(Event.class, schema.getTypeInfo().getTypeClass());
+
+ TypeInformation> tuple2TypeInformation = TypeInfoParser.parse("Tuple2");
+ assertEquals("Java Tuple2>", tuple2TypeInformation.toString());
+ }
+
+ @Test
+ public void testStreamTupleSerializerWithTuple() {
+ TypeInformation typeInfo = TypeInfoParser.parse("Tuple4");
+ StreamSchema schema = new StreamSchema<>(typeInfo, "id", "timestamp", "name", "price");
+ assertEquals(Tuple4.class, schema.getTypeInfo().getTypeClass());
+ TypeInformation> tuple2TypeInformation = TypeInfoParser.parse("Tuple2");
+ assertEquals("Java Tuple2>", tuple2TypeInformation.toString());
+ }
+
+ @Test
+ public void testStreamTupleSerializerWithPrimitive() {
+ TypeInformation typeInfo = TypeInfoParser.parse("String");
+ StreamSchema schema = new StreamSchema<>(typeInfo, "words");
+ assertEquals(String.class, schema.getTypeInfo().getTypeClass());
+ TypeInformation> tuple2TypeInformation = TypeInfoParser.parse("Tuple2");
+ assertEquals("Java Tuple2", tuple2TypeInformation.toString());
+ }
+}
diff --git a/src/test/java/org/apache/flink/contrib/siddhi/schema/StreamSerializerTest.java b/src/test/java/org/apache/flink/contrib/siddhi/schema/StreamSerializerTest.java
new file mode 100644
index 0000000..0005306
--- /dev/null
+++ b/src/test/java/org/apache/flink/contrib/siddhi/schema/StreamSerializerTest.java
@@ -0,0 +1,40 @@
+/*
+ * 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.contrib.siddhi.schema;
+
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+import org.apache.flink.contrib.siddhi.source.Event;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class StreamSerializerTest {
+ private final static long CURRENT = System.currentTimeMillis();
+
+ @Test
+ public void testSimplePojoRead() {
+ Event event = new Event();
+ event.setId(1);
+ event.setName("test");
+ event.setPrice(56.7);
+ event.setTimestamp(CURRENT);
+
+ StreamSchema schema = new StreamSchema<>(TypeExtractor.createTypeInfo(Event.class), "id", "name", "price", "timestamp");
+ StreamSerializer reader = new StreamSerializer<>(schema);
+ Assert.assertArrayEquals(new Object[]{1, "test", 56.7, CURRENT}, reader.getRow(event));
+ }
+}
diff --git a/src/test/java/org/apache/flink/contrib/siddhi/source/Event.java b/src/test/java/org/apache/flink/contrib/siddhi/source/Event.java
new file mode 100644
index 0000000..eda8eed
--- /dev/null
+++ b/src/test/java/org/apache/flink/contrib/siddhi/source/Event.java
@@ -0,0 +1,110 @@
+/*
+ * 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.contrib.siddhi.source;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.typeutils.TypeExtractor;
+
+import java.util.Objects;
+
+public class Event {
+ private long timestamp;
+ private String name;
+ private double price;
+ private int id;
+
+ public double getPrice() {
+ return price;
+ }
+
+ public int getId() {
+ return id;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ @Override
+ public String toString() {
+ return "Event(" + id + ", " + name + ", " + price + ", " + timestamp + ")";
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof Event) {
+ Event other = (Event) obj;
+
+ return name.equals(other.name) && price == other.price && id == other.id && timestamp == other.timestamp;
+ } else {
+ return false;
+ }
+ }
+
+ public static Event of(int id, String name, double price) {
+ Event event = new Event();
+ event.setId(id);
+ event.setName(name);
+ event.setPrice(price);
+ event.setTimestamp(System.currentTimeMillis());
+ return event;
+ }
+
+ public static Event of(int id, String name, double price, long timestamp) {
+ Event event = new Event();
+ event.setId(id);
+ event.setName(name);
+ event.setPrice(price);
+ event.setTimestamp(timestamp);
+ return event;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, price, id);
+ }
+
+ public static TypeSerializer createTypeSerializer() {
+ TypeInformation typeInformation = (TypeInformation) TypeExtractor.createTypeInfo(Event.class);
+
+ return typeInformation.createSerializer(new ExecutionConfig());
+ }
+
+ public long getTimestamp() {
+ return timestamp;
+ }
+
+ public void setTimestamp(long timestamp) {
+ this.timestamp = timestamp;
+ }
+
+ public void setPrice(double price) {
+ this.price = price;
+ }
+
+ public void setId(int id) {
+ this.id = id;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+}
diff --git a/src/test/java/org/apache/flink/contrib/siddhi/source/RandomEventSource.java b/src/test/java/org/apache/flink/contrib/siddhi/source/RandomEventSource.java
new file mode 100644
index 0000000..5848930
--- /dev/null
+++ b/src/test/java/org/apache/flink/contrib/siddhi/source/RandomEventSource.java
@@ -0,0 +1,72 @@
+/*
+ * 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.contrib.siddhi.source;
+
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+
+import java.util.Random;
+
+public class RandomEventSource implements SourceFunction {
+ private final int count;
+ private final Random random;
+ private final long initialTimestamp;
+
+ private volatile boolean isRunning = true;
+ private volatile int number = 0;
+ private volatile long closeDelayTimestamp = 1000;
+
+ public RandomEventSource(int count, long initialTimestamp) {
+ this.count = count;
+ this.random = new Random();
+ this.initialTimestamp = initialTimestamp;
+ }
+
+ public RandomEventSource() {
+ this(Integer.MAX_VALUE, System.currentTimeMillis());
+ }
+
+ public RandomEventSource(int count) {
+ this(count, System.currentTimeMillis());
+ }
+
+ public RandomEventSource closeDelay(long delayTimestamp) {
+ this.closeDelayTimestamp = delayTimestamp;
+ return this;
+ }
+
+ @Override
+ public void run(SourceContext ctx) throws Exception {
+ while (isRunning) {
+ ctx.collect(Event.of(number, "test_event", random.nextDouble(), initialTimestamp + 1000 * number));
+ number++;
+ if (number >= this.count) {
+ cancel();
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ this.isRunning = false;
+ try {
+ Thread.sleep(closeDelayTimestamp);
+ } catch (InterruptedException e) {
+ // ignored
+ }
+ }
+}
diff --git a/src/test/java/org/apache/flink/contrib/siddhi/source/RandomTupleSource.java b/src/test/java/org/apache/flink/contrib/siddhi/source/RandomTupleSource.java
new file mode 100644
index 0000000..271b906
--- /dev/null
+++ b/src/test/java/org/apache/flink/contrib/siddhi/source/RandomTupleSource.java
@@ -0,0 +1,74 @@
+/*
+ * 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.contrib.siddhi.source;
+
+import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+
+import java.util.Random;
+
+public class RandomTupleSource implements SourceFunction> {
+ private final int count;
+ private final Random random;
+ private final long initialTimestamp;
+
+ private volatile boolean isRunning = true;
+ private volatile int number = 0;
+ private long closeDelayTimestamp;
+
+ public RandomTupleSource(int count, long initialTimestamp) {
+ this.count = count;
+ this.random = new Random();
+ this.initialTimestamp = initialTimestamp;
+ }
+
+ public RandomTupleSource() {
+ this(Integer.MAX_VALUE, System.currentTimeMillis());
+ }
+
+ public RandomTupleSource(int count) {
+ this(count, System.currentTimeMillis());
+ }
+
+
+ public RandomTupleSource closeDelay(long delayTimestamp) {
+ this.closeDelayTimestamp = delayTimestamp;
+ return this;
+ }
+
+ @Override
+ public void run(SourceContext> ctx) throws Exception {
+ while (isRunning) {
+ ctx.collect(Tuple4.of(number, "test_tuple", random.nextDouble(), initialTimestamp + 1000 * number));
+ number++;
+ if (number >= this.count) {
+ cancel();
+ }
+ }
+ }
+
+ @Override
+ public void cancel() {
+ this.isRunning = false;
+ try {
+ Thread.sleep(this.closeDelayTimestamp);
+ } catch (InterruptedException e) {
+ // ignored
+ }
+ }
+}
diff --git a/src/test/java/org/apache/flink/contrib/siddhi/source/RandomWordSource.java b/src/test/java/org/apache/flink/contrib/siddhi/source/RandomWordSource.java
new file mode 100644
index 0000000..23fc506
--- /dev/null
+++ b/src/test/java/org/apache/flink/contrib/siddhi/source/RandomWordSource.java
@@ -0,0 +1,111 @@
+/*
+ * 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.contrib.siddhi.source;
+
+import org.apache.flink.streaming.api.functions.source.SourceFunction;
+
+import java.util.Random;
+
+public class RandomWordSource implements SourceFunction {
+ private static final String[] WORDS = new String[]{
+ "To be, or not to be,--that is the question:--",
+ "Whether 'tis nobler in the mind to suffer",
+ "The slings and arrows of outrageous fortune",
+ "Or to take arms against a sea of troubles,",
+ "And by opposing end them?--To die,--to sleep,--",
+ "No more; and by a sleep to say we end",
+ "The heartache, and the thousand natural shocks",
+ "That flesh is heir to,--'tis a consummation",
+ "Devoutly to be wish'd. To die,--to sleep;--",
+ "To sleep! perchance to dream:--ay, there's the rub;",
+ "For in that sleep of death what dreams may come,",
+ "When we have shuffled off this mortal coil,",
+ "Must give us pause: there's the respect",
+ "That makes calamity of so long life;",
+ "For who would bear the whips and scorns of time,",
+ "The oppressor's wrong, the proud man's contumely,",
+ "The pangs of despis'd love, the law's delay,",
+ "The insolence of office, and the spurns",
+ "That patient merit of the unworthy takes,",
+ "When he himself might his quietus make",
+ "With a bare bodkin? who would these fardels bear,",
+ "To grunt and sweat under a weary life,",
+ "But that the dread of something after death,--",
+ "The undiscover'd country, from whose bourn",
+ "No traveller returns,--puzzles the will,",
+ "And makes us rather bear those ills we have",
+ "Than fly to others that we know not of?",
+ "Thus conscience does make cowards of us all;",
+ "And thus the native hue of resolution",
+ "Is sicklied o'er with the pale cast of thought;",
+ "And enterprises of great pith and moment,",
+ "With this regard, their currents turn awry,",
+ "And lose the name of action.--Soft you now!",
+ "The fair Ophelia!--Nymph, in thy orisons",
+ "Be all my sins remember'd."
+ };
+
+ private final int count;
+ private final Random random;
+ private final long initialTimestamp;
+
+ private volatile boolean isRunning = true;
+ private volatile int number = 0;
+ private long closeDelayTimestamp;
+
+ public RandomWordSource(int count, long initialTimestamp) {
+ this.count = count;
+ this.random = new Random();
+ this.initialTimestamp = initialTimestamp;
+ }
+
+ public RandomWordSource() {
+ this(Integer.MAX_VALUE, System.currentTimeMillis());
+ }
+
+ public RandomWordSource(int count) {
+ this(count, System.currentTimeMillis());
+ }
+
+
+ public RandomWordSource closeDelay(long delayTimestamp) {
+ this.closeDelayTimestamp = delayTimestamp;
+ return this;
+ }
+
+ @Override
+ public void run(SourceContext