LogReader { void writeLogs(OutputStream out, Instant from, Instant to, Optional<String> hostname) { double fromSeconds = from.getEpochSecond() + from.getNano() / 1e9; double toSeconds = to.getEpochSecond() + to.getNano() / 1e9; BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(out)); for (List<Path> logs : getMatchingFiles(from, to)) { List<LogLineIterator> logLineIterators = new ArrayList<>(); try { for (Path log : logs) logLineIterators.add(new LogLineIterator(log, fromSeconds, toSeconds, hostname)); Iterator<LineWithTimestamp> lines = Iterators.mergeSorted(logLineIterators, Comparator.comparingDouble(LineWithTimestamp::timestamp)); long linesWritten = 0; while (lines.hasNext()) { writer.write(lines.next().line()); writer.newLine(); if ((++linesWritten & ((1 << 16) - 1)) == 0) writer.flush(); } } catch (IOException e) { throw new UncheckedIOException(e); } finally { for (LogLineIterator ll : logLineIterators) { try { ll.close(); } catch (IOException ignored) { } } Exceptions.uncheck(writer::flush); } } } LogReader(String logDirectory, String logFilePattern);   LogReader(Path logDirectory, Pattern logFilePattern);   }
@Test public void testThatLogsOutsideRangeAreExcluded() { ByteArrayOutputStream baos = new ByteArrayOutputStream(); LogReader logReader = new LogReader(logDirectory, Pattern.compile(".*")); logReader.writeLogs(baos, Instant.ofEpochMilli(150), Instant.ofEpochMilli(3601050), Optional.empty()); assertEquals(log100 + logv11 + log110, baos.toString(UTF_8)); }
@Test public void testThatLogsNotMatchingRegexAreExcluded() { ByteArrayOutputStream baos = new ByteArrayOutputStream(); LogReader logReader = new LogReader(logDirectory, Pattern.compile(".*-1.*")); logReader.writeLogs(baos, Instant.EPOCH, Instant.EPOCH.plus(Duration.ofDays(2)), Optional.empty()); assertEquals(log101 + logv11, baos.toString(UTF_8)); }
@Test public void testZippedStreaming() { ByteArrayOutputStream zippedBaos = new ByteArrayOutputStream(); LogReader logReader = new LogReader(logDirectory, Pattern.compile(".*")); logReader.writeLogs(zippedBaos, Instant.EPOCH, Instant.EPOCH.plus(Duration.ofDays(2)), Optional.empty()); assertEquals(log101 + log100 + logv11 + log110 + log200 + logv, zippedBaos.toString(UTF_8)); }
@Test public void logsForSingeNodeIsRetrieved() { ByteArrayOutputStream baos = new ByteArrayOutputStream(); LogReader logReader = new LogReader(logDirectory, Pattern.compile(".*")); logReader.writeLogs(baos, Instant.EPOCH, Instant.EPOCH.plus(Duration.ofDays(2)), Optional.of("node2.com")); assertEquals(log101 + log100 + log200, baos.toString(UTF_8)); }