You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2017/02/24 15:14:02 UTC
nifi git commit: NIFI-2876 refactored demarcators into a common
abstract class
Repository: nifi
Updated Branches:
refs/heads/master fba17159a -> 8e17929d6
NIFI-2876 refactored demarcators into a common abstract class
This closes #1214.
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/8e17929d
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/8e17929d
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/8e17929d
Branch: refs/heads/master
Commit: 8e17929d6a68171666a79711023b79c28f476d1a
Parents: fba1715
Author: Oleg Zhurakousky <ol...@suitcase.io>
Authored: Fri Nov 11 17:36:18 2016 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Fri Feb 24 10:06:53 2017 -0500
----------------------------------------------------------------------
.../nifi/stream/io/util/AbstractDemarcator.java | 179 ++++++++++++++++
.../nifi/stream/io/util/StreamDemarcator.java | 165 +++++----------
.../nifi/stream/io/util/TextLineDemarcator.java | 207 ++++++-------------
.../stream/io/util/StreamDemarcatorTest.java | 40 +++-
.../stream/io/util/TextLineDemarcatorTest.java | 39 +++-
.../nifi/processors/standard/SplitText.java | 6 +-
6 files changed, 376 insertions(+), 260 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/8e17929d/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/AbstractDemarcator.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/AbstractDemarcator.java b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/AbstractDemarcator.java
new file mode 100644
index 0000000..37bbcd7
--- /dev/null
+++ b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/AbstractDemarcator.java
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.stream.io.util;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.BufferOverflowException;
+
+import org.apache.nifi.stream.io.exception.TokenTooLargeException;
+
+/**
+ * Base class for implementing streaming demarcators.
+ * <p>
+ * NOTE: Not intended for multi-thread usage hence not Thread-safe.
+ * </p>
+ */
+abstract class AbstractDemarcator implements Closeable {
+
+ final static int INIT_BUFFER_SIZE = 8192;
+
+ private final InputStream is;
+
+ /*
+ * The size of the initial buffer. Its value is also used when bufer needs
+ * to be expanded.
+ */
+ private final int initialBufferSize;
+
+ /*
+ * The maximum allowed size of the token. In the event such size is exceeded
+ * TokenTooLargeException is thrown.
+ */
+ private final int maxDataSize;
+
+ /*
+ * Buffer into which the bytes are read from the provided stream. The size
+ * of the buffer is defined by the 'initialBufferSize' provided in the
+ * constructor or defaults to the value of INIT_BUFFER_SIZE constant.
+ */
+ byte[] buffer;
+
+ /*
+ * Starting offset of the demarcated token within the current 'buffer'.
+ */
+ int index;
+
+ /*
+ * Starting offset of the demarcated token within the current 'buffer'. Keep
+ * in mind that while most of the time it is the same as the 'index' it may
+ * also have a value of 0 at which point it serves as a signal to the fill()
+ * operation that buffer needs to be expended if end of token is not reached
+ * (see fill() operation for more details).
+ */
+ int mark;
+
+ /*
+ * Starting offset (from the beginning of the stream) of the demarcated
+ * token.
+ */
+ long offset;
+
+ /*
+ * The length of the bytes valid for reading. It is different from the
+ * buffer length, since this number may be smaller (e.g., at he end of the
+ * stream) then actual buffer length. It is set by the fill() operation
+ * every time more bytes read into buffer.
+ */
+ int availableBytesLength;
+
+ /**
+ * Constructs an instance of demarcator with provided {@link InputStream}
+ * and max buffer size. Each demarcated token must fit within max buffer
+ * size, otherwise the exception will be raised.
+ */
+ AbstractDemarcator(InputStream is, int maxDataSize) {
+ this(is, maxDataSize, INIT_BUFFER_SIZE);
+ }
+
+ /**
+ * Constructs an instance of demarcator with provided {@link InputStream}
+ * and max buffer size and initial buffer size. Each demarcated token must
+ * fit within max buffer size, otherwise the exception will be raised.
+ */
+ AbstractDemarcator(InputStream is, int maxDataSize, int initialBufferSize) {
+ this.validate(is, maxDataSize, initialBufferSize);
+ this.is = is;
+ this.initialBufferSize = initialBufferSize;
+ this.buffer = new byte[initialBufferSize];
+ this.maxDataSize = maxDataSize;
+ }
+
+ @Override
+ public void close() throws IOException {
+ this.is.close();
+ }
+
+ /**
+ * Will fill the current buffer from current 'index' position, expanding it
+ * and or shuffling it if necessary. If buffer exceeds max buffer size a
+ * {@link TokenTooLargeException} will be thrown.
+ *
+ * @throws IOException
+ * if unable to read from the stream
+ */
+ void fill() throws IOException {
+ if (this.index >= this.buffer.length) {
+ if (this.mark == 0) { // expand
+ long expandedSize = this.buffer.length + this.initialBufferSize;
+ if (expandedSize > Integer.MAX_VALUE) {
+ throw new BufferOverflowException(); // will probably OOM before this will ever happen, but just in case.
+ }
+ byte[] newBuff = new byte[(int) expandedSize];
+ System.arraycopy(this.buffer, 0, newBuff, 0, this.buffer.length);
+ this.buffer = newBuff;
+ } else { // shuffle
+ int length = this.index - this.mark;
+ System.arraycopy(this.buffer, this.mark, this.buffer, 0, length);
+ this.index = length;
+ this.mark = 0;
+ }
+ }
+
+ int bytesRead;
+ /*
+ * The do/while pattern is used here similar to the way it is used in
+ * BufferedReader essentially protecting from assuming the EOS until it
+ * actually is since not every implementation of InputStream guarantees
+ * that bytes are always available while the stream is open.
+ */
+ do {
+ bytesRead = this.is.read(this.buffer, this.index, this.buffer.length - this.index);
+ } while (bytesRead == 0);
+ this.availableBytesLength = bytesRead != -1 ? this.index + bytesRead : -1;
+ }
+
+ /**
+ * Will extract data token of the provided length from the current buffer
+ * starting at the 'mark'.
+ */
+ byte[] extractDataToken(int length) throws IOException {
+ if (length > this.maxDataSize) {
+ throw new TokenTooLargeException("A message in the stream exceeds the maximum allowed message size of " + this.maxDataSize + " bytes.");
+ }
+ byte[] data = null;
+ if (length > 0) {
+ data = new byte[length];
+ System.arraycopy(this.buffer, this.mark, data, 0, data.length);
+ }
+ return data;
+ }
+
+ /**
+ * Validates prerequisites for constructor arguments
+ */
+ private void validate(InputStream is, int maxDataSize, int initialBufferSize) {
+ if (is == null) {
+ throw new IllegalArgumentException("'is' must not be null");
+ } else if (maxDataSize <= 0) {
+ throw new IllegalArgumentException("'maxDataSize' must be > 0");
+ } else if (initialBufferSize <= 0) {
+ throw new IllegalArgumentException("'initialBufferSize' must be > 0");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/8e17929d/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/StreamDemarcator.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/StreamDemarcator.java b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/StreamDemarcator.java
index 840bdb0..ce738df 100644
--- a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/StreamDemarcator.java
+++ b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/StreamDemarcator.java
@@ -16,40 +16,24 @@
*/
package org.apache.nifi.stream.io.util;
-import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
-import org.apache.nifi.stream.io.exception.TokenTooLargeException;
-
/**
* The <code>StreamDemarcator</code> class takes an input stream and demarcates
* it so it could be read (see {@link #nextToken()}) as individual byte[]
- * demarcated by the provided delimiter. If delimiter is not provided the entire
- * stream will be read into a single token which may result in
- * {@link OutOfMemoryError} if stream is too large.
+ * demarcated by the provided delimiter (see 'delimiterBytes'). If delimiter is
+ * not provided the entire stream will be read into a single token which may
+ * result in {@link OutOfMemoryError} if stream is too large. The 'maxDataSize'
+ * controls the maximum size of the buffer that accumulates a token.
+ * <p>
+ * NOTE: Not intended for multi-thread usage hence not Thread-safe.
+ * </p>
*/
-public class StreamDemarcator implements Closeable {
-
- private final static int INIT_BUFFER_SIZE = 8192;
-
- private final InputStream is;
+public class StreamDemarcator extends AbstractDemarcator {
private final byte[] delimiterBytes;
- private final int maxDataSize;
-
- private final int initialBufferSize;
-
-
- private byte[] buffer;
-
- private int index;
-
- private int mark;
-
- private int readAheadLength;
-
/**
* Constructs a new instance
*
@@ -57,10 +41,14 @@ public class StreamDemarcator implements Closeable {
* instance of {@link InputStream} representing the data
* @param delimiterBytes
* byte array representing delimiter bytes used to split the
- * input stream. Can be null
+ * input stream. Can be 'null'. NOTE: the 'null' is allowed only
+ * for convenience and consistency since without delimiter this
+ * class is no different then BufferedReader which reads the
+ * entire stream into a byte array and there may be a more
+ * efficient ways to do that (if that is the case).
* @param maxDataSize
* maximum size of data derived from the input stream. This means
- * that neither {@link InputStream} nor its individual chunks (if
+ * that neither {@link InputStream} nor its individual tokens (if
* delimiter is used) can ever be greater then this size.
*/
public StreamDemarcator(InputStream is, byte[] delimiterBytes, int maxDataSize) {
@@ -74,10 +62,14 @@ public class StreamDemarcator implements Closeable {
* instance of {@link InputStream} representing the data
* @param delimiterBytes
* byte array representing delimiter bytes used to split the
- * input stream. Can be null
+ * input stream. Can be 'null'. NOTE: the 'null' is allowed only
+ * for convenience and consistency since without delimiter this
+ * class is no different then BufferedReader which reads the
+ * entire stream into a byte array and there may be a more
+ * efficient ways to do that (if that is the case).
* @param maxDataSize
* maximum size of data derived from the input stream. This means
- * that neither {@link InputStream} nor its individual chunks (if
+ * that neither {@link InputStream} nor its individual tokens (if
* delimiter is used) can ever be greater then this size.
* @param initialBufferSize
* initial size of the buffer used to buffer {@link InputStream}
@@ -87,12 +79,9 @@ public class StreamDemarcator implements Closeable {
*
*/
public StreamDemarcator(InputStream is, byte[] delimiterBytes, int maxDataSize, int initialBufferSize) {
- this.validateInput(is, delimiterBytes, maxDataSize, initialBufferSize);
- this.is = is;
+ super(is, maxDataSize, initialBufferSize);
+ this.validate(delimiterBytes);
this.delimiterBytes = delimiterBytes;
- this.initialBufferSize = initialBufferSize;
- this.buffer = new byte[initialBufferSize];
- this.maxDataSize = maxDataSize;
}
/**
@@ -102,99 +91,55 @@ public class StreamDemarcator implements Closeable {
* @throws IOException if unable to read from the stream
*/
public byte[] nextToken() throws IOException {
- byte[] data = null;
+ byte[] token = null;
int j = 0;
-
- while (data == null && this.buffer != null) {
- if (this.index >= this.readAheadLength) {
+ nextTokenLoop:
+ while (token == null && this.availableBytesLength != -1) {
+ if (this.index >= this.availableBytesLength) {
this.fill();
}
- if (this.index >= this.readAheadLength) {
- data = this.extractDataToken(0);
- this.buffer = null;
- } else {
- byte byteVal = this.buffer[this.index++];
- if (this.delimiterBytes != null && this.delimiterBytes[j] == byteVal) {
- if (++j == this.delimiterBytes.length) {
- data = this.extractDataToken(this.delimiterBytes.length);
+ if (this.availableBytesLength != -1) {
+ byte byteVal;
+ int i;
+ for (i = this.index; i < this.availableBytesLength; i++) {
+ byteVal = this.buffer[i];
+
+ boolean delimiterFound = false;
+ if (this.delimiterBytes != null && this.delimiterBytes[j] == byteVal) {
+ if (++j == this.delimiterBytes.length) {
+ delimiterFound = true;
+ }
+ } else {
+ j = 0;
+ }
+
+ if (delimiterFound) {
+ this.index = i + 1;
+ int size = this.index - this.mark - this.delimiterBytes.length;
+ token = this.extractDataToken(size);
this.mark = this.index;
j = 0;
+ if (token != null) {
+ break nextTokenLoop;
+ }
}
- } else {
- j = 0;
}
- }
- }
- return data;
- }
-
- /**
- * Will fill the current buffer from current 'index' position, expanding it
- * and or shuffling it if necessary
- *
- * @throws IOException if unable to read from the stream
- */
- private void fill() throws IOException {
- if (this.index >= this.buffer.length) {
- if (this.mark == 0) { // expand
- byte[] newBuff = new byte[this.buffer.length + this.initialBufferSize];
- System.arraycopy(this.buffer, 0, newBuff, 0, this.buffer.length);
- this.buffer = newBuff;
- } else { // shuffle
- int length = this.index - this.mark;
- System.arraycopy(this.buffer, this.mark, this.buffer, 0, length);
- this.index = length;
- this.mark = 0;
- this.readAheadLength = length;
+ this.index = i;
+ } else {
+ token = this.extractDataToken(this.index - this.mark);
}
}
- int bytesRead;
- do {
- bytesRead = this.is.read(this.buffer, this.index, this.buffer.length - this.index);
- } while (bytesRead == 0);
-
- if (bytesRead != -1) {
- this.readAheadLength = this.index + bytesRead;
- if (this.readAheadLength > this.maxDataSize) {
- throw new TokenTooLargeException("A message in the stream exceeds the maximum allowed message size of " + this.maxDataSize + " bytes.");
- }
- }
+ return token;
}
- /**
- * Will extract data token from the current buffer. The length of the data
- * token is between the current 'mark' and 'index' minus 'lengthSubtract'
- * which signifies the length of the delimiter (if any). If the above
- * subtraction results in length 0, null is returned.
- */
- private byte[] extractDataToken(int lengthSubtract) {
- byte[] data = null;
- int length = this.index - this.mark - lengthSubtract;
- if (length > 0) {
- data = new byte[length];
- System.arraycopy(this.buffer, this.mark, data, 0, data.length);
- }
- return data;
- }
/**
- *
+ * Validates prerequisites for constructor arguments
*/
- private void validateInput(InputStream is, byte[] delimiterBytes, int maxDataSize, int initialBufferSize) {
- if (is == null) {
- throw new IllegalArgumentException("'is' must not be null");
- } else if (maxDataSize <= 0) {
- throw new IllegalArgumentException("'maxDataSize' must be > 0");
- } else if (initialBufferSize <= 0) {
- throw new IllegalArgumentException("'initialBufferSize' must be > 0");
- } else if (delimiterBytes != null && delimiterBytes.length == 0){
+ private void validate(byte[] delimiterBytes) {
+ if (delimiterBytes != null && delimiterBytes.length == 0) {
throw new IllegalArgumentException("'delimiterBytes' is an optional argument, but when provided its length must be > 0");
}
}
-
- @Override
- public void close() throws IOException {
- is.close();
- }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/8e17929d/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/TextLineDemarcator.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/TextLineDemarcator.java b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/TextLineDemarcator.java
index 2b5c7fe..080bd14 100644
--- a/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/TextLineDemarcator.java
+++ b/nifi-commons/nifi-utils/src/main/java/org/apache/nifi/stream/io/util/TextLineDemarcator.java
@@ -28,26 +28,14 @@ import java.io.InputStream;
* computed text line. See {@link #nextOffsetInfo()} and
* {@link #nextOffsetInfo(byte[])} for more details.
* <p>
- * This class is NOT thread-safe.
+ * NOTE: Not intended for multi-thread usage hence not Thread-safe.
* </p>
*/
-public class TextLineDemarcator {
+public class TextLineDemarcator extends AbstractDemarcator {
- private final static int INIT_BUFFER_SIZE = 8192;
+ private static int CR = 13; // \r
- private final InputStream is;
-
- private final int initialBufferSize;
-
- private byte[] buffer;
-
- private int index;
-
- private int mark;
-
- private long offset;
-
- private int bufferLength;
+ private static int LF = 10; // \n
/**
* Constructs an instance of demarcator with provided {@link InputStream}
@@ -62,15 +50,7 @@ public class TextLineDemarcator {
* and initial buffer size.
*/
public TextLineDemarcator(InputStream is, int initialBufferSize) {
- if (is == null) {
- throw new IllegalArgumentException("'is' must not be null.");
- }
- if (initialBufferSize < 1) {
- throw new IllegalArgumentException("'initialBufferSize' must be > 0.");
- }
- this.is = is;
- this.initialBufferSize = initialBufferSize;
- this.buffer = new byte[initialBufferSize];
+ super(is, Integer.MAX_VALUE, initialBufferSize);
}
/**
@@ -81,7 +61,7 @@ public class TextLineDemarcator {
*
* @return offset info
*/
- public OffsetInfo nextOffsetInfo() {
+ public OffsetInfo nextOffsetInfo() throws IOException {
return this.nextOffsetInfo(null);
}
@@ -90,59 +70,78 @@ public class TextLineDemarcator {
* by either '\r', '\n' or '\r\n'). <br>
* The <i>offset info</i> computed and returned as {@link OffsetInfo} where
* {@link OffsetInfo#isStartsWithMatch()} will return true if
- * <code>startsWith</code> was successfully matched with the stsarting bytes
+ * <code>startsWith</code> was successfully matched with the starting bytes
* of the text line.
*
+ * NOTE: The reason for 2 'nextOffsetInfo(..)' operations is that the
+ * 'startsWith' argument will force the actual token to be extracted and
+ * then matched introducing the overhead for System.arrayCopy and matching
+ * logic which is an optional scenario and is avoided all together if
+ * 'startsWith' is not provided (i.e., null).
+ *
* @return offset info
*/
- public OffsetInfo nextOffsetInfo(byte[] startsWith) {
+ public OffsetInfo nextOffsetInfo(byte[] startsWith) throws IOException {
OffsetInfo offsetInfo = null;
- int lineLength = 0;
- byte[] token = null;
- lineLoop:
- while (this.bufferLength != -1) {
- if (this.index >= this.bufferLength) {
+ byte previousByteVal = 0;
+ byte[] data = null;
+ nextTokenLoop:
+ while (data == null && this.availableBytesLength != -1) {
+ if (this.index >= this.availableBytesLength) {
this.fill();
}
- if (this.bufferLength != -1) {
- int i;
+ int delimiterSize = 0;
+ if (this.availableBytesLength != -1) {
byte byteVal;
- for (i = this.index; i < this.bufferLength; i++) {
+ int i;
+ for (i = this.index; i < this.availableBytesLength; i++) {
byteVal = this.buffer[i];
- lineLength++;
- int crlfLength = computeEol(byteVal, i + 1);
- if (crlfLength > 0) {
- i += crlfLength;
- if (crlfLength == 2) {
- lineLength++;
- }
- offsetInfo = new OffsetInfo(this.offset, lineLength, crlfLength);
+
+ if (byteVal == LF) {
+ delimiterSize = previousByteVal == CR ? 2 : 1;
+ } else if (previousByteVal == CR) {
+ delimiterSize = 1;
+ i--;
+ }
+ previousByteVal = byteVal;
+ if (delimiterSize > 0) {
+ this.index = i + 1;
+ int size = Math.max(1, this.index - this.mark);
+ offsetInfo = new OffsetInfo(this.offset, size, delimiterSize);
+ this.offset += size;
if (startsWith != null) {
- token = this.extractDataToken(lineLength);
+ data = this.extractDataToken(size);
}
this.mark = this.index;
- break lineLoop;
+ break nextTokenLoop;
}
}
this.index = i;
+ } else {
+ delimiterSize = previousByteVal == CR || previousByteVal == LF ? 1 : 0;
+ if (offsetInfo == null) {
+ int size = this.index - this.mark;
+ if (size > 0) {
+ offsetInfo = new OffsetInfo(this.offset, size, delimiterSize);
+ this.offset += size;
+ }
+ }
+ if (startsWith != null) {
+ data = this.extractDataToken(this.index - this.mark);
+ }
}
}
- // EOF where last char(s) are not CRLF.
- if (lineLength > 0 && offsetInfo == null) {
- offsetInfo = new OffsetInfo(this.offset, lineLength, 0);
- if (startsWith != null) {
- token = this.extractDataToken(lineLength);
- }
- }
- this.offset += lineLength;
- // checks if the new line starts with 'startsWith' chars
- if (startsWith != null) {
- for (int i = 0; i < startsWith.length; i++) {
- byte sB = startsWith[i];
- if (token != null && sB != token[i]) {
- offsetInfo.setStartsWithMatch(0);
- break;
+ if (startsWith != null && data != null) {
+ if (startsWith.length > data.length) {
+ offsetInfo.setStartsWithMatch(false);
+ } else {
+ for (int i = 0; i < startsWith.length; i++) {
+ byte sB = startsWith[i];
+ if (sB != data[i]) {
+ offsetInfo.setStartsWithMatch(false);
+ break;
+ }
}
}
}
@@ -150,81 +149,6 @@ public class TextLineDemarcator {
}
/**
- * Determines if the line terminates. Returns int specifying the length of
- * the CRLF (i.e., only CR or LF or CR and LF) and therefore can only have
- * values of:
- * 0 - not the end of the line
- * 1 - the end of the line either via CR or LF
- * 2 - the end of the line with both CR and LF
- *
- * It performs the read ahead on the buffer if need to.
- */
- private int computeEol(byte currentByte, int providedIndex) {
- int actualIndex = providedIndex - 1;
- boolean readAhead = false;
- int crlfLength = 0;
- if (currentByte == '\n') {
- crlfLength = 1;
- } else if (currentByte == '\r') {
- if (providedIndex >= this.bufferLength) {
- this.index = this.bufferLength;
- this.fill();
- providedIndex = this.index;
- readAhead = true;
- }
- crlfLength = 1;
- if (providedIndex < this.buffer.length - 1) {
- currentByte = this.buffer[providedIndex];
- crlfLength = currentByte == '\n' ? 2 : 1;
- }
- }
-
- if (crlfLength > 0) {
- this.index = readAhead ? this.index + (crlfLength - 1) : (actualIndex + crlfLength);
- }
-
- return crlfLength;
- }
-
- private byte[] extractDataToken(int length) {
- byte[] data = null;
- if (length > 0) {
- data = new byte[length];
- System.arraycopy(this.buffer, this.mark, data, 0, data.length);
- }
- return data;
- }
-
- /**
- * Will fill the current buffer from current 'index' position, expanding it
- * and or shuffling it if necessary
- */
- private void fill() {
- if (this.index >= this.buffer.length) {
- if (this.mark == 0) { // expand
- byte[] newBuff = new byte[this.buffer.length + this.initialBufferSize];
- System.arraycopy(this.buffer, 0, newBuff, 0, this.buffer.length);
- this.buffer = newBuff;
- } else { // shuffle
- int length = this.index - this.mark;
- System.arraycopy(this.buffer, this.mark, this.buffer, 0, length);
- this.index = length;
- this.mark = 0;
- }
- }
-
- try {
- int bytesRead;
- do {
- bytesRead = this.is.read(this.buffer, this.index, this.buffer.length - this.index);
- } while (bytesRead == 0);
- this.bufferLength = bytesRead != -1 ? this.index + bytesRead : -1;
- } catch (IOException e) {
- throw new IllegalStateException("Failed while reading InputStream", e);
- }
- }
-
- /**
* Container to hold offset and meta info for a computed text line.
* The offset and meta info is represented with the following 4 values:
* <ul>
@@ -245,7 +169,7 @@ public class TextLineDemarcator {
private boolean startsWithMatch = true;
- OffsetInfo(long startOffset, long length, int crlfLength) {
+ private OffsetInfo(long startOffset, long length, int crlfLength) {
this.startOffset = startOffset;
this.length = length;
this.crlfLength = crlfLength;
@@ -267,8 +191,13 @@ public class TextLineDemarcator {
return this.startsWithMatch;
}
- void setStartsWithMatch(int startsWithMatch) {
- this.startsWithMatch = startsWithMatch == 1 ? true : false;
+ void setStartsWithMatch(boolean startsWithMatch) {
+ this.startsWithMatch = startsWithMatch;
+ }
+
+ @Override
+ public String toString() {
+ return "offset:" + this.startOffset + "; length:" + this.length + "; crlfLength:" + this.crlfLength;
}
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/8e17929d/nifi-commons/nifi-utils/src/test/java/org/apache/nifi/stream/io/util/StreamDemarcatorTest.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-utils/src/test/java/org/apache/nifi/stream/io/util/StreamDemarcatorTest.java b/nifi-commons/nifi-utils/src/test/java/org/apache/nifi/stream/io/util/StreamDemarcatorTest.java
index cfac6de..16af077 100644
--- a/nifi-commons/nifi-utils/src/test/java/org/apache/nifi/stream/io/util/StreamDemarcatorTest.java
+++ b/nifi-commons/nifi-utils/src/test/java/org/apache/nifi/stream/io/util/StreamDemarcatorTest.java
@@ -32,8 +32,8 @@ import java.util.Arrays;
import org.junit.Test;
+@SuppressWarnings("resource")
public class StreamDemarcatorTest {
-
@Test
public void validateInitializationFailure() {
try {
@@ -66,6 +66,41 @@ public class StreamDemarcatorTest {
}
@Test
+ public void validateLargeBufferSmallMaxSize() throws IOException {
+ final byte[] inputData = "A Great Benefit To Us All".getBytes(StandardCharsets.UTF_8);
+
+ try (final InputStream is = new ByteArrayInputStream(inputData);
+ final StreamDemarcator demarcator = new StreamDemarcator(is, "B".getBytes(StandardCharsets.UTF_8), 24, 4096)) {
+
+ final byte[] first = demarcator.nextToken();
+ assertNotNull(first);
+ assertEquals("A Great ", new String(first));
+
+ final byte[] second = demarcator.nextToken();
+ assertNotNull(second);
+ assertEquals("enefit To Us All", new String(second));
+
+ assertNull(demarcator.nextToken());
+
+ }
+ }
+
+ @Test
+ public void vaidateOnPartialMatchThenSubsequentPartialMatch() throws IOException {
+ final byte[] inputData = "A Great Big Boy".getBytes(StandardCharsets.UTF_8);
+ final byte[] delimBytes = "AB".getBytes(StandardCharsets.UTF_8);
+
+ try (final InputStream is = new ByteArrayInputStream(inputData);
+ final StreamDemarcator demarcator = new StreamDemarcator(is, delimBytes, 4096)) {
+
+ final byte[] bytes = demarcator.nextToken();
+ assertArrayEquals(inputData, bytes);
+
+ assertNull(demarcator.nextToken());
+ }
+ }
+
+ @Test
public void validateNoDelimiter() throws IOException {
String data = "Learn from yesterday, live for today, hope for tomorrow. The important thing is not to stop questioning.";
ByteArrayInputStream is = new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8));
@@ -196,10 +231,11 @@ public class StreamDemarcatorTest {
@Test(expected = IOException.class)
public void validateMaxBufferSize() throws IOException {
- String data = "THIS IS MY TEXT<MY DELIMITER>THIS IS MY NEW TEXT<MY DELIMITER>THIS IS MY NEWEST TEXT";
+ String data = "THIS IS MY TEXT<MY DELIMITER>THIS IS MY NEW TEXT THEN<MY DELIMITER>THIS IS MY NEWEST TEXT";
ByteArrayInputStream is = new ByteArrayInputStream(data.getBytes());
StreamDemarcator scanner = new StreamDemarcator(is, "<MY DELIMITER>".getBytes(StandardCharsets.UTF_8), 20);
scanner.nextToken();
+ scanner.nextToken();
}
@Test
http://git-wip-us.apache.org/repos/asf/nifi/blob/8e17929d/nifi-commons/nifi-utils/src/test/java/org/apache/nifi/stream/io/util/TextLineDemarcatorTest.java
----------------------------------------------------------------------
diff --git a/nifi-commons/nifi-utils/src/test/java/org/apache/nifi/stream/io/util/TextLineDemarcatorTest.java b/nifi-commons/nifi-utils/src/test/java/org/apache/nifi/stream/io/util/TextLineDemarcatorTest.java
index 321c51b..fefd9eb 100644
--- a/nifi-commons/nifi-utils/src/test/java/org/apache/nifi/stream/io/util/TextLineDemarcatorTest.java
+++ b/nifi-commons/nifi-utils/src/test/java/org/apache/nifi/stream/io/util/TextLineDemarcatorTest.java
@@ -31,6 +31,7 @@ import java.nio.charset.StandardCharsets;
import org.apache.nifi.stream.io.util.TextLineDemarcator.OffsetInfo;
import org.junit.Test;
+@SuppressWarnings("resource")
public class TextLineDemarcatorTest {
@Test(expected = IllegalArgumentException.class)
@@ -44,7 +45,7 @@ public class TextLineDemarcatorTest {
}
@Test
- public void emptyStreamNoStartWithFilter() {
+ public void emptyStreamNoStartWithFilter() throws IOException {
String data = "";
InputStream is = stringToIs(data);
TextLineDemarcator demarcator = new TextLineDemarcator(is);
@@ -53,7 +54,7 @@ public class TextLineDemarcatorTest {
@Test
- public void emptyStreamAndStartWithFilter() {
+ public void emptyStreamAndStartWithFilter() throws IOException {
String data = "";
InputStream is = stringToIs(data);
TextLineDemarcator demarcator = new TextLineDemarcator(is);
@@ -63,7 +64,7 @@ public class TextLineDemarcatorTest {
// this test has no assertions. It's success criteria is validated by lack
// of failure (see NIFI-3278)
@Test
- public void endsWithCRWithBufferLengthEqualStringLengthA() {
+ public void endsWithCRWithBufferLengthEqualStringLengthA() throws Exception {
String str = "\r";
InputStream is = stringToIs(str);
TextLineDemarcator demarcator = new TextLineDemarcator(is, str.length());
@@ -72,7 +73,7 @@ public class TextLineDemarcatorTest {
}
@Test
- public void endsWithCRWithBufferLengthEqualStringLengthB() {
+ public void endsWithCRWithBufferLengthEqualStringLengthB() throws Exception {
String str = "abc\r";
InputStream is = stringToIs(str);
TextLineDemarcator demarcator = new TextLineDemarcator(is, str.length());
@@ -81,7 +82,7 @@ public class TextLineDemarcatorTest {
}
@Test
- public void singleCR() {
+ public void singleCR() throws IOException {
InputStream is = stringToIs("\r");
TextLineDemarcator demarcator = new TextLineDemarcator(is);
OffsetInfo offsetInfo = demarcator.nextOffsetInfo();
@@ -92,7 +93,7 @@ public class TextLineDemarcatorTest {
}
@Test
- public void singleLF() {
+ public void singleLF() throws IOException {
InputStream is = stringToIs("\n");
TextLineDemarcator demarcator = new TextLineDemarcator(is);
OffsetInfo offsetInfo = demarcator.nextOffsetInfo();
@@ -121,7 +122,7 @@ public class TextLineDemarcatorTest {
}
@Test
- public void validateNiFi_3495() {
+ public void validateNiFi_3495() throws IOException {
String str = "he\ra-to-a\rb-to-b\rc-to-c\r\nd-to-d";
InputStream is = stringToIs(str);
TextLineDemarcator demarcator = new TextLineDemarcator(is, 10);
@@ -313,6 +314,30 @@ public class TextLineDemarcatorTest {
assertEquals(0, second.getCrlfLength());
}
+ @Test
+ public void validateStartsWithLongerThanLastToken() throws IOException {
+ final byte[] inputData = "This is going to be a spectacular test\nThis is".getBytes(StandardCharsets.UTF_8);
+ final byte[] startsWith = "This is going to be".getBytes(StandardCharsets.UTF_8);
+
+ try (final InputStream is = new ByteArrayInputStream(inputData);
+ final TextLineDemarcator demarcator = new TextLineDemarcator(is)) {
+
+ final OffsetInfo first = demarcator.nextOffsetInfo(startsWith);
+ assertNotNull(first);
+ assertEquals(0, first.getStartOffset());
+ assertEquals(39, first.getLength());
+ assertEquals(1, first.getCrlfLength());
+ assertTrue(first.isStartsWithMatch());
+
+ final OffsetInfo second = demarcator.nextOffsetInfo(startsWith);
+ assertNotNull(second);
+ assertEquals(39, second.getStartOffset());
+ assertEquals(7, second.getLength());
+ assertEquals(0, second.getCrlfLength());
+ assertFalse(second.isStartsWithMatch());
+ }
+ }
+
private InputStream stringToIs(String data) {
return new ByteArrayInputStream(data.getBytes(StandardCharsets.UTF_8));
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/8e17929d/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java
index 4e62993..5738632 100644
--- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java
+++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/SplitText.java
@@ -377,7 +377,8 @@ public class SplitText extends AbstractProcessor {
* will vary but the length of the split will never be > {@link #maxSplitSize} and {@link IllegalStateException} will be thrown.
* This method also allows one to provide 'startsWithFilter' to allow headers to be determined via such filter (see {@link #HEADER_MARKER}.
*/
- private SplitInfo computeHeader(TextLineDemarcator demarcator, long startOffset, long splitMaxLineCount, byte[] startsWithFilter, SplitInfo previousSplitInfo) {
+ private SplitInfo computeHeader(TextLineDemarcator demarcator, long startOffset, long splitMaxLineCount,
+ byte[] startsWithFilter, SplitInfo previousSplitInfo) throws IOException {
long length = 0;
long actualLineCount = 0;
OffsetInfo offsetInfo = null;
@@ -419,7 +420,8 @@ public class SplitText extends AbstractProcessor {
* If split size is controlled by the {@link #maxSplitSize}, then the resulting {@link SplitInfo}
* line count will vary but the length of the split will never be > {@link #maxSplitSize}.
*/
- private SplitInfo nextSplit(TextLineDemarcator demarcator, long startOffset, long splitMaxLineCount, SplitInfo remainderSplitInfo, long startingLength) {
+ private SplitInfo nextSplit(TextLineDemarcator demarcator, long startOffset, long splitMaxLineCount,
+ SplitInfo remainderSplitInfo, long startingLength) throws IOException {
long length = 0;
long trailingCrlfLength = 0;
long actualLineCount = 0;