public static Write write() {
    return new AutoValue_AmqpIO_Write();
  }
@Test
  public void testWrite() throws Exception {
    List<Message> data = new ArrayList<>();
    for (int i = 0; i < 100; i++) {
      Message message = Message.Factory.create();
      message.setBody(new AmqpValue("Test " + i));
      message.setAddress(broker.getQueueUri("testWrite"));
      message.setSubject("test");
      data.add(message);
    }
    pipeline.apply(Create.of(data).withCoder(AmqpMessageCoder.of())).apply(AmqpIO.write());
    pipeline.run().waitUntilFinish();

    List<String> received = new ArrayList<>();
    Messenger messenger = Messenger.Factory.create();
    messenger.start();
    messenger.subscribe(broker.getQueueUri("testWrite"));
    while (received.size() < 100) {
      messenger.recv();
      while (messenger.incoming() > 0) {
        Message message = messenger.get();
        LOG.info("Received: " + message.getBody().toString());
        received.add(message.getBody().toString());
      }
    }
    messenger.stop();

    assertEquals(100, received.size());
    for (int i = 0; i < 100; i++) {
      assertTrue(received.contains("AmqpValue{Test " + i + "}"));
    }
  }