@Test public void testWithGeoCityLookup() { final List<String> input = Arrays.asList( "{\"attributeMap\":{},\"payload\":\"\"}", "{\"attributeMap\":" + "{\"x_forwarded_for\":\"_, 202.196.224.0, _\"" + "},\"payload\":\"notProxied++\"}", "{\"attributeMap\":" + "{\"x_pipeline_proxy\":1" + ",\"x_forwarded_for\":\"_, 202.196.224.0, _, _\"" + "},\"payload\":\"proxied+\"}", "{\"attributeMap\":" + "{\"x_pipeline_proxy\":\"2000-01-01T00:00:00.000000Z\"" + ",\"x_forwarded_for\":\"_, 202.196.224.0, _, _\"" + "},\"payload\":\"proxiedWithTimestamp\"}"); final List<String> expected = Arrays.asList( "{\"attributeMap\":{\"geo_db_version\":\"2019-01-03T21:26:19Z\"},\"payload\":\"\"}", "{\"attributeMap\":" + "{\"geo_country\":\"PH\"" + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\"" + "},\"payload\":\"notProxied++\"}", "{\"attributeMap\":" + "{\"geo_country\":\"PH\"" + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\"" + "},\"payload\":\"proxied+\"}", "{\"attributeMap\":" + "{\"geo_country\":\"PH\"" + ",\"geo_db_version\":\"2019-01-03T21:26:19Z\"" + ",\"submission_timestamp\":\"2000-01-01T00:00:00.000000Z\"" + "},\"payload\":\"proxiedWithTimestamp\"}"); final PCollection<String> output = pipeline .apply(Create.of(input)) .apply(InputFileFormat.json.decode()) .apply(ParseProxy.of()) .apply(GeoCityLookup .of(pipeline.newProvider("src/test/resources/cityDB/GeoIP2-City-Test.mmdb"), null)) .apply(OutputFileFormat.json.encode()); PAssert.that(output).containsInAnyOrder(expected); final PipelineResult result = pipeline.run(); }
public static ParseProxy of() { return INSTANCE; }
ParseProxy extends PTransform<PCollection<PubsubMessage>, PCollection<PubsubMessage>> { public static ParseProxy of() { return INSTANCE; } }
ParseProxy extends PTransform<PCollection<PubsubMessage>, PCollection<PubsubMessage>> { public static ParseProxy of() { return INSTANCE; } private  ParseProxy(); }
ParseProxy extends PTransform<PCollection<PubsubMessage>, PCollection<PubsubMessage>> { public static ParseProxy of() { return INSTANCE; } private  ParseProxy(); static ParseProxy of(); @Override PCollection<PubsubMessage> expand(PCollection<PubsubMessage> input); }
ParseProxy extends PTransform<PCollection<PubsubMessage>, PCollection<PubsubMessage>> { public static ParseProxy of() { return INSTANCE; } private  ParseProxy(); static ParseProxy of(); @Override PCollection<PubsubMessage> expand(PCollection<PubsubMessage> input);  }