AmqpIO { public static Write write() { return new AutoValue_AmqpIO_Write(); } private  AmqpIO(); static Read read(); static Write 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 + "}")); } }