ExecNioSource extends EventSource.Base { static boolean dropUntilNewLine(ByteBuffer in) { in.flip(); while (in.hasRemaining()) { if (in.get() == '\n') { in.compact(); return false; } } in.clear(); return true; } ExecNioSource(String command, boolean aggregate, boolean restart, int period); void close(); Event next(); void open(); static SourceBuilder buildPeriodic(); static SourceBuilder buildStream(); static SourceBuilder builder(); static final String A_PROC_SOURCE; static final String A_EXEC_CMD; }
@Test public void testDropUntilNewLine() { long maxSz = FlumeConfiguration.get().getEventMaxSizeBytes(); ByteBuffer buf = ByteBuffer.allocate((int) maxSz * 4); boolean ret = ExecNioSource.dropUntilNewLine(buf); assertTrue(ret); assertEquals(0, buf.position()); for (int i = 0; i < maxSz; i++) { buf.put((byte) 'a'); } ret = ExecNioSource.dropUntilNewLine(buf); assertTrue(ret); assertEquals(0, buf.position()); buf.put((byte) '\n'); buf.put((byte) 'd'); ret = ExecNioSource.dropUntilNewLine(buf); assertFalse(ret); assertEquals(1, buf.position()); buf.clear(); for (int i = 0; i < maxSz; i++) { if (i % 1024 == 1023) { buf.put((byte) '\n'); } else { buf.put((byte) 'b'); } } ret = ExecNioSource.dropUntilNewLine(buf); assertFalse(ret); assertEquals(maxSz - 1024, buf.position()); }