public static Read read() {
    return new AutoValue_AmqpIO_Read.Builder().setMaxNumRecords(Long.MAX_VALUE).build();
  }
@Test
  public void testRead() throws Exception {
    PCollection<Message> output =
        pipeline.apply(
            AmqpIO.read()
                .withMaxNumRecords(100)
                .withAddresses(Collections.singletonList(broker.getQueueUri("testRead"))));
    PAssert.thatSingleton(output.apply(Count.globally())).isEqualTo(100L);

    Messenger sender = Messenger.Factory.create();
    sender.start();
    for (int i = 0; i < 100; i++) {
      Message message = Message.Factory.create();
      message.setAddress(broker.getQueueUri("testRead"));
      message.setBody(new AmqpValue("Test " + i));
      sender.put(message);
      sender.send();
    }
    sender.stop();

    pipeline.run();
  }