public static Read<DynamicMessage> readProtoDynamicMessages(
      ProtoDomain domain, String fullMessageName) {
    SerializableFunction<PubsubMessage, DynamicMessage> parser =
        message -> {
          try {
            return DynamicMessage.parseFrom(
                domain.getDescriptor(fullMessageName), message.getPayload());
          } catch (InvalidProtocolBufferException e) {
            throw new RuntimeException("Could not parse Pub/Sub message", e);
          }
        };

    ProtoDynamicMessageSchema<DynamicMessage> schema =
        ProtoDynamicMessageSchema.forDescriptor(domain, domain.getDescriptor(fullMessageName));
    return Read.newBuilder(parser)
        .setCoder(
            SchemaCoder.of(
                schema.getSchema(),
                TypeDescriptor.of(DynamicMessage.class),
                schema.getToRowFunction(),
                schema.getFromRowFunction()))
        .build();
  }
@Test
  public void testProtoDynamicMessagesFromDescriptor() {
    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.readProtoDynamicMessages(Primitive.getDescriptor())
                    .fromSubscription(SUBSCRIPTION.getPath())
                    .withClock(CLOCK)
                    .withClientFactory(clientFactory))
            .apply(
                "Return To Primitive",
                MapElements.into(TypeDescriptor.of(Primitive.class))
                    .via(
                        (DynamicMessage message) -> {
                          try {
                            return Primitive.parseFrom(message.toByteArray());
                          } catch (InvalidProtocolBufferException e) {
                            throw new RuntimeException("Could not return to Primitive", e);
                          }
                        }));

    PAssert.that(read).containsInAnyOrder(inputs);
    pipeline.run();
  }