ListeningContainerMessageReceiver implements MessageReceiver, MessageListener { @Override public List<SourceRecord> poll() { return IntStream.range(0, 10) .peek(i -> logger.trace("op=requestMessage i={}", i)) .filter(i -> this.repeatListener.allowConsumption()) .peek(i -> logger.trace("op=waitForMessage i={}", i)) .mapToObj(i -> getMessage()) .filter(Optional::isPresent) .map(opt -> (TextMessage) opt.get()) .peek(msg -> logger.debug("op=prepareRecord hash={}", msg.hashCode())) .map(message -> new SourceRecord( singletonMap("destination", destinationName), singletonMap(destinationName, getJmsMessageID(message)), topicsMappingHolder.getTopic(destinationName).get(), STRING_SCHEMA, MessageReceiver.getText(message))) .peek(record -> logger.trace("op=collect record={}", record)) .collect(Collectors.toList()); } ListeningContainerMessageReceiver(ConnectionFactory connectionFactory, JmsConnectorConfigurationProperties connectorConfigurationProperties, TopicsMappingHolder topicsMappingHolder, String destinationName); @Override List<SourceRecord> poll(); @Override void commitIfNecessary(); @Override void close(); @Override void onMessage(Message message);  }
@Test public void testPoll() throws Exception { JmsSinkConnector connector = new JmsSinkConnector(); connector.start(testProperties); ApplicationContext context = ApplicationContextHolder.getApplicationContext(); ListeningContainerMessageReceiver messageReceiver = context .getBean(ListeningContainerMessageReceiver.class, context.getBean(ConnectionFactory.class), context.getBean(JmsConnectorConfigurationProperties.class), context.getBean(TopicsMappingHolder.class), "testQueue"); Assert.assertNotNull(messageReceiver.poll()); messageReceiver.commitIfNecessary(); connector.stop(); }