SiddhiTypeFactory { public static <T extends Tuple> TypeInformation<T> getTupleTypeInformation(AbstractDefinition definition) { List<TypeInformation> types = new ArrayList<>(); for (Attribute attribute : definition.getAttributeList()) { types.add(TypeInformation.of(getJavaType(attribute.getType()))); } try { return Types.TUPLE(types.toArray(new TypeInformation[0])); } catch (IllegalArgumentException ex) { throw new IllegalArgumentException("Unable to parse ", ex); } }  static void registerType(Class<?> javaType, Attribute.Type siddhiType); static AbstractDefinition getStreamDefinition(String executionPlan, String streamId); static AbstractDefinition getStreamDefinition(String executionPlan, String streamId, SiddhiOperatorContext siddhiOperatorContext); static TypeInformation<T> getTupleTypeInformation(AbstractDefinition definition); static TypeInformation<T> getTupleTypeInformation(String executionPlan, String streamId); static TypeInformation<T> getTupleTypeInformation(String executionPlan, String streamId, SiddhiOperatorContext siddhiOperatorContext); static TypeInformation<GenericRecord> getMapTypeInformation(); static Attribute.Type getAttributeType(TypeInformation<F> fieldType); static Class<?> getJavaType(Attribute.Type attributeType); static TypeInformation<Tuple2<StreamRoute, T>> getStreamTupleTypeInformation(TypeInformation<T> typeInformation);  }
@Test public void testBuildTypeInformationForSiddhiStream() { String query = "define stream inputStream (timestamp long, name string, value double);" + "from inputStream select name, value insert into outputStream;"; TypeInformation<Tuple3<Long, String, Double>> inputStreamType = SiddhiTypeFactory.getTupleTypeInformation(query, "inputStream"); TypeInformation<Tuple2<String, Double>> outputStreamType = SiddhiTypeFactory.getTupleTypeInformation(query, "outputStream"); Assert.assertNotNull(inputStreamType); Assert.assertNotNull(outputStreamType); }