AmqpIO { public static Read read() { return new AutoValue_AmqpIO_Read.Builder().setMaxNumRecords(Long.MAX_VALUE).build(); } private  AmqpIO(); static Read read(); static Write write();  }
@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(); }