@Test public void testParse() throws Exception { String inputPath = Resources.getResource("testdata").getPath(); String input = inputPath + "/logentry.ndjson"; String expected = Json.asString(ImmutableMap.builder() .put("ecosystem_anon_id", "fake") .put("oauth_client_id", "2882386c6d801776") .put("event", "oauth.token.created") .put("country", "United States") .put("region", "Virginia") .build()); PCollection<PubsubMessage> output = pipeline .apply(new FileInput(StaticValueProvider.of(input), InputFileFormat.json)) .apply(ParseLogEntry.of()).output(); PAssert.that(output.apply(OutputFileFormat.text.encode())) .containsInAnyOrder(ImmutableList.of(expected)); pipeline.run(); }
public static ParseLogEntry of() { return new ParseLogEntry(); }
ParseLogEntry extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParseLogEntry of() { return new ParseLogEntry(); } }
ParseLogEntry extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParseLogEntry of() { return new ParseLogEntry(); }  }
ParseLogEntry extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParseLogEntry of() { return new ParseLogEntry(); }  static ParseLogEntry of(); @Override Result<PCollection<PubsubMessage>, PubsubMessage> expand(
      PCollection<PubsubMessage> input); }
ParseLogEntry extends
    PTransform<PCollection<PubsubMessage>, Result<PCollection<PubsubMessage>, PubsubMessage>> { public static ParseLogEntry of() { return new ParseLogEntry(); }  static ParseLogEntry of(); @Override Result<PCollection<PubsubMessage>, PubsubMessage> expand(
      PCollection<PubsubMessage> input);  }