public static Read read() {
    return new AutoValue_RabbitMqIO_Read.Builder()
        .setQueueDeclare(false)
        .setExchangeDeclare(false)
        .setMaxReadTime(null)
        .setMaxNumRecords(Long.MAX_VALUE)
        .setUseCorrelationId(false)
        .build();
  }
@Test
  public void testReadQueue() throws Exception {
    final int maxNumRecords = 10;
    PCollection<RabbitMqMessage> raw =
        p.apply(
            RabbitMqIO.read()
                .withUri("amqp://guest:guest@localhost:" + port)
                .withQueue("READ")
                .withMaxNumRecords(maxNumRecords));
    PCollection<String> output =
        raw.apply(
            MapElements.into(TypeDescriptors.strings())
                .via(
                    (RabbitMqMessage message) ->
                        RabbitMqTestUtils.recordToString(message.getBody())));

    List<String> records =
        RabbitMqTestUtils.generateRecords(maxNumRecords).stream()
            .map(RabbitMqTestUtils::recordToString)
            .collect(Collectors.toList());
    PAssert.that(output).containsInAnyOrder(records);

    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setUri("amqp://guest:guest@localhost:" + port);
    Connection connection = null;
    Channel channel = null;
    try {
      connection = connectionFactory.newConnection();
      channel = connection.createChannel();
      channel.queueDeclare("READ", false, false, false, null);
      for (String record : records) {
        channel.basicPublish("", "READ", null, record.getBytes(StandardCharsets.UTF_8));
      }

      p.run();
    } finally {
      if (channel != null) {
        channel.close();
      }
      if (connection != null) {
        connection.close();
      }
    }
  }