Plc4XProducer extends DefaultAsyncProducer { @Override public void process(Exchange exchange) throws Exception { Message in = exchange.getIn(); Object body = in.getBody(); PlcWriteRequest.Builder builder = plcConnection.writeRequestBuilder(); if (body instanceof Map) { Map<String, Map<String, Object>> tags = (Map<String, Map<String, Object>>) body; for (Map.Entry<String, Map<String, Object>> entry : tags.entrySet()) { String name = entry.getKey(); String query = entry.getValue().keySet().iterator().next(); Object value = entry.getValue().get(query); builder.addItem(name,query,value); } } else { throw new PlcInvalidFieldException("The body must contain a Map<String,Map<String,Object>"); } CompletableFuture<? extends PlcWriteResponse> completableFuture = builder.build().execute(); int currentlyOpenRequests = openRequests.incrementAndGet(); try { log.debug("Currently open requests including {}:{}", exchange, currentlyOpenRequests); Object plcWriteResponse = completableFuture.get(); if (exchange.getPattern().isOutCapable()) { Message out = exchange.getOut(); out.copyFrom(exchange.getIn()); out.setBody(plcWriteResponse); } else { in.setBody(plcWriteResponse); } } finally { int openRequestsAfterFinish = openRequests.decrementAndGet(); log.trace("Open Requests after {}:{}", exchange, openRequestsAfterFinish); } } Plc4XProducer(Plc4XEndpoint endpoint); @Override void process(Exchange exchange); @Override boolean process(Exchange exchange, AsyncCallback callback);  }
@Test public void process() throws Exception { when(testExchange.getPattern()).thenReturn(ExchangePattern.InOnly); SUT.process(testExchange); when(testExchange.getPattern()).thenReturn(ExchangePattern.InOut); SUT.process(testExchange); when(testExchange.getIn().getBody()).thenReturn(2); }
@Test public void process_Async() { SUT.process(testExchange, doneSync -> { }); when(testExchange.getPattern()).thenReturn(ExchangePattern.InOnly); SUT.process(testExchange, doneSync -> { }); when(testExchange.getPattern()).thenReturn(ExchangePattern.InOut); SUT.process(testExchange, doneSync -> { }); }