@Test public void testReadLogMessageAndPosition() throws Exception { String path = FilenameUtils.concat(getTempPath(), "test2.log"); List<String> dataWritten = generateSampleMessagesToFile(path); long inode = SingerUtils.getFileInode(SingerUtils.getPath(path)); LogFile logFile = new LogFile(inode); LogFileReader reader = new TextLogFileReader(logFile, path, 0, 8192, 102400, 1, Pattern.compile("^.*$"), TextLogMessageType.PLAIN_TEXT, false, false, true, null, null, null); for (int i = 0; i < 100; i++) { LogMessageAndPosition log = reader.readLogMessageAndPosition(); assertEquals(dataWritten.get(i).trim(), new String(log.getLogMessage().getMessage())); } reader.close(); }
@Override public LogMessageAndPosition readLogMessageAndPosition() throws LogFileReaderException { if (closed) { throw new LogFileReaderException("Reader for " + path + " is closed."); } try { TextMessageReader.resetByteBuffer(maxBuffer); for (int i = 0; i < numMessagesPerLogMessage; ++i) { ByteBuffer message = textMessageReader.readMessage(true); if (message == null) { break; } String prependStr = ""; if (prependTimestamp) { prependStr += System.currentTimeMillis() + prependFieldDelimiter; } if (prependHostname) { prependStr += hostname + prependFieldDelimiter; } if (prependEnvironmentVariables != null) { prependStr += prependEnvironmentVariables; } if (prependStr.length() > 0) { maxBuffer.put(prependStr.getBytes()); } maxBuffer.put(message); } if (maxBuffer.position() <= 0) { return null; } maxBuffer.flip(); ByteBuffer out = ByteBuffer.allocate(maxBuffer.limit()).put(maxBuffer); out.flip(); if (trimTailingNewlineCharacter && out.get(out.limit() - 1) == '\n') { out.limit(out.limit() - 1); } LogMessage logMessage = null; switch (textLogMessageType) { case THRIFT_TEXT_MESSAGE: TextMessage textMessage = new TextMessage(); textMessage.setFilename(path); textMessage.setHost(hostname); textMessage.addToMessages(TextMessageReader.bufToString(out)); logMessage = new LogMessage(ByteBuffer.wrap(serializer.serialize(textMessage))); break; case PLAIN_TEXT: logMessage = new LogMessage(out); break; default: throw new UnsupportedOperationException( "Unknown text log message type:" + textLogMessageType.name()); } LogPosition position = new LogPosition(logFile, textMessageReader.getByteOffset()); return new LogMessageAndPosition(logMessage, position); } catch (Exception e) { LOG.error("Caught exception when read a log message from log file: " + logFile, e); throw new LogFileReaderException("Cannot read a log message.", e); } }
TextLogFileReader implements LogFileReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogFileReaderException { if (closed) { throw new LogFileReaderException("Reader for " + path + " is closed."); } try { TextMessageReader.resetByteBuffer(maxBuffer); for (int i = 0; i < numMessagesPerLogMessage; ++i) { ByteBuffer message = textMessageReader.readMessage(true); if (message == null) { break; } String prependStr = ""; if (prependTimestamp) { prependStr += System.currentTimeMillis() + prependFieldDelimiter; } if (prependHostname) { prependStr += hostname + prependFieldDelimiter; } if (prependEnvironmentVariables != null) { prependStr += prependEnvironmentVariables; } if (prependStr.length() > 0) { maxBuffer.put(prependStr.getBytes()); } maxBuffer.put(message); } if (maxBuffer.position() <= 0) { return null; } maxBuffer.flip(); ByteBuffer out = ByteBuffer.allocate(maxBuffer.limit()).put(maxBuffer); out.flip(); if (trimTailingNewlineCharacter && out.get(out.limit() - 1) == '\n') { out.limit(out.limit() - 1); } LogMessage logMessage = null; switch (textLogMessageType) { case THRIFT_TEXT_MESSAGE: TextMessage textMessage = new TextMessage(); textMessage.setFilename(path); textMessage.setHost(hostname); textMessage.addToMessages(TextMessageReader.bufToString(out)); logMessage = new LogMessage(ByteBuffer.wrap(serializer.serialize(textMessage))); break; case PLAIN_TEXT: logMessage = new LogMessage(out); break; default: throw new UnsupportedOperationException( "Unknown text log message type:" + textLogMessageType.name()); } LogPosition position = new LogPosition(logFile, textMessageReader.getByteOffset()); return new LogMessageAndPosition(logMessage, position); } catch (Exception e) { LOG.error("Caught exception when read a log message from log file: " + logFile, e); throw new LogFileReaderException("Cannot read a log message.", e); } } }
TextLogFileReader implements LogFileReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogFileReaderException { if (closed) { throw new LogFileReaderException("Reader for " + path + " is closed."); } try { TextMessageReader.resetByteBuffer(maxBuffer); for (int i = 0; i < numMessagesPerLogMessage; ++i) { ByteBuffer message = textMessageReader.readMessage(true); if (message == null) { break; } String prependStr = ""; if (prependTimestamp) { prependStr += System.currentTimeMillis() + prependFieldDelimiter; } if (prependHostname) { prependStr += hostname + prependFieldDelimiter; } if (prependEnvironmentVariables != null) { prependStr += prependEnvironmentVariables; } if (prependStr.length() > 0) { maxBuffer.put(prependStr.getBytes()); } maxBuffer.put(message); } if (maxBuffer.position() <= 0) { return null; } maxBuffer.flip(); ByteBuffer out = ByteBuffer.allocate(maxBuffer.limit()).put(maxBuffer); out.flip(); if (trimTailingNewlineCharacter && out.get(out.limit() - 1) == '\n') { out.limit(out.limit() - 1); } LogMessage logMessage = null; switch (textLogMessageType) { case THRIFT_TEXT_MESSAGE: TextMessage textMessage = new TextMessage(); textMessage.setFilename(path); textMessage.setHost(hostname); textMessage.addToMessages(TextMessageReader.bufToString(out)); logMessage = new LogMessage(ByteBuffer.wrap(serializer.serialize(textMessage))); break; case PLAIN_TEXT: logMessage = new LogMessage(out); break; default: throw new UnsupportedOperationException( "Unknown text log message type:" + textLogMessageType.name()); } LogPosition position = new LogPosition(logFile, textMessageReader.getByteOffset()); return new LogMessageAndPosition(logMessage, position); } catch (Exception e) { LOG.error("Caught exception when read a log message from log file: " + logFile, e); throw new LogFileReaderException("Cannot read a log message.", e); } } TextLogFileReader(LogFile logFile,
                           String path,
                           long byteOffset,
                           int readBufferSize,
                           int maxMessageSize,
                           int numMessagesPerLogMessage,
                           Pattern messageStartPattern,
                           TextLogMessageType messageType,
                           boolean prependTimestamp,
                           boolean prependHostName,
                           boolean trimTailingNewlineCharacter,
                           String hostname,
                           String prependFieldDelimiter,
                           String prependEnvironmentVariables); }
