public static <T> Read<T> readMessagesWithCoderAndParseFn(
      Coder<T> coder, SimpleFunction<PubsubMessage, T> parseFn) {
    return Read.newBuilder(parseFn).setCoder(coder).build();
  }
@Test
  public void testReadMessagesWithCoderAndParseFn() {
    Coder<PubsubMessage> coder = PubsubMessagePayloadOnlyCoder.of();
    List<PubsubMessage> inputs =
        ImmutableList.of(
            new PubsubMessage("foo".getBytes(StandardCharsets.UTF_8), new HashMap<>()),
            new PubsubMessage("bar".getBytes(StandardCharsets.UTF_8), new HashMap<>()));
    setupTestClient(inputs, coder);

    PCollection<String> read =
        pipeline.apply(
            PubsubIO.readMessagesWithCoderAndParseFn(
                    StringUtf8Coder.of(), new StringPayloadParseFn())
                .fromSubscription(SUBSCRIPTION.getPath())
                .withClock(CLOCK)
                .withClientFactory(clientFactory));

    List<String> outputs = ImmutableList.of("foo", "bar");
    PAssert.that(read).containsInAnyOrder(outputs);
    pipeline.run();
  }