PubsubIO { public static <T> Read<T> readMessagesWithCoderAndParseFn( Coder<T> coder, SimpleFunction<PubsubMessage, T> parseFn) { return Read.newBuilder(parseFn).setCoder(coder).build(); } private  PubsubIO(); static Read<PubsubMessage> readMessages(); static Read<PubsubMessage> readMessagesWithMessageId(); static Read<PubsubMessage> readMessagesWithAttributes(); static Read<PubsubMessage> readMessagesWithAttributesAndMessageId(); static Read<String> readStrings(); static Read<T> readProtos(Class<T> messageClass); static Read<T> readAvros(Class<T> clazz); static Read<T> readMessagesWithCoderAndParseFn(
      Coder<T> coder, SimpleFunction<PubsubMessage, T> parseFn); @Experimental(Kind.SCHEMAS) static Read<GenericRecord> readAvroGenericRecords(org.apache.avro.Schema avroSchema); @Experimental(Kind.SCHEMAS) static Read<T> readAvrosWithBeamSchema(Class<T> clazz); static Write<PubsubMessage> writeMessages(); static Write<String> writeStrings(); static Write<T> writeProtos(Class<T> messageClass); static Write<T> writeAvros(Class<T> clazz);  }
@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 = readPipeline.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); readPipeline.run(); }