@Test public void testConsumeWhenNewLineFound() throws IOException, InterruptedException { runner.run(); runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 0); Thread.sleep(1100L); raf.write("Hello, World".getBytes()); runner.run(); runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 0); raf.write("\r\n".getBytes()); runner.run(); runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 1); final TailFileState state = ((TailFile) runner.getProcessor()).getState().get("target/log.txt").getState(); assertNotNull(state); assertEquals("target/log.txt", state.getFilename()); assertTrue(state.getTimestamp() <= System.currentTimeMillis()); assertEquals(14, state.getPosition()); runner.getFlowFilesForRelationship(TailFile.REL_SUCCESS).get(0).assertContentEquals("Hello, World\r\n"); runner.clearTransferState(); raf.write("12345".getBytes()); runner.run(); runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 0); raf.write("\n".getBytes()); runner.run(); runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 1); runner.getFlowFilesForRelationship(TailFile.REL_SUCCESS).get(0).assertContentEquals("12345\n"); runner.clearTransferState(); raf.write("carriage\rreturn\r".getBytes()); runner.run(); runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 1); runner.getFlowFilesForRelationship(TailFile.REL_SUCCESS).get(0).assertContentEquals("carriage\r"); runner.clearTransferState(); raf.write("\r\n".getBytes()); runner.run(); runner.assertAllFlowFilesTransferred(TailFile.REL_SUCCESS, 1); runner.getFlowFilesForRelationship(TailFile.REL_SUCCESS).get(0).assertContentEquals("return\r\r\n"); }
Map<String, TailFileObject> getState() { return states; }
TailFile extends AbstractProcessor { Map<String, TailFileObject> getState() { return states; } }
TailFile extends AbstractProcessor { Map<String, TailFileObject> getState() { return states; }  }
TailFile extends AbstractProcessor { Map<String, TailFileObject> getState() { return states; }  @Override Set<Relationship> getRelationships(); @Override void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue); @OnScheduled void recoverState(final ProcessContext context); @OnStopped void cleanup(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); }
TailFile extends AbstractProcessor { Map<String, TailFileObject> getState() { return states; }  @Override Set<Relationship> getRelationships(); @Override void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue); @OnScheduled void recoverState(final ProcessContext context); @OnStopped void cleanup(); @Override void onTrigger(final ProcessContext context, final ProcessSession session);  }