ExecNioSource extends EventSource.Base { static boolean extractLines(ByteBuffer buf, String command, String tag, BlockingQueue<Event> sync) throws InterruptedException { buf.flip(); boolean madeProgress = false; int start = buf.position(); buf.mark(); while (buf.hasRemaining()) { byte b = buf.get(); if (b == '\n') { int end = buf.position(); int sz = end - start - 1; int maxEventSz = (int) FlumeConfiguration.get().getEventMaxSizeBytes(); if (sz > maxEventSz) { byte[] body = new byte[(int) maxEventSz]; buf.reset(); buf.get(body, 0, maxEventSz); buf.position(end); buf.mark(); start = buf.position(); sync.put(buildExecEvent(body, tag, command)); madeProgress = true; } else { byte[] body = new byte[sz]; buf.reset(); buf.get(body, 0, sz); buf.get(); buf.mark(); start = buf.position(); sync.put(buildExecEvent(body, tag, command)); madeProgress = true; } } } buf.reset(); buf.compact(); return madeProgress; } 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 testExtractLines() throws InterruptedException { LinkedBlockingQueue<Event> q = new LinkedBlockingQueue<Event>(); long maxSz = FlumeConfiguration.get().getEventMaxSizeBytes(); ByteBuffer buf = ByteBuffer.allocate((int) maxSz * 4); boolean ret = ExecNioSource.extractLines(buf, "cmd", "tag", q); assertFalse(ret); assertEquals(0, q.size()); for (int i = 0; i < maxSz; i++) { buf.put((byte) 'a'); } ret = ExecNioSource.extractLines(buf, "cmd", "tag", q); assertFalse(ret); assertEquals(maxSz, buf.position()); assertEquals(0, q.size()); buf.put((byte) '\n'); ret = ExecNioSource.extractLines(buf, "cmd", "tag", q); assertTrue(ret); assertEquals(0, buf.position()); assertEquals(1, q.size()); Event e = q.poll(); assertEquals(maxSz, e.getBody().length); for (int i = 0; i < maxSz; i++) { if (i % 1024 == 1023) { buf.put((byte) '\n'); } else { buf.put((byte) 'b'); } } ret = ExecNioSource.extractLines(buf, "cmd", "tag", q); assertTrue(ret); assertEquals(0, buf.position()); assertEquals(32, q.size()); for (int i = 0; i < 32; i++) { e = q.poll(); assertEquals(1023, e.getBody().length); } }