@Test public void testMetadataInPayload() { ValueProvider<String> schemasLocation = pipeline.newProvider("schemas.tar.gz"); String input = "{\"id\":null,\"document_id\":\"2c3a0767-d84a-4d02-8a92-fa54a3376049\"" + ",\"metadata\":{\"document_namespace\":\"test\",\"document_type\":\"test\"" + ",\"document_version\":\"1\",\"geo\":{\"country\":\"FI\"}}}"; Result<PCollection<PubsubMessage>, PubsubMessage> result = pipeline .apply(Create.of(input)) .apply(InputFileFormat.text.decode()) .apply(ParsePayload.of(schemasLocation)); PAssert.that(result.failures()).empty(); final PCollection<Integer> attributeCounts = result.output().apply(MapElements .into(TypeDescriptors.integers()).via(message -> message.getAttributeMap().size())); PAssert.thatSingleton(attributeCounts).isEqualTo(5); final String expectedMain = "{\"id\":null}"; final PCollection<String> main = result.output() .apply("encodeTextMain", OutputFileFormat.text.encode()); PAssert.thatSingleton(main).isEqualTo(expectedMain); pipeline.run(); }
public static ParsePayload of(ValueProvider<String> schemasLocation) { return new ParsePayload(schemasLocation); }
ParsePayload extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParsePayload of(ValueProvider<String> schemasLocation) { return new ParsePayload(schemasLocation); } }
ParsePayload extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParsePayload of(ValueProvider<String> schemasLocation) { return new ParsePayload(schemasLocation); } private  ParsePayload(ValueProvider<String> schemasLocation); }
ParsePayload extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParsePayload of(ValueProvider<String> schemasLocation) { return new ParsePayload(schemasLocation); } private  ParsePayload(ValueProvider<String> schemasLocation); static ParsePayload of(ValueProvider<String> schemasLocation); @Override Result<PCollection<PubsubMessage>, PubsubMessage> expand(
      PCollection<PubsubMessage> messages); static String getVersionFromTelemetryPayload(JsonNode json); static void addClientIdFromPayload(Map<String, String> attributes, ObjectNode json); }
ParsePayload extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParsePayload of(ValueProvider<String> schemasLocation) { return new ParsePayload(schemasLocation); } private  ParsePayload(ValueProvider<String> schemasLocation); static ParsePayload of(ValueProvider<String> schemasLocation); @Override Result<PCollection<PubsubMessage>, PubsubMessage> expand(
      PCollection<PubsubMessage> messages); static String getVersionFromTelemetryPayload(JsonNode json); static void addClientIdFromPayload(Map<String, String> attributes, ObjectNode json);  }