public static Write write() {
    return new AutoValue_RabbitMqIO_Write.Builder()
        .setExchangeDeclare(false)
        .setQueueDeclare(false)
        .build();
  }
@Test
  public void testWriteQueue() throws Exception {
    final int maxNumRecords = 1000;
    List<RabbitMqMessage> data =
        RabbitMqTestUtils.generateRecords(maxNumRecords).stream()
            .map(RabbitMqMessage::new)
            .collect(Collectors.toList());
    p.apply(Create.of(data))
        .apply(
            RabbitMqIO.write().withUri("amqp://guest:guest@localhost:" + port).withQueue("TEST"));

    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("TEST", true, false, false, null);
      RabbitMqTestUtils.TestConsumer consumer = new RabbitMqTestUtils.TestConsumer(channel);
      channel.basicConsume("TEST", true, consumer);

      p.run();

      while (consumer.getReceived().size() < maxNumRecords) {
        Thread.sleep(500);
      }

      assertEquals(maxNumRecords, consumer.getReceived().size());
      for (int i = 0; i < maxNumRecords; i++) {
        assertTrue(consumer.getReceived().contains("Test " + i));
      }
    } finally {
      if (channel != null) {
        channel.close();
      }
      if (connection != null) {
        connection.close();
      }
    }
  }