public static <T extends Message> Read<T> readProtos(Class<T> messageClass) {
    // TODO: Stop using ProtoCoder and instead parse the payload directly.
    // We should not be relying on the fact that ProtoCoder's wire format is identical to
    // the protobuf wire format, as the wire format is not part of a coder's API.
    ProtoCoder<T> coder = ProtoCoder.of(messageClass);
    return Read.newBuilder(parsePayloadUsingCoder(coder)).setCoder(coder).build();
  }
@Test
  public void testProto() {
    ProtoCoder<Primitive> coder = ProtoCoder.of(Primitive.class);
    ImmutableList<Primitive> inputs =
        ImmutableList.of(
            Primitive.newBuilder().setPrimitiveInt32(42).build(),
            Primitive.newBuilder().setPrimitiveBool(true).build(),
            Primitive.newBuilder().setPrimitiveString("Hello, World!").build());
    setupTestClient(inputs, coder);
    PCollection<Primitive> read =
        pipeline.apply(
            PubsubIO.readProtos(Primitive.class)
                .fromSubscription(SUBSCRIPTION.getPath())
                .withClock(CLOCK)
                .withClientFactory(clientFactory));
    PAssert.that(read).containsInAnyOrder(inputs);
    pipeline.run();
  }