JmsRecordReader implements RecordReader<Message> { @Override public void open() throws Exception { LOGGER.debug("Opening JMS connection"); connection = connectionFactory.createConnection(); session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); messageConsumer = session.createConsumer(destination); connection.start(); } JmsRecordReader(final ConnectionFactory connectionFactory, final Destination destination);  JmsRecordReader(final ConnectionFactory connectionFactory, final Destination destination, final long timeout); @Override void open(); @Override JmsRecord readRecord(); @Override void close(); static final long DEFAULT_TIMEOUT; }
@Test public void testOpen() throws Exception { jmsRecordReader.open(); verify(connectionFactory).createConnection(); verify(connection).createSession(false, Session.AUTO_ACKNOWLEDGE); verify(session).createConsumer(destination); verify(connection).start(); }