CommunicationDetailsDeriver extends AbstractProcessor<Trace, CommunicationDetails> { @Override public void initialise(String tenantId, List<Trace> items) throws RetryAttemptException { List<SourceInfo> sourceInfoList = SourceInfoUtil.getSourceInfo(tenantId, items); try { sourceInfoCache.store(tenantId, sourceInfoList); } catch (CacheException e) { throw new RetryAttemptException(e); } } CommunicationDetailsDeriver(); SourceInfoCache getSourceInfoCache(); void setSourceInfoCache(SourceInfoCache sourceInfoCache); @Override void initialise(String tenantId, List<Trace> items); @Override CommunicationDetails processOneToOne(String tenantId, Trace item);  }
@Test public void testInitialise() throws RetryAttemptException { TestSourceInfoCache cache=new TestSourceInfoCache(); CommunicationDetailsDeriver deriver = new CommunicationDetailsDeriver(); deriver.setSourceInfoCache(cache); List<Trace> traces = new ArrayList<Trace>(); Trace trace1 = new Trace(); trace1.setTraceId("trace1"); trace1.setFragmentId(trace1.getTraceId()); trace1.setTimestamp(TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis())); traces.add(trace1); Consumer c1 = new Consumer(); c1.setTimestamp(TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis())); CorrelationIdentifier cid1 = new CorrelationIdentifier(); cid1.setScope(Scope.Interaction); cid1.setValue("cid1"); c1.getCorrelationIds().add(cid1); trace1.getNodes().add(c1); Producer p1 = new Producer(); p1.setTimestamp(TimeUnit.MILLISECONDS.toMicros(System.currentTimeMillis())); CorrelationIdentifier pid1 = new CorrelationIdentifier(); pid1.setScope(Scope.Interaction); pid1.setValue("pid1"); p1.getCorrelationIds().add(pid1); c1.getNodes().add(p1); deriver.initialise(null, traces); assertNotNull(deriver.getSourceInfoCache().get(null, "pid1")); assertNull(deriver.getSourceInfoCache().get(null, "cid1")); }