JmsIO { public static Write write() { return new AutoValue_JmsIO_Write.Builder().build(); } private  JmsIO(); static Read<JmsRecord> read(); static Read<T> readMessage(); static Write write();  }
@Test public void testWriteMessage() throws Exception { ArrayList<String> data = new ArrayList<>(); for (int i = 0; i < 100; i++) { data.add("Message " + i); } pipeline .apply(Create.of(data)) .apply( JmsIO.write() .withConnectionFactory(connectionFactory) .withQueue(QUEUE) .withUsername(USERNAME) .withPassword(PASSWORD)); pipeline.run(); Connection connection = connectionFactory.createConnection(USERNAME, PASSWORD); connection.start(); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE)); int count = 0; while (consumer.receive(1000) != null) { count++; } assertEquals(100, count); }