@Test public void testProcessFileGranularity() throws IOException, MalformedChunkException, XMLStreamException { String basename = "basename"; int chunkNum = 5; int offset = 10001; byte[] badChunk = {8}; RootNodeHandler rootNodeHandler = mock(RootNodeHandler.class); when(rootNodeHandlerFactory.create(out)).thenReturn(rootNodeHandler); ChunkHeader chunkHeader1 = mock(ChunkHeader.class); ChunkHeader chunkHeader2 = mock(ChunkHeader.class); Record record1 = mock(Record.class); Record record2 = mock(Record.class); Record record3 = mock(Record.class); RootNode rootNode1 = mock(RootNode.class); RootNode rootNode2 = mock(RootNode.class); RootNode rootNode3 = mock(RootNode.class); ProcessSession session = mock(ProcessSession.class); FlowFile flowFile = mock(FlowFile.class); AtomicReference<Exception> reference = new AtomicReference<>(); MalformedChunkException malformedChunkException = new MalformedChunkException("Test", null, offset, chunkNum, badChunk); when(record1.getRootNode()).thenReturn(rootNode1); when(record2.getRootNode()).thenReturn(rootNode2); when(record3.getRootNode()).thenReturn(rootNode3); when(fileHeader.hasNext()).thenReturn(true).thenReturn(true).thenReturn(true).thenReturn(false); when(fileHeader.next()).thenThrow(malformedChunkException).thenReturn(chunkHeader1).thenReturn(chunkHeader2).thenReturn(null); when(chunkHeader1.hasNext()).thenReturn(true).thenReturn(false); when(chunkHeader1.next()).thenReturn(record1).thenReturn(null); when(chunkHeader2.hasNext()).thenReturn(true).thenReturn(true).thenReturn(false); when(chunkHeader2.next()).thenReturn(record2).thenReturn(record3).thenReturn(null); parseEvtx.processFileGranularity(session, componentLog, flowFile, basename, reference, in, out); verify(malformedChunkHandler).handle(flowFile, session, parseEvtx.getName(basename, chunkNum, null, ParseEvtx.EVTX_EXTENSION), badChunk); verify(rootNodeHandler).handle(rootNode1); verify(rootNodeHandler).handle(rootNode2); verify(rootNodeHandler).handle(rootNode3); verify(rootNodeHandler).close(); }
protected void processFileGranularity(ProcessSession session, ComponentLog componentLog, FlowFile original, String basename, AtomicReference<Exception> exceptionReference, InputStream in, OutputStream out) throws IOException { FileHeader fileHeader = fileHeaderFactory.create(in, componentLog); try (RootNodeHandler rootNodeHandler = rootNodeHandlerFactory.create(out)) { while (fileHeader.hasNext()) { try { ChunkHeader chunkHeader = fileHeader.next(); try { while (chunkHeader.hasNext()) { rootNodeHandler.handle(chunkHeader.next().getRootNode()); } } catch (IOException e) { malformedChunkHandler.handle(original, session, getName(basename, chunkHeader.getChunkNumber(), null, EVTX_EXTENSION), chunkHeader.getBinaryReader().getBytes()); exceptionReference.set(e); } } catch (MalformedChunkException e) { malformedChunkHandler.handle(original, session, getName(basename, e.getChunkNum(), null, EVTX_EXTENSION), e.getBadChunk()); } } } catch (IOException e) { exceptionReference.set(e); } }
ParseEvtx extends AbstractProcessor { protected void processFileGranularity(ProcessSession session, ComponentLog componentLog, FlowFile original, String basename, AtomicReference<Exception> exceptionReference, InputStream in, OutputStream out) throws IOException { FileHeader fileHeader = fileHeaderFactory.create(in, componentLog); try (RootNodeHandler rootNodeHandler = rootNodeHandlerFactory.create(out)) { while (fileHeader.hasNext()) { try { ChunkHeader chunkHeader = fileHeader.next(); try { while (chunkHeader.hasNext()) { rootNodeHandler.handle(chunkHeader.next().getRootNode()); } } catch (IOException e) { malformedChunkHandler.handle(original, session, getName(basename, chunkHeader.getChunkNumber(), null, EVTX_EXTENSION), chunkHeader.getBinaryReader().getBytes()); exceptionReference.set(e); } } catch (MalformedChunkException e) { malformedChunkHandler.handle(original, session, getName(basename, e.getChunkNum(), null, EVTX_EXTENSION), e.getBadChunk()); } } } catch (IOException e) { exceptionReference.set(e); } } }
ParseEvtx extends AbstractProcessor { protected void processFileGranularity(ProcessSession session, ComponentLog componentLog, FlowFile original, String basename, AtomicReference<Exception> exceptionReference, InputStream in, OutputStream out) throws IOException { FileHeader fileHeader = fileHeaderFactory.create(in, componentLog); try (RootNodeHandler rootNodeHandler = rootNodeHandlerFactory.create(out)) { while (fileHeader.hasNext()) { try { ChunkHeader chunkHeader = fileHeader.next(); try { while (chunkHeader.hasNext()) { rootNodeHandler.handle(chunkHeader.next().getRootNode()); } } catch (IOException e) { malformedChunkHandler.handle(original, session, getName(basename, chunkHeader.getChunkNumber(), null, EVTX_EXTENSION), chunkHeader.getBinaryReader().getBytes()); exceptionReference.set(e); } } catch (MalformedChunkException e) { malformedChunkHandler.handle(original, session, getName(basename, e.getChunkNum(), null, EVTX_EXTENSION), e.getBadChunk()); } } } catch (IOException e) { exceptionReference.set(e); } } ParseEvtx();  ParseEvtx(FileHeaderFactory fileHeaderFactory, MalformedChunkHandler malformedChunkHandler, RootNodeHandlerFactory rootNodeHandlerFactory, ResultProcessor resultProcessor); }
ParseEvtx extends AbstractProcessor { protected void processFileGranularity(ProcessSession session, ComponentLog componentLog, FlowFile original, String basename, AtomicReference<Exception> exceptionReference, InputStream in, OutputStream out) throws IOException { FileHeader fileHeader = fileHeaderFactory.create(in, componentLog); try (RootNodeHandler rootNodeHandler = rootNodeHandlerFactory.create(out)) { while (fileHeader.hasNext()) { try { ChunkHeader chunkHeader = fileHeader.next(); try { while (chunkHeader.hasNext()) { rootNodeHandler.handle(chunkHeader.next().getRootNode()); } } catch (IOException e) { malformedChunkHandler.handle(original, session, getName(basename, chunkHeader.getChunkNumber(), null, EVTX_EXTENSION), chunkHeader.getBinaryReader().getBytes()); exceptionReference.set(e); } } catch (MalformedChunkException e) { malformedChunkHandler.handle(original, session, getName(basename, e.getChunkNum(), null, EVTX_EXTENSION), e.getBadChunk()); } } } catch (IOException e) { exceptionReference.set(e); } } ParseEvtx();  ParseEvtx(FileHeaderFactory fileHeaderFactory, MalformedChunkHandler malformedChunkHandler, RootNodeHandlerFactory rootNodeHandlerFactory, ResultProcessor resultProcessor); @Override void onTrigger(ProcessContext context, ProcessSession session); @Override Set<Relationship> getRelationships(); }
ParseEvtx extends AbstractProcessor { protected void processFileGranularity(ProcessSession session, ComponentLog componentLog, FlowFile original, String basename, AtomicReference<Exception> exceptionReference, InputStream in, OutputStream out) throws IOException { FileHeader fileHeader = fileHeaderFactory.create(in, componentLog); try (RootNodeHandler rootNodeHandler = rootNodeHandlerFactory.create(out)) { while (fileHeader.hasNext()) { try { ChunkHeader chunkHeader = fileHeader.next(); try { while (chunkHeader.hasNext()) { rootNodeHandler.handle(chunkHeader.next().getRootNode()); } } catch (IOException e) { malformedChunkHandler.handle(original, session, getName(basename, chunkHeader.getChunkNumber(), null, EVTX_EXTENSION), chunkHeader.getBinaryReader().getBytes()); exceptionReference.set(e); } } catch (MalformedChunkException e) { malformedChunkHandler.handle(original, session, getName(basename, e.getChunkNum(), null, EVTX_EXTENSION), e.getBadChunk()); } } } catch (IOException e) { exceptionReference.set(e); } } ParseEvtx();  ParseEvtx(FileHeaderFactory fileHeaderFactory, MalformedChunkHandler malformedChunkHandler, RootNodeHandlerFactory rootNodeHandlerFactory, ResultProcessor resultProcessor); @Override void onTrigger(ProcessContext context, ProcessSession session); @Override Set<Relationship> getRelationships(); static final String RECORD; static final String CHUNK; static final String FILE; static final String EVTX_EXTENSION; static final String XML_EXTENSION; }