@Test public void testProcess() throws Exception { init(); Exchange exchange = createFileEntryExchange(); Resource xsd = Mockito.mock(Resource.class); Map<String, Resource> xsdList = new HashMap<String, Resource>(); FileEntryWorkNote workNote = (FileEntryWorkNote) exchange.getIn().getMandatoryBody(WorkNote.class); xsdList.put(workNote.getFileEntry().getFileType().getName(), xsd); xsdSelector.setXsdList(xsdList); processor.process(exchange); Mockito.verify(processor, Mockito.times(1)).parse(Mockito.any(InputStream.class), Mockito.any(Resource.class), Mockito.any(SimpleReportStats.class), Mockito.any(JobSource.class)); }
@Override protected void process(Exchange exchange, ProcessorArgs<FileEntryWorkNote> args) { prepareState(exchange, args.workNote); Source source = new FileSource(args.workNote.getFileEntry().getResourceId()); Metrics metrics = Metrics.newInstance(args.workNote.getFileEntry().getResourceId()); InputStream input = null; boolean validData = true; try { input = args.workNote.getFileEntry().getFileStream(); Resource xsdSchema = xsdSelector.provideXsdResource(args.workNote.getFileEntry()); parse(input, xsdSchema, args.reportStats, source); metrics.setValidationErrorCount(ignoredRecordCount.get()); } catch (IOException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0061); } catch (SAXException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0062); } catch (XmlParseException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0063); validData = false; } finally { IOUtils.closeQuietly(input); if (validData) { sendDataBatch(); } cleanUpState(); args.job = refreshjob(args.job.getId()); args.stage.addMetrics(metrics); } }
EdFiParserProcessor extends IngestionProcessor<FileEntryWorkNote, IngestionFileEntry> implements
        RecordVisitor { @Override protected void process(Exchange exchange, ProcessorArgs<FileEntryWorkNote> args) { prepareState(exchange, args.workNote); Source source = new FileSource(args.workNote.getFileEntry().getResourceId()); Metrics metrics = Metrics.newInstance(args.workNote.getFileEntry().getResourceId()); InputStream input = null; boolean validData = true; try { input = args.workNote.getFileEntry().getFileStream(); Resource xsdSchema = xsdSelector.provideXsdResource(args.workNote.getFileEntry()); parse(input, xsdSchema, args.reportStats, source); metrics.setValidationErrorCount(ignoredRecordCount.get()); } catch (IOException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0061); } catch (SAXException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0062); } catch (XmlParseException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0063); validData = false; } finally { IOUtils.closeQuietly(input); if (validData) { sendDataBatch(); } cleanUpState(); args.job = refreshjob(args.job.getId()); args.stage.addMetrics(metrics); } } }
EdFiParserProcessor extends IngestionProcessor<FileEntryWorkNote, IngestionFileEntry> implements
        RecordVisitor { @Override protected void process(Exchange exchange, ProcessorArgs<FileEntryWorkNote> args) { prepareState(exchange, args.workNote); Source source = new FileSource(args.workNote.getFileEntry().getResourceId()); Metrics metrics = Metrics.newInstance(args.workNote.getFileEntry().getResourceId()); InputStream input = null; boolean validData = true; try { input = args.workNote.getFileEntry().getFileStream(); Resource xsdSchema = xsdSelector.provideXsdResource(args.workNote.getFileEntry()); parse(input, xsdSchema, args.reportStats, source); metrics.setValidationErrorCount(ignoredRecordCount.get()); } catch (IOException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0061); } catch (SAXException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0062); } catch (XmlParseException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0063); validData = false; } finally { IOUtils.closeQuietly(input); if (validData) { sendDataBatch(); } cleanUpState(); args.job = refreshjob(args.job.getId()); args.stage.addMetrics(metrics); } }  }
EdFiParserProcessor extends IngestionProcessor<FileEntryWorkNote, IngestionFileEntry> implements
        RecordVisitor { @Override protected void process(Exchange exchange, ProcessorArgs<FileEntryWorkNote> args) { prepareState(exchange, args.workNote); Source source = new FileSource(args.workNote.getFileEntry().getResourceId()); Metrics metrics = Metrics.newInstance(args.workNote.getFileEntry().getResourceId()); InputStream input = null; boolean validData = true; try { input = args.workNote.getFileEntry().getFileStream(); Resource xsdSchema = xsdSelector.provideXsdResource(args.workNote.getFileEntry()); parse(input, xsdSchema, args.reportStats, source); metrics.setValidationErrorCount(ignoredRecordCount.get()); } catch (IOException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0061); } catch (SAXException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0062); } catch (XmlParseException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0063); validData = false; } finally { IOUtils.closeQuietly(input); if (validData) { sendDataBatch(); } cleanUpState(); args.job = refreshjob(args.job.getId()); args.stage.addMetrics(metrics); } }  @Override void visit(RecordMeta recordMeta, Map<String, Object> record); @Override void ignored(); void sendDataBatch(); @Override AbstractMessageReport getMessageReport(); @Override void setMessageReport(AbstractMessageReport messageReport); @Override String getStageDescription(); @Override BatchJobStageType getStage(); ProducerTemplate getProducer(); void setProducer(ProducerTemplate producer); TypeProvider getTypeProvider(); void setTypeProvider(TypeProvider typeProvider); XsdSelector getXsdSelector(); void setHelper( ReferenceHelper helper ); void setXsdSelector(XsdSelector xsdSelector); int getBatchSize(); void setBatchSize(int batchSize); void audit(SecurityEvent event); }
EdFiParserProcessor extends IngestionProcessor<FileEntryWorkNote, IngestionFileEntry> implements
        RecordVisitor { @Override protected void process(Exchange exchange, ProcessorArgs<FileEntryWorkNote> args) { prepareState(exchange, args.workNote); Source source = new FileSource(args.workNote.getFileEntry().getResourceId()); Metrics metrics = Metrics.newInstance(args.workNote.getFileEntry().getResourceId()); InputStream input = null; boolean validData = true; try { input = args.workNote.getFileEntry().getFileStream(); Resource xsdSchema = xsdSelector.provideXsdResource(args.workNote.getFileEntry()); parse(input, xsdSchema, args.reportStats, source); metrics.setValidationErrorCount(ignoredRecordCount.get()); } catch (IOException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0061); } catch (SAXException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0062); } catch (XmlParseException e) { getMessageReport().error(args.reportStats, source, CoreMessageCode.CORE_0063); validData = false; } finally { IOUtils.closeQuietly(input); if (validData) { sendDataBatch(); } cleanUpState(); args.job = refreshjob(args.job.getId()); args.stage.addMetrics(metrics); } }  @Override void visit(RecordMeta recordMeta, Map<String, Object> record); @Override void ignored(); void sendDataBatch(); @Override AbstractMessageReport getMessageReport(); @Override void setMessageReport(AbstractMessageReport messageReport); @Override String getStageDescription(); @Override BatchJobStageType getStage(); ProducerTemplate getProducer(); void setProducer(ProducerTemplate producer); TypeProvider getTypeProvider(); void setTypeProvider(TypeProvider typeProvider); XsdSelector getXsdSelector(); void setHelper( ReferenceHelper helper ); void setXsdSelector(XsdSelector xsdSelector); int getBatchSize(); void setBatchSize(int batchSize); void audit(SecurityEvent event);  }