TextLogFileReader implements LogFileReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogFileReaderException { if (closed) { throw new LogFileReaderException("Reader for " + path + " is closed."); } try { TextMessageReader.resetByteBuffer(maxBuffer); for (int i = 0; i < numMessagesPerLogMessage; ++i) { ByteBuffer message = textMessageReader.readMessage(true); if (message == null) { break; } String prependStr = ""; if (prependTimestamp) { prependStr += System.currentTimeMillis() + prependFieldDelimiter; } if (prependHostname) { prependStr += hostname + prependFieldDelimiter; } if (prependEnvironmentVariables != null) { prependStr += prependEnvironmentVariables; } if (prependStr.length() > 0) { maxBuffer.put(prependStr.getBytes()); } maxBuffer.put(message); } if (maxBuffer.position() <= 0) { return null; } maxBuffer.flip(); ByteBuffer out = ByteBuffer.allocate(maxBuffer.limit()).put(maxBuffer); out.flip(); if (trimTailingNewlineCharacter && out.get(out.limit() - 1) == '\n') { out.limit(out.limit() - 1); } LogMessage logMessage = null; switch (textLogMessageType) { case THRIFT_TEXT_MESSAGE: TextMessage textMessage = new TextMessage(); textMessage.setFilename(path); textMessage.setHost(hostname); textMessage.addToMessages(TextMessageReader.bufToString(out)); logMessage = new LogMessage(ByteBuffer.wrap(serializer.serialize(textMessage))); break; case PLAIN_TEXT: logMessage = new LogMessage(out); break; default: throw new UnsupportedOperationException( "Unknown text log message type:" + textLogMessageType.name()); } LogPosition position = new LogPosition(logFile, textMessageReader.getByteOffset()); return new LogMessageAndPosition(logMessage, position); } catch (Exception e) { LOG.error("Caught exception when read a log message from log file: " + logFile, e); throw new LogFileReaderException("Cannot read a log message.", e); } } TextLogFileReader(LogFile logFile,
                           String path,
                           long byteOffset,
                           int readBufferSize,
                           int maxMessageSize,
                           int numMessagesPerLogMessage,
                           Pattern messageStartPattern,
                           TextLogMessageType messageType,
                           boolean prependTimestamp,
                           boolean prependHostName,
                           boolean trimTailingNewlineCharacter,
                           String hostname,
                           String prependFieldDelimiter,
                           String prependEnvironmentVariables); @Override LogMessageAndPosition readLogMessageAndPosition(); void close(); @Override LogFile getLogFile(); @Override long getByteOffset(); @Override void setByteOffset(long byteOffset); @Override boolean isClosed(); }
