public static <T> Read<T> readMessage() {
    return new AutoValue_JmsIO_Read.Builder<T>()
        .setMaxNumRecords(Long.MAX_VALUE)
        .setCloseTimeout(DEFAULT_CLOSE_TIMEOUT)
        .setRequiresDeduping(false)
        .build();
  }
@Test
  public void testReadBytesMessagesWithCFProviderFn() throws Exception {
    long count = 5;
    produceTestMessages(count, JmsIOTest::createBytesMessage);

    PCollection<String> output =
        pipeline.apply(
            JmsIO.<String>readMessage()
                .withConnectionFactoryProviderFn(
                    toSerializableFunction(commonJms::createConnectionFactory))
                .withQueue(QUEUE)
                .withUsername(USERNAME)
                .withPassword(PASSWORD)
                .withMaxNumRecords(count)
                .withMessageMapper(new CommonJms.BytesMessageToStringMessageMapper())
                .withCoder(StringUtf8Coder.of()));

    PAssert.thatSingleton(output.apply("Count", Count.globally())).isEqualTo(count);
    pipeline.run();

    assertQueueIsEmpty();
  }