@Test public void testOutput() { ValueProvider<String> schemasLocation = pipeline.newProvider("schemas.tar.gz"); final List<String> input = Arrays.asList("{}", "{\"id\":null}", "[]", "{", "{\"clientId\":\"2907648d-711b-4e9f-94b5-52a2b40a44b1\"}", "{\"clientId\":\"2907648D-711B-4E9F-94B5-52A2B40A44B1\"}", "{\"impression_id\":\"{2907648d-711b-4e9f-94b5-52a2b40a44b1}\"}", "{\"client_id\":\"n/a\"}", "{\"client_id\":\"n/a\",\"impression_id\":\"{2907648d-711b-4e9f-94b5-52a2b40a44b1}\"}"); Result<PCollection<PubsubMessage>, PubsubMessage> output = pipeline.apply(Create.of(input)) .apply(InputFileFormat.text.decode()) .apply("AddAttributes", MapElements.into(TypeDescriptor.of(PubsubMessage.class)) .via(element -> new PubsubMessage(element.getPayload(), ImmutableMap.of( "document_namespace", "test", "document_type", "test", "document_version", "1")))) .apply(ParsePayload.of(schemasLocation)); final List<String> expectedMain = Arrays.asList("{}", "{\"id\":null}", "{\"clientId\":\"2907648d-711b-4e9f-94b5-52a2b40a44b1\"}", "{\"clientId\":\"2907648d-711b-4e9f-94b5-52a2b40a44b1\"}", "{\"impression_id\":\"{2907648d-711b-4e9f-94b5-52a2b40a44b1}\"}", "{\"client_id\":\"n/a\"}", "{\"client_id\":\"n/a\",\"impression_id\":\"{2907648d-711b-4e9f-94b5-52a2b40a44b1}\"}"); final PCollection<String> main = output.output().apply("encodeTextMain", OutputFileFormat.text.encode()); PAssert.that(main).containsInAnyOrder(expectedMain); final List<String> expectedAttributes = Arrays.asList( "{\"document_namespace\":\"test\",\"document_version\":\"1\",\"document_type\":\"test\"}", "{\"document_namespace\":\"test\",\"document_version\":\"1\",\"document_type\":\"test\"}", "{\"document_namespace\":\"test\",\"document_version\":\"1\"" + ",\"client_id\":\"2907648d-711b-4e9f-94b5-52a2b40a44b1\"" + ",\"document_type\":\"test\",\"sample_id\":\"67\"}", "{\"document_namespace\":\"test\",\"document_version\":\"1\"" + ",\"client_id\":\"2907648d-711b-4e9f-94b5-52a2b40a44b1\"" + ",\"document_type\":\"test\",\"sample_id\":\"67\"}", "{\"document_namespace\":\"test\",\"document_version\":\"1\"" + ",\"document_type\":\"test\",\"sample_id\":\"67\"}", "{\"document_namespace\":\"test\",\"document_version\":\"1\"" + ",\"document_type\":\"test\"}", "{\"document_namespace\":\"test\",\"document_version\":\"1\"" + ",\"document_type\":\"test\",\"sample_id\":\"67\"}"); final PCollection<String> attributes = output.output() .apply(MapElements.into(TypeDescriptors.strings()).via(m -> { try { return Json.asString(m.getAttributeMap()); } catch (IOException e) { throw new UncheckedIOException(e); } })); PAssert.that(attributes).containsInAnyOrder(expectedAttributes); final List<String> expectedError = Arrays.asList("[]", "{"); final PCollection<String> error = output.failures().apply("encodeTextError", OutputFileFormat.text.encode()); PAssert.that(error).containsInAnyOrder(expectedError); 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);  }