@Test public void testFormatWithNestedObjectStruct() { byte[] data = generateNestedObject(); PubsubMessage message = new PubsubMessage(data, Collections.emptyMap()); Schema schema = SchemaBuilder .record("root").fields() .name("shape").type().record("shape").fields() .name("quadrilateral").type().record("quadrilateral").fields() .name("square").type().booleanType().noDefault() .name("rectangle").type().booleanType().noDefault() .name("rhombus").type().booleanType().noDefault() .name("parallelogram").type().booleanType().noDefault() .name("trapezoid").type().booleanType().noDefault() .name("kite").type().booleanType().noDefault() .endRecord().noDefault() .endRecord().noDefault() .endRecord(); PubsubMessageRecordFormatter formatter = new PubsubMessageRecordFormatter(); GenericRecord record = formatter.formatRecord(message, schema); GenericRecord shape = (GenericRecord) record.get("shape"); GenericRecord quad = (GenericRecord) shape.get("quadrilateral"); assertEquals(true, quad.get("rhombus")); assertEquals(null, shape.get("triangle")); }
@Override public GenericRecord formatRecord(PubsubMessage element, Schema schema) { GenericRecord result = null; try { InputStream in = new ByteArrayInputStream(element.getPayload()); DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema); Decoder decoder = new GuidedJsonDecoder(schema, in); result = reader.read(null, decoder); } catch (IOException e) { throw new UncheckedIOException(e); } catch (AvroTypeException e) { throw e; } return result; }
PubsubMessageRecordFormatter implements RecordFormatter<PubsubMessage> { @Override public GenericRecord formatRecord(PubsubMessage element, Schema schema) { GenericRecord result = null; try { InputStream in = new ByteArrayInputStream(element.getPayload()); DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema); Decoder decoder = new GuidedJsonDecoder(schema, in); result = reader.read(null, decoder); } catch (IOException e) { throw new UncheckedIOException(e); } catch (AvroTypeException e) { throw e; } return result; } }
PubsubMessageRecordFormatter implements RecordFormatter<PubsubMessage> { @Override public GenericRecord formatRecord(PubsubMessage element, Schema schema) { GenericRecord result = null; try { InputStream in = new ByteArrayInputStream(element.getPayload()); DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema); Decoder decoder = new GuidedJsonDecoder(schema, in); result = reader.read(null, decoder); } catch (IOException e) { throw new UncheckedIOException(e); } catch (AvroTypeException e) { throw e; } return result; }  }
PubsubMessageRecordFormatter implements RecordFormatter<PubsubMessage> { @Override public GenericRecord formatRecord(PubsubMessage element, Schema schema) { GenericRecord result = null; try { InputStream in = new ByteArrayInputStream(element.getPayload()); DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema); Decoder decoder = new GuidedJsonDecoder(schema, in); result = reader.read(null, decoder); } catch (IOException e) { throw new UncheckedIOException(e); } catch (AvroTypeException e) { throw e; } return result; }  @Override GenericRecord formatRecord(PubsubMessage element, Schema schema); }
PubsubMessageRecordFormatter implements RecordFormatter<PubsubMessage> { @Override public GenericRecord formatRecord(PubsubMessage element, Schema schema) { GenericRecord result = null; try { InputStream in = new ByteArrayInputStream(element.getPayload()); DatumReader<GenericRecord> reader = new GenericDatumReader<>(schema); Decoder decoder = new GuidedJsonDecoder(schema, in); result = reader.read(null, decoder); } catch (IOException e) { throw new UncheckedIOException(e); } catch (AvroTypeException e) { throw e; } return result; }  @Override GenericRecord formatRecord(PubsubMessage element, Schema schema);  }