JmsIO { public static <T> Read<T> readMessage() { return new AutoValue_JmsIO_Read.Builder<T>().setMaxNumRecords(Long.MAX_VALUE).build(); } private  JmsIO(); static Read<JmsRecord> read(); static Read<T> readMessage(); static Write write();  }
@Test public void testReadBytesMessages() throws Exception { Connection connection = connectionFactory.createConnection(USERNAME, PASSWORD); Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageProducer producer = session.createProducer(session.createQueue(QUEUE)); BytesMessage message = session.createBytesMessage(); message.writeBytes("This Is A Test".getBytes(StandardCharsets.UTF_8)); producer.send(message); producer.close(); session.close(); connection.close(); PCollection<String> output = pipeline.apply( JmsIO.<String>readMessage() .withConnectionFactory(connectionFactory) .withQueue(QUEUE) .withUsername(USERNAME) .withPassword(PASSWORD) .withMaxNumRecords(1) .withCoder(SerializableCoder.of(String.class)) .withMessageMapper(new BytesMessageToStringMessageMapper())); PAssert.thatSingleton(output.apply("Count", Count.<String>globally())).isEqualTo(1L); pipeline.run(); connection = connectionFactory.createConnection(USERNAME, PASSWORD); session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); MessageConsumer consumer = session.createConsumer(session.createQueue(QUEUE)); Message msg = consumer.receiveNoWait(); assertNull(msg); }