Plc4XConsumer extends DefaultConsumer { @Override protected void doStart() throws InterruptedException, ExecutionException { if(trigger==null) { PlcReadRequest.Builder builder = plcConnection.readRequestBuilder(); for( String tag : tags.keySet()){ try{ String query = (String)tags.get(tag); builder.addItem(tag,query); } catch (PlcIncompatibleDatatypeException e){ LOGGER.error("For consumer, please use Map<String,String>, currently using {}",tags.getClass().getSimpleName()); } } PlcReadRequest request = builder.build(); future = executorService.schedule(() -> { request.execute().thenAccept(response -> { try { Exchange exchange = plc4XEndpoint.createExchange(); Map<String,Object> rsp = new HashMap<>(); for(String field : response.getFieldNames()){ rsp.put(field,response.getObject(field)); } exchange.getIn().setBody(rsp); getProcessor().process(exchange); } catch (Exception e) { exceptionHandler.handleException(e); } }); }, 500, TimeUnit.MILLISECONDS); } else{ try { ScraperConfiguration configuration = getScraperConfig(validateTags()); TriggerCollector collector = new TriggerCollectorImpl(plc4XEndpoint.getPlcDriverManager()); TriggeredScraperImpl scraper = new TriggeredScraperImpl(configuration, (job, alias, response) -> { try { Exchange exchange = plc4XEndpoint.createExchange(); exchange.getIn().setBody(response); getProcessor().process(exchange); } catch (Exception e) { exceptionHandler.handleException(e); }; },collector); scraper.start(); collector.start(); } catch (ScraperException e) { e.printStackTrace(); } } } Plc4XConsumer(Plc4XEndpoint endpoint, Processor processor); @Override String toString(); @Override Endpoint getEndpoint(); ExceptionHandler getExceptionHandler(); void setExceptionHandler(ExceptionHandler exceptionHandler);  }
@Test public void doStart() { }