@Test public void testPrevIngested() throws Exception{ DidSchemaParser didSchemaParser = Mockito.mock(DidSchemaParser.class); Mockito.when(dIdResolver.getDidSchemaParser()).thenReturn(didSchemaParser); Map<String, List<DidNaturalKey>> naturalKeys = new HashMap<String, List<DidNaturalKey>>(); naturalKeys.put("student", new ArrayList<DidNaturalKey>()); Mockito.when(didSchemaParser.getNaturalKeys()).thenReturn(naturalKeys); Mockito.when(dIdStrategy.generateId(Mockito.any(NaturalKeyDescriptor.class))).thenReturn("recordId"); NeutralRecord nr = createNeutralRecord("student"); List<NeutralRecord> neutralRecords = new ArrayList<NeutralRecord>(); neutralRecords.add(nr); RecordHash recordHash = Mockito.mock(RecordHash.class); String recordHashValues = nr.generateRecordHash("tenantId"); Mockito.when(recordHash.getHash()).thenReturn(recordHashValues); Mockito.when(batchJobDAO.findRecordHash("tenantId", "recordId")).thenReturn(recordHash); Exchange exchange = new DefaultExchange(new DefaultCamelContext()); NeutralRecordWorkNote workNote = new NeutralRecordWorkNote(neutralRecords, "batchJobId", false); exchange.getIn().setBody(workNote); deltaProcessor.process(exchange); NeutralRecordWorkNote newWorkNote = exchange.getIn().getBody(NeutralRecordWorkNote.class); List<NeutralRecord> filteredRecords = newWorkNote.getNeutralRecords(); Assert.assertNotNull(filteredRecords); Assert.assertEquals(0, filteredRecords.size()); }
@Override public void process(Exchange exchange, ProcessorArgs<NeutralRecordWorkNote> args) { List<NeutralRecord> filteredRecords = null; List<NeutralRecord> records = args.workNote.getNeutralRecords(); if (records != null && records.size() > 0) { Metrics metrics = Metrics.newInstance(records.get(0).getSourceFile()); args.stage.addMetrics(metrics); if (!isDeltaDisabled()) { filteredRecords = filterRecords(metrics, args.workNote, args.reportStats); args.workNote.setNeutralRecords(filteredRecords); exchange.getIn().setBody(args.workNote); } exchange.getIn().setHeader(IngestionRouteBuilder.INGESTION_MESSAGE_TYPE, MessageType.PERSIST_REQUEST.name()); } }
DeltaProcessor extends IngestionProcessor<NeutralRecordWorkNote, Resource> { @Override public void process(Exchange exchange, ProcessorArgs<NeutralRecordWorkNote> args) { List<NeutralRecord> filteredRecords = null; List<NeutralRecord> records = args.workNote.getNeutralRecords(); if (records != null && records.size() > 0) { Metrics metrics = Metrics.newInstance(records.get(0).getSourceFile()); args.stage.addMetrics(metrics); if (!isDeltaDisabled()) { filteredRecords = filterRecords(metrics, args.workNote, args.reportStats); args.workNote.setNeutralRecords(filteredRecords); exchange.getIn().setBody(args.workNote); } exchange.getIn().setHeader(IngestionRouteBuilder.INGESTION_MESSAGE_TYPE, MessageType.PERSIST_REQUEST.name()); } } }
DeltaProcessor extends IngestionProcessor<NeutralRecordWorkNote, Resource> { @Override public void process(Exchange exchange, ProcessorArgs<NeutralRecordWorkNote> args) { List<NeutralRecord> filteredRecords = null; List<NeutralRecord> records = args.workNote.getNeutralRecords(); if (records != null && records.size() > 0) { Metrics metrics = Metrics.newInstance(records.get(0).getSourceFile()); args.stage.addMetrics(metrics); if (!isDeltaDisabled()) { filteredRecords = filterRecords(metrics, args.workNote, args.reportStats); args.workNote.setNeutralRecords(filteredRecords); exchange.getIn().setBody(args.workNote); } exchange.getIn().setHeader(IngestionRouteBuilder.INGESTION_MESSAGE_TYPE, MessageType.PERSIST_REQUEST.name()); } }  }
DeltaProcessor extends IngestionProcessor<NeutralRecordWorkNote, Resource> { @Override public void process(Exchange exchange, ProcessorArgs<NeutralRecordWorkNote> args) { List<NeutralRecord> filteredRecords = null; List<NeutralRecord> records = args.workNote.getNeutralRecords(); if (records != null && records.size() > 0) { Metrics metrics = Metrics.newInstance(records.get(0).getSourceFile()); args.stage.addMetrics(metrics); if (!isDeltaDisabled()) { filteredRecords = filterRecords(metrics, args.workNote, args.reportStats); args.workNote.setNeutralRecords(filteredRecords); exchange.getIn().setBody(args.workNote); } exchange.getIn().setHeader(IngestionRouteBuilder.INGESTION_MESSAGE_TYPE, MessageType.PERSIST_REQUEST.name()); } }  @Override void process(Exchange exchange, ProcessorArgs<NeutralRecordWorkNote> args); void setdIdStrategy(DeterministicUUIDGeneratorStrategy dIdStrategy); void setdIdResolver(DeterministicIdResolver dIdResolver); void setRecordLevelDeltaEnabledEntities(Set<String> recordLevelDeltaEnabledEntities); }
DeltaProcessor extends IngestionProcessor<NeutralRecordWorkNote, Resource> { @Override public void process(Exchange exchange, ProcessorArgs<NeutralRecordWorkNote> args) { List<NeutralRecord> filteredRecords = null; List<NeutralRecord> records = args.workNote.getNeutralRecords(); if (records != null && records.size() > 0) { Metrics metrics = Metrics.newInstance(records.get(0).getSourceFile()); args.stage.addMetrics(metrics); if (!isDeltaDisabled()) { filteredRecords = filterRecords(metrics, args.workNote, args.reportStats); args.workNote.setNeutralRecords(filteredRecords); exchange.getIn().setBody(args.workNote); } exchange.getIn().setHeader(IngestionRouteBuilder.INGESTION_MESSAGE_TYPE, MessageType.PERSIST_REQUEST.name()); } }  @Override void process(Exchange exchange, ProcessorArgs<NeutralRecordWorkNote> args); void setdIdStrategy(DeterministicUUIDGeneratorStrategy dIdStrategy); void setdIdResolver(DeterministicIdResolver dIdResolver); void setRecordLevelDeltaEnabledEntities(Set<String> recordLevelDeltaEnabledEntities);  }