[ https://issues.apache.org/jira/browse/NIFI-1118?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15310953#comment-15310953 ]
ASF GitHub Bot commented on NIFI-1118: -------------------------------------- Github user mosermw commented on a diff in the pull request: https://github.com/apache/nifi/pull/444#discussion_r65425886 --- Diff: nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java --- @@ -147,138 +193,200 @@ protected void init(final ProcessorInitializationContext context) { return properties; } - private int readLines(final InputStream in, final int maxNumLines, final OutputStream out, final boolean keepAllNewLines, final byte[] leadingNewLineBytes) throws IOException { + private int readLines(final InputStream in, final int maxNumLines, final long maxByteCount, final OutputStream out, + final boolean includeLineDelimiter, final byte[] leadingNewLineBytes) throws IOException { final EndOfLineBuffer eolBuffer = new EndOfLineBuffer(); - int numLines = 0; byte[] leadingBytes = leadingNewLineBytes; + int numLines = 0; + long totalBytes = 0L; for (int i = 0; i < maxNumLines; i++) { - final EndOfLineMarker eolMarker = locateEndOfLine(in, out, false, eolBuffer, leadingBytes); + final EndOfLineMarker eolMarker = countBytesToSplitPoint(in, out, totalBytes, maxByteCount, includeLineDelimiter, eolBuffer, leadingBytes); + final long bytes = eolMarker.getBytesConsumed(); leadingBytes = eolMarker.getLeadingNewLineBytes(); - if (keepAllNewLines && out != null) { + if (includeLineDelimiter && out != null) { if (leadingBytes != null) { out.write(leadingBytes); leadingBytes = null; } - eolBuffer.drainTo(out); } - - if (eolBuffer.length() > 0 || eolMarker.getBytesConsumed() > 0L) { - numLines++; + totalBytes += bytes; + if (bytes <= 0) { + return numLines; } - - if (eolMarker.isStreamEnded()) { + numLines++; + if (totalBytes >= maxByteCount) { break; } } - return numLines; } - private EndOfLineMarker locateEndOfLine(final InputStream in, final OutputStream out, final boolean includeLineDelimiter, - final EndOfLineBuffer eolBuffer, final byte[] leadingNewLineBytes) throws IOException { - + private EndOfLineMarker countBytesToSplitPoint(final InputStream in, final OutputStream out, final long bytesReadSoFar, final long maxSize, + final boolean includeLineDelimiter, final EndOfLineBuffer eolBuffer, final byte[] leadingNewLineBytes) throws IOException { int lastByte = -1; long bytesRead = 0L; + final ByteArrayOutputStream buffer; + if (out != null) { + buffer = new ByteArrayOutputStream(); + } else { + buffer = null; + } byte[] bytesToWriteFirst = leadingNewLineBytes; + in.mark(Integer.MAX_VALUE); while (true) { - in.mark(1); final int nextByte = in.read(); - final boolean isNewLineChar = nextByte == '\r' || nextByte == '\n'; - - // if we hit end of stream or new line we're done + // if we hit end of stream we're done if (nextByte == -1) { - if (lastByte == '\r') { - eolBuffer.addEndOfLine(true, false); + if (buffer != null) { + buffer.writeTo(out); + buffer.close(); } - - return new EndOfLineMarker(bytesRead, eolBuffer, true, bytesToWriteFirst); + return new EndOfLineMarker(bytesRead, eolBuffer, true, bytesToWriteFirst); // bytesToWriteFirst should be "null"? } - // If we get a character that's not an end-of-line char, then - // we need to write out the EOL's that we have buffered (if out != null). - // Then, we need to reset our EOL buffer because we no longer have consecutive EOL's - if (!isNewLineChar) { + // Verify leading bytes do not violate size limitation + if (bytesToWriteFirst != null && (bytesToWriteFirst.length + bytesRead) > (maxSize - bytesReadSoFar) && includeLineDelimiter) { + return new EndOfLineMarker(-1, eolBuffer, false, leadingNewLineBytes); + } + // Write leadingNewLines, if appropriate + if ( buffer != null && includeLineDelimiter && bytesToWriteFirst != null) { + bytesRead += bytesToWriteFirst.length; + buffer.write(bytesToWriteFirst); + bytesToWriteFirst = null; + } + // buffer the output + bytesRead++; + if (buffer != null && nextByte != '\n' && nextByte != '\r') { if (bytesToWriteFirst != null) { - if (out != null) { - out.write(bytesToWriteFirst); - } - - bytesToWriteFirst = null; - } - - if (out != null) { - eolBuffer.drainTo(out); + buffer.write(bytesToWriteFirst); } - + bytesToWriteFirst = null; + eolBuffer.drainTo(buffer); eolBuffer.clear(); + buffer.write(nextByte); } - // if there's an OutputStream to copy the data to, copy it, if appropriate. - // "if appropriate" means that it's not a line delimiter or that we want to copy line delimiters - bytesRead++; - if (out != null && (includeLineDelimiter || !isNewLineChar)) { - if (bytesToWriteFirst != null) { - out.write(bytesToWriteFirst); - bytesToWriteFirst = null; + // check the size limit + if (bytesRead > (maxSize-bytesReadSoFar) && bytesReadSoFar > 0) { + in.reset(); + if (buffer != null) { + buffer.close(); } - - out.write(nextByte); + return new EndOfLineMarker(-1, eolBuffer, false, leadingNewLineBytes); } // if we have a new line, then we're done if (nextByte == '\n') { - eolBuffer.addEndOfLine(lastByte == '\r', true); + if (buffer != null) { + buffer.writeTo(out); + buffer.close(); + eolBuffer.addEndOfLine(false, true); //TODO: verify "false" is equivalent to "lastByte == '\r'" --- End diff -- Is this TODO a reminder to future @markobean? Does a current unit test verify it? > Enable SplitText processor to limit line length and filter header lines > ----------------------------------------------------------------------- > > Key: NIFI-1118 > URL: https://issues.apache.org/jira/browse/NIFI-1118 > Project: Apache NiFi > Issue Type: Improvement > Components: Extensions > Reporter: Mark Bean > Assignee: Mark Bean > Fix For: 0.7.0 > > > Include the following functionality to the SplitText processor: > 1) Maximum size limit of the split file(s) > A new split file will be created if the next line to be added to the current > split file exceeds a user-defined maximum file size > 2) Header line marker > User-defined character(s) can be used to identify the header line(s) of the > data file rather than a predetermined number of lines > These changes are additions, not a replacement of any property or behavior. > In the case of header line marker, the existing property "Header Line Count" > must be zero for the new property and behavior to be used. -- This message was sent by Atlassian JIRA (v6.3.4#6332)