@Test public void testOutput() { final List<String> input = Arrays.asList( "{\"attributeMap\":{},\"payload\":\"\"}", "{\"attributeMap\":" + "{\"x_pipeline_proxy\":1" + "},\"payload\":\"proxied+\"}", "{\"attributeMap\":" + "{\"x_pipeline_proxy\":\"\"" + "},\"payload\":\"emptyXPP\"}", "{\"attributeMap\":" + "{\"submission_timestamp\":\"2000-01-01T00:00:00.000000Z\"" + ",\"x_forwarded_for\":\"3, 2, 1\"" + "},\"payload\":\"notProxied++\"}", "{\"attributeMap\":" + "{\"submission_timestamp\":\"2000-01-01T00:00:00.000000Z\"" + ",\"x_forwarded_for\":\"4, 3, 2, 1\"" + ",\"x_pipeline_proxy\":\"1999-12-31T23:59:59.999999Z\"" + "},\"payload\":\"proxiedWithTimestamp\"}", "{\"attributeMap\":" + "{\"submission_timestamp\":\"1999-12-31T23:59:59.999999Z\"" + ",\"proxy_timestamp\":\"2000-01-01T00:00:00.000000Z\"" + ",\"x_forwarded_for\":\"4, 3, 2\"" + "},\"payload\":\"retried+\"}"); final List<String> expected = Arrays.asList( "{\"attributeMap\":{},\"payload\":\"\"}", "{\"attributeMap\":{},\"payload\":\"proxied+\"}", "{\"attributeMap\":{},\"payload\":\"emptyXPP\"}", "{\"attributeMap\":" + "{\"submission_timestamp\":\"2000-01-01T00:00:00.000000Z\"" + ",\"x_forwarded_for\":\"3, 2, 1\"" + "},\"payload\":\"notProxied++\"}", "{\"attributeMap\":" + "{\"proxy_timestamp\":\"2000-01-01T00:00:00.000000Z\"" + ",\"submission_timestamp\":\"1999-12-31T23:59:59.999999Z\"" + ",\"x_forwarded_for\":\"4, 3, 2\"" + "},\"payload\":\"proxiedWithTimestamp\"}", "{\"attributeMap\":" + "{\"proxy_timestamp\":\"2000-01-01T00:00:00.000000Z\"" + ",\"submission_timestamp\":\"1999-12-31T23:59:59.999999Z\"" + ",\"x_forwarded_for\":\"4, 3, 2\"" + "},\"payload\":\"retried+\"}"); final PCollection<String> output = pipeline .apply(Create.of(input)) .apply(InputFileFormat.json.decode()) .apply(ParseProxy.of()) .apply(OutputFileFormat.json.encode()); PAssert.that(output).containsInAnyOrder(expected); final PipelineResult result = pipeline.run(); final List<MetricResult<Long>> counters = Lists.newArrayList(result.metrics() .queryMetrics(MetricsFilter.builder() .addNameFilter(MetricNameFilter.inNamespace(ParseProxy.Fn.class)).build()) .getCounters()); assertEquals(1, counters.size()); counters.forEach(counter -> assertThat(counter.getCommitted(), greaterThan(0L))); }
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);  }