CompoundLogReader implements LogEntryReader<CompoundLogAccess> { @Override public void readEntries(final Log log, final CompoundLogAccess logAccess, final LogPointer startOffset, final com.logsniffer.reader.LogEntryReader.LogEntryConsumer consumer) throws IOException, FormatException { readEntries(log, logAccess, startOffset, consumer, false); } CompoundLogReader(final List<LogInstance> composedLogs); @Override LinkedHashMap<String, FieldBaseTypes> getFieldTypes(); @Override List<SeverityLevel> getSupportedSeverities(); @Override void readEntries(final Log log, final CompoundLogAccess logAccess, final LogPointer startOffset,
			final com.logsniffer.reader.LogEntryReader.LogEntryConsumer consumer); @Override void readEntriesReverse(final Log log, final CompoundLogAccess logAccess, final LogPointer startOffset,
			final com.logsniffer.reader.LogEntryReader.LogEntryConsumer consumer);  }
@Test public void testErrorInOneSource() throws FormatException, IOException { final List<LogInstance> subLogs = new ArrayList<>(); final CompoundLogReader r = new CompoundLogReader(subLogs); final Log log1 = new ByteArrayLog("log1", new byte[0]); final Log log2 = new ByteArrayLog("log2", new byte[0]); subLogs.add(new LogInstance(1, log1, Mockito.mock(LogRawAccess.class), new DummySubReader(200, 2, 0, 100))); subLogs.add(new LogInstance(2, log2, Mockito.mock(LogRawAccess.class), new DummySubReader(250, 2, 1))); final BufferedConsumer c = new BufferedConsumer(15000); try { r.readEntries(Mockito.mock(Log.class), new CompoundLogAccess(Mockito.mock(Log.class), subLogs), null, c); } catch (final IOException e) { Assert.assertTrue("Actual buffer size: " + c.getBuffer().size(), c.getBuffer().size() < 200); return; } Assert.fail("Exception expected"); }
@Test(timeout = 1000 * 100 * 100) @Repeat(1) public void testLongComposition() throws FormatException, IOException { final long start = System.currentTimeMillis(); final List<LogInstance> subLogs = new ArrayList<>(); final CompoundLogReader r = new CompoundLogReader(subLogs); final Log log1 = new ByteArrayLog("log1", new byte[0]); final Log log2 = new ByteArrayLog("log2", new byte[0]); subLogs.add(new LogInstance(1, log1, Mockito.mock(LogRawAccess.class), new DummySubReader(20000000, 2, 0))); subLogs.add(new LogInstance(2, log2, Mockito.mock(LogRawAccess.class), new DummySubReader(25000000, 2, 1))); final AtomicInteger count = new AtomicInteger(); r.readEntries(Mockito.mock(Log.class), new CompoundLogAccess(Mockito.mock(Log.class), subLogs), null, new LogEntryConsumer() { @Override public boolean consume(final Log log, final LogPointerFactory pointerFactory, final LogEntry entry) throws IOException { count.incrementAndGet(); return true; } }); final long end = System.currentTimeMillis() - start; logger.info("Read composed {} entries in {}ms, throughput: {} entries/s", count.get(), end, Math.round((double) count.get() / (end / 1000))); Assert.assertEquals(45000000, count.get()); }
@Test public void testCorrectComposition() throws FormatException, IOException { final List<LogInstance> subLogs = new ArrayList<>(); final CompoundLogReader r = new CompoundLogReader(subLogs); final Log log1 = new ByteArrayLog("log1", new byte[0]); final Log log2 = new ByteArrayLog("log2", new byte[0]); subLogs.add(new LogInstance(1, log1, Mockito.mock(LogRawAccess.class), new DummySubReader(200, 0.5f, 0))); subLogs.add(new LogInstance(2, log2, Mockito.mock(LogRawAccess.class), new DummySubReader(250, 0.5f, 1))); final BufferedConsumer c = new BufferedConsumer(15000); r.readEntries(Mockito.mock(Log.class), new CompoundLogAccess(Mockito.mock(Log.class), subLogs), null, c); Assert.assertEquals(450, c.getBuffer().size()); for (int i = 0; i < 400; i++) { final LogEntry e = c.getBuffer().get(i); if (i < 2) { Assert.assertEquals("Error at entry " + i, 0, e.getTimeStamp().getTime()); } else { Assert.assertEquals("Error at entry " + i, 1 + (i - 2) / 4, e.getTimeStamp().getTime()); } if ((i % 4 == 2 || i % 4 == 3 || i < 4) && i < 398) { Assert.assertEquals("Error at entry " + i, 1l, e.get(Event.FIELD_SOURCE_ID)); Assert.assertEquals("Error at entry " + i, "log1", e.get(Event.FIELD_LOG_PATH)); } else if (i % 4 == 0 || i % 4 == 1) { Assert.assertEquals("Error at entry " + i, 2l, e.get(Event.FIELD_SOURCE_ID)); Assert.assertEquals("Error at entry " + i, "log2", e.get(Event.FIELD_LOG_PATH)); } } for (int i = 400; i < c.getBuffer().size(); i++) { final LogEntry e = c.getBuffer().get(i); Assert.assertEquals((long) ((i - 400) * 0.5 + 101), e.getTimeStamp().getTime()); Assert.assertEquals("Error at entry " + i, 2l, e.get(Event.FIELD_SOURCE_ID)); Assert.assertEquals("Error at entry " + i, "log2", e.get(Event.FIELD_LOG_PATH)); } }