[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user asfgit closed the pull request at: https://github.com/apache/nifi/pull/444 --- If your project is set up for it, you can reply to this email and have your reply appear on GitHub as well. If your project does not have this feature enabled and wishes so, or if the feature is enabled but not working, please contact infrastructure at infrastruct...@apache.org or file a JIRA ticket with INFRA. ---
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user markobean commented on a diff in the pull request:
https://github.com/apache/nifi/pull/444#discussion_r65471941
--- Diff:
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestSplitText.java
---
@@ -39,6 +39,303 @@
+
"\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\nLastLine\n";
@Test
+public void testLastLineExceedsSizeLimit() {
+final TestRunner runner = TestRunners.newTestRunner(new
SplitText());
+runner.setProperty(SplitText.HEADER_LINE_COUNT, "0");
+runner.setProperty(SplitText.LINE_SPLIT_COUNT, "2");
+runner.setProperty(SplitText.FRAGMENT_MAX_SIZE, "20 B");
+
+runner.enqueue("Line #1\nLine #2\nLine #3\nLong line exceeding
limit");
+runner.run();
+
+runner.assertTransferCount(SplitText.REL_FAILURE, 0);
+runner.assertTransferCount(SplitText.REL_ORIGINAL, 1);
+runner.assertTransferCount(SplitText.REL_SPLITS, 3);
+}
--- End diff --
Talk about thorough! I don't know how you managed to catch this edge case.
However, for completeness, the RTN functionality once again increased
complexity and another buffer count was added. Now, if the size limit is
exceeded (with RTN = true and no header lines), then the number of EOL bytes
previously added to the info.lengthBytes are subtracted from the previous line
effectively removing the final EOL characters of the split.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user mosermw commented on a diff in the pull request:
https://github.com/apache/nifi/pull/444#discussion_r65469870
--- 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;
+}
--- End diff --
No objection to leaving it as-is.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user markobean commented on a diff in the pull request:
https://github.com/apache/nifi/pull/444#discussion_r65468881
--- 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;
+}
--- End diff --
It made more sense when this method had the possibility of having 'out' be
null; this occurred by design in a previous version. Now, it is only called
with a valid OutputStream. Still, I think this doesn't hurt and adds a layer of
safety should the method be called differently in the future. However, if you
still object, the if/else can be removed and the final buffer be instantiated
outright; all unit tests confirmed to pass in this configuration (although not
staged to be committed at this time.)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user markobean commented on a diff in the pull request:
https://github.com/apache/nifi/pull/444#discussion_r65467699
--- 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 && includ
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user markobean commented on a diff in the pull request:
https://github.com/apache/nifi/pull/444#discussion_r65467195
--- 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;
--- End diff --
removed lastByte variable
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user markobean commented on a diff in the pull request:
https://github.com/apache/nifi/pull/444#discussion_r65467161
--- Diff:
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java
---
@@ -94,12 +118,16 @@
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
.defaultValue("0")
.build();
+public static final PropertyDescriptor HEADER_MARKER = new
PropertyDescriptor.Builder()
+.name("Header Line Marker Characters")
+.description("The first character(s) on the line of the
datafile which signifies a header line. This value is ignored when Header Line
Count is non-zero. " +
+"The first line not containing the Header Line Marker
Characters and all subsequent lines are considered non-header")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.build();
public static final PropertyDescriptor REMOVE_TRAILING_NEWLINES = new
PropertyDescriptor.Builder()
.name("Remove Trailing Newlines")
-.description("Whether to remove newlines at the end of each
split file. This should be false if you intend to merge the split files later.
If this is set to "
-+ "'true' and a FlowFile is generated that contains only
'empty lines' (i.e., consists only of \r and \n characters), the FlowFile will
not be emitted. "
-+ "Note, however, that if the Header Line Count is greater
than 0, the resultant FlowFile will never be empty as it will consist of the
header lines, so "
-+ "a FlowFile may be emitted that contians only the header
lines.")
+.description("Whether to remove newlines at the end of each
split file. This should be false if you intend to merge the split files later.")
--- End diff --
@mosermw, agreed. Original description put back in place (with spelling
correction)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user mosermw commented on a diff in the pull request:
https://github.com/apache/nifi/pull/444#discussion_r65440697
--- Diff:
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestSplitText.java
---
@@ -39,6 +39,303 @@
+
"\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\n\nLastLine\n";
@Test
+public void testLastLineExceedsSizeLimit() {
+final TestRunner runner = TestRunners.newTestRunner(new
SplitText());
+runner.setProperty(SplitText.HEADER_LINE_COUNT, "0");
+runner.setProperty(SplitText.LINE_SPLIT_COUNT, "2");
+runner.setProperty(SplitText.FRAGMENT_MAX_SIZE, "20 B");
+
+runner.enqueue("Line #1\nLine #2\nLine #3\nLong line exceeding
limit");
+runner.run();
+
+runner.assertTransferCount(SplitText.REL_FAILURE, 0);
+runner.assertTransferCount(SplitText.REL_ORIGINAL, 1);
+runner.assertTransferCount(SplitText.REL_SPLITS, 3);
+}
--- End diff --
I examined the content of each output FlowFile. The first and third look
fine, but the second FlowFile had a trailing newline even though the dreaded
REMOVE_TRAILING_NEWLINES defaults to "true".
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
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 && includeL
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user mosermw commented on a diff in the pull request:
https://github.com/apache/nifi/pull/444#discussion_r65425598
--- 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;
+}
--- End diff --
I stared at this if-else for longer than I care to admit. A ternary
operator for conditional assignment to a final variable would have helped me,
at least, maybe others too.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user mosermw commented on a diff in the pull request:
https://github.com/apache/nifi/pull/444#discussion_r65418261
--- 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;
--- End diff --
Looks like the lastByte local var is not used anymore.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi pull request #444: NIFI-1118 adding SplitText features of size limit an...
Github user mosermw commented on a diff in the pull request:
https://github.com/apache/nifi/pull/444#discussion_r65414058
--- Diff:
nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java
---
@@ -94,12 +118,16 @@
.addValidator(StandardValidators.NON_NEGATIVE_INTEGER_VALIDATOR)
.defaultValue("0")
.build();
+public static final PropertyDescriptor HEADER_MARKER = new
PropertyDescriptor.Builder()
+.name("Header Line Marker Characters")
+.description("The first character(s) on the line of the
datafile which signifies a header line. This value is ignored when Header Line
Count is non-zero. " +
+"The first line not containing the Header Line Marker
Characters and all subsequent lines are considered non-header")
+.required(false)
+.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+.build();
public static final PropertyDescriptor REMOVE_TRAILING_NEWLINES = new
PropertyDescriptor.Builder()
.name("Remove Trailing Newlines")
-.description("Whether to remove newlines at the end of each
split file. This should be false if you intend to merge the split files later.
If this is set to "
-+ "'true' and a FlowFile is generated that contains only
'empty lines' (i.e., consists only of \r and \n characters), the FlowFile will
not be emitted. "
-+ "Note, however, that if the Header Line Count is greater
than 0, the resultant FlowFile will never be empty as it will consist of the
header lines, so "
-+ "a FlowFile may be emitted that contians only the header
lines.")
+.description("Whether to remove newlines at the end of each
split file. This should be false if you intend to merge the split files later.")
--- End diff --
@markobean I think the longer description is still valid and should remain
in place.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
