TraceCompletionInformationProcessor extends
        AbstractProcessor<TraceCompletionInformation, TraceCompletionInformation> { @Override public TraceCompletionInformation processOneToOne(String tenantId, TraceCompletionInformation item) throws RetryAttemptException { if (!item.getCommunications().isEmpty()) { long currentTime = System.currentTimeMillis(); for (int i = 0; i < item.getCommunications().size(); i++) { Communication c = item.getCommunications().get(i); if (c.getExpire() < currentTime) { if (log.isLoggable(Level.FINEST)) { log.finest("Completion info " + item + ": communication expired = " + c); } if (c.isMultipleConsumers()) { for (int j = 0; j < c.getIds().size(); j++) { List<CommunicationDetails> cds = communicationDetailsCache.getById(tenantId, c.getIds().get(j)); if (log.isLoggable(Level.FINEST)) { log.finest("Multiconsumer comms details: id = " + c.getIds().get(j) + " communication details = " + cds); } for (int k = 0; k < cds.size(); k++) { processCommunication(tenantId, item, c, cds.get(k)); } } } item.getCommunications().remove(i); i--; } else if (!c.isMultipleConsumers()) { CommunicationDetails cd = null; for (int j = 0; cd == null && j < c.getIds().size(); j++) { cd = communicationDetailsCache.get(tenantId, c.getIds().get(j)); } if (log.isLoggable(Level.FINEST)) { log.finest("Completion info " + item + ": communication details for communication " + c + " = " + cd); } if (cd != null) { processCommunication(tenantId, item, c, cd); item.getCommunications().remove(i); i--; } } } if (log.isLoggable(Level.FINEST)) { log.finest("Updated completion information = " + item); } return item; } if (log.isLoggable(Level.FINEST)) { log.finest("No communications to be processed for completion information = " + item); } return null; } TraceCompletionInformationProcessor(); @Override long getDeliveryDelay(List<TraceCompletionInformation> results); CommunicationDetailsCache getCommunicationDetailsCache(); void setCommunicationDetailsCache(CommunicationDetailsCache communicationDetailsCache); @Override TraceCompletionInformation processOneToOne(String tenantId,
            TraceCompletionInformation item);  }
@Test public void testProcessSingleNoCommns() { TraceCompletionInformationProcessor processor = new TraceCompletionInformationProcessor(); TraceCompletionInformation info = new TraceCompletionInformation(); TraceCompletionInformation result = null; try { result = processor.processOneToOne(null, info); } catch (Exception e) { fail("Failed: " + e); } assertNull(result); }
@Test public void testProcessSingleExpireComms() { TraceCompletionInformationProcessor processor = new TraceCompletionInformationProcessor(); TraceCompletionInformation info = new TraceCompletionInformation(); TraceCompletionInformation.Communication c1 = new TraceCompletionInformation.Communication(); c1.setExpire(System.currentTimeMillis() - 10); info.getCommunications().add(c1); TraceCompletionInformation result = null; try { result = processor.processOneToOne(null, info); } catch (Exception e) { fail("Failed: " + e); } assertNotNull(result); assertEquals(0, result.getCommunications().size()); }