TextLogFileReader implements LogFileReader { @Override public LogMessageAndPosition readLogMessageAndPosition() throws LogFileReaderException { if (closed) { throw new LogFileReaderException("Reader for " + path + " is closed."); } try { TextMessageReader.resetByteBuffer(maxBuffer); for (int i = 0; i < numMessagesPerLogMessage; ++i) { ByteBuffer message = textMessageReader.readMessage(true); if (message == null) { break; } String prependStr = ""; if (prependTimestamp) { prependStr += System.currentTimeMillis() + prependFieldDelimiter; } if (prependHostname) { prependStr += hostname + prependFieldDelimiter; } if (prependEnvironmentVariables != null) { prependStr += prependEnvironmentVariables; } if (prependStr.length() > 0) { maxBuffer.put(prependStr.getBytes()); } maxBuffer.put(message); } if (maxBuffer.position() <= 0) { return null; } maxBuffer.flip(); ByteBuffer out = ByteBuffer.allocate(maxBuffer.limit()).put(maxBuffer); out.flip(); if (trimTailingNewlineCharacter && out.get(out.limit() - 1) == '\n') { out.limit(out.limit() - 1); } LogMessage logMessage = null; switch (textLogMessageType) { case THRIFT_TEXT_MESSAGE: TextMessage textMessage = new TextMessage(); textMessage.setFilename(path); textMessage.setHost(hostname); textMessage.addToMessages(TextMessageReader.bufToString(out)); logMessage = new LogMessage(ByteBuffer.wrap(serializer.serialize(textMessage))); break; case PLAIN_TEXT: logMessage = new LogMessage(out); break; default: throw new UnsupportedOperationException( "Unknown text log message type:" + textLogMessageType.name()); } LogPosition position = new LogPosition(logFile, textMessageReader.getByteOffset()); return new LogMessageAndPosition(logMessage, position); } catch (Exception e) { LOG.error("Caught exception when read a log message from log file: " + logFile, e); throw new LogFileReaderException("Cannot read a log message.", e); } } TextLogFileReader(LogFile logFile,
                           String path,
                           long byteOffset,
                           int readBufferSize,
                           int maxMessageSize,
                           int numMessagesPerLogMessage,
                           Pattern messageStartPattern,
                           TextLogMessageType messageType,
                           boolean prependTimestamp,
                           boolean prependHostName,
                           boolean trimTailingNewlineCharacter,
                           String hostname,
                           String prependFieldDelimiter,
                           String prependEnvironmentVariables); @Override LogMessageAndPosition readLogMessageAndPosition(); void close(); @Override LogFile getLogFile(); @Override long getByteOffset(); @Override void setByteOffset(long byteOffset); @Override boolean isClosed();  }