StreamSerializer implements Serializable { 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.isRowType()) { Row row = (Row) input; data = new Object[schema.getFieldIndexes().length]; for (int i = 0; i < schema.getFieldIndexes().length; i++) { data[i] = row.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; } StreamSerializer(StreamSchema<T> schema); Object[] getRow(T input);  }
@Test public void testAtomicType() { StreamSchema<String> schema = new StreamSchema<>(TypeExtractor.createTypeInfo(String.class), "word"); StreamSerializer<String> reader = new StreamSerializer<>(schema); Assert.assertArrayEquals(new Object[]{"Siddhi"}, reader.getRow("Siddhi")); }
@Test public void testPojoType() { Event event = new Event(); event.setId(1); event.setName("test"); event.setPrice(56.7); event.setTimestamp(CURRENT); StreamSchema<Event> schema = new StreamSchema<>(TypeExtractor.createTypeInfo(Event.class), "id", "name", "price", "timestamp"); StreamSerializer<Event> reader = new StreamSerializer<>(schema); Assert.assertArrayEquals(new Object[]{1, "test", 56.7, CURRENT}, reader.getRow(event)); }
@Test public void testRowType() { Row row = Row.of(1, "test", 56.7, CURRENT); StreamSchema<Row> schema = new StreamSchema<>(new RowTypeInfo( TypeExtractor.createTypeInfo(Integer.class), TypeExtractor.createTypeInfo(String.class), TypeExtractor.createTypeInfo(Double.class), TypeExtractor.createTypeInfo(Long.class) ), "id", "name", "price", "timestamp"); StreamSerializer<Row> reader = new StreamSerializer<>(schema); Assert.assertArrayEquals(new Object[]{1, "test", 56.7, CURRENT}, reader.getRow(row)); }
@Test public void testTupleType() { Tuple4 row = Tuple4.of(1, "test", 56.7, CURRENT); StreamSchema<Tuple4> schema = new StreamSchema<>(new TupleTypeInfo<>( TypeExtractor.createTypeInfo(Integer.class), TypeExtractor.createTypeInfo(String.class), TypeExtractor.createTypeInfo(Double.class), TypeExtractor.createTypeInfo(Long.class)) , "id", "name", "price", "timestamp"); StreamSerializer<Tuple4> reader = new StreamSerializer<>(schema); Assert.assertArrayEquals(new Object[]{1, "test", 56.7, CURRENT}, reader.getRow(row)); }