@Test public void testFormatCastsFieldsToJsonString() throws Exception { ObjectNode json = Json.createObjectNode(); json.put("test_bool", true) .put("test_long", -7) .put("test_double", 0.99) .put("test_string", "hello world"); json.set("test_object", Json.createObjectNode().put("test_field", true)); json.set("test_array", Json.createArrayNode().add(1).add(2).add(3)); byte[] data = Json.asBytes(json); PubsubMessage message = new PubsubMessage(data, Collections.emptyMap()); Schema schema = SchemaBuilder.record("root").fields() .name("test_bool").type().stringType().noDefault() .name("test_long").type().stringType().noDefault() .name("test_double").type().stringType().noDefault() .name("test_string").type().stringType().noDefault() .name("test_object").type().stringType().noDefault() .name("test_array").type().stringType().noDefault() .endRecord(); PubsubMessageRecordFormatter formatter = new PubsubMessageRecordFormatter(); GenericRecord record = formatter.formatRecord(message, schema); assertEquals("true", record.get("test_bool").toString()); assertEquals("-7", record.get("test_long").toString()); assertEquals("0.99", record.get("test_double").toString()); assertEquals("hello world", record.get("test_string").toString()); assertEquals("{\"test_field\":true}", record.get("test_object").toString()); assertEquals("[1,2,3]", record.get("test_array").toString()); }
@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);  }