### Question:
ParseIp extends PTransform<PCollection<PubsubMessage>, PCollection<PubsubMessage>> { public static ParseIp of() { return INSTANCE; }  static ParseIp of(); @Override PCollection<PubsubMessage> expand(PCollection<PubsubMessage> input);  }### Answer:
@Test public void testParseIp() { final List<String> input = Collections .singletonList("{\"attributeMap\":" + "{\"remote_addr\":\"127.0.0.1\"" + ",\"x_forwarded_for\":\"123.321.123,45.54.45,67.76.67\"" + "},\"payload\":\"\"}"); PCollection<String> output = pipeline.apply(Create.of(input)) .apply(InputFileFormat.json.decode()).apply(ParseIp.of()) .apply(OutputFileFormat.json.encode()); final List<String> expected = Collections.singletonList( "{\"attributeMap\":" + "{\"client_ip\":\"45.54.45\"" + ",\"remote_addr\":\"127.0.0.1\"" + ",\"x_forwarded_for\":\"123.321.123,45.54.45,67.76.67\"" + "},\"payload\":\"\"}"); PAssert.that(output).containsInAnyOrder(expected); pipeline.run(); }