Repository: nifi Updated Branches: refs/heads/master 903f4981a -> ee14d8f9d
http://git-wip-us.apache.org/repos/asf/nifi/blob/f378ee90/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java index fb51d45..3cb7eda 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java @@ -25,17 +25,19 @@ import java.io.OutputStreamWriter; import java.nio.charset.Charset; import java.util.ArrayList; import java.util.Collections; +import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.regex.Matcher; import java.util.regex.Pattern; +import org.apache.commons.io.IOUtils; import org.apache.nifi.annotation.behavior.EventDriven; import org.apache.nifi.annotation.behavior.InputRequirement; import org.apache.nifi.annotation.behavior.InputRequirement.Requirement; -import org.apache.commons.io.IOUtils; import org.apache.nifi.annotation.behavior.SideEffectFree; import org.apache.nifi.annotation.behavior.SupportsBatching; import org.apache.nifi.annotation.documentation.CapabilityDescription; @@ -53,6 +55,7 @@ import org.apache.nifi.processor.ProcessSession; import org.apache.nifi.processor.ProcessorInitializationContext; import org.apache.nifi.processor.Relationship; import org.apache.nifi.processor.exception.ProcessException; +import org.apache.nifi.processor.io.InputStreamCallback; import org.apache.nifi.processor.io.OutputStreamCallback; import org.apache.nifi.processor.io.StreamCallback; import org.apache.nifi.processor.util.FlowFileFilters; @@ -77,7 +80,8 @@ public class ReplaceText extends AbstractProcessor { public static final String appendValue = "Append"; public static final String regexReplaceValue = "Regex Replace"; public static final String literalReplaceValue = "Literal Replace"; - private final Pattern backReferencePattern = Pattern.compile("\\$(\\d+)"); + public static final String alwaysReplace = "Always Replace"; + private static final Pattern backReferencePattern = Pattern.compile("\\$(\\d+)"); private static final String DEFAULT_REGEX = "(?s:^.*$)"; private static final String DEFAULT_REPLACEMENT_VALUE = "$1"; @@ -95,6 +99,9 @@ public class ReplaceText extends AbstractProcessor { "Interpret the Search Value as a Regular Expression and replace all matches with the Replacement Value. The Replacement Value may reference Capturing Groups used " + "in the Search Value by using a dollar-sign followed by the Capturing Group number, such as $1 or $2. If the Search Value is set to .* then everything is replaced without " + "even evaluating the Regular Expression."); + static final AllowableValue ALWAYS_REPLACE = new AllowableValue(alwaysReplace, alwaysReplace, + "Always replaces the entire line or the entire contents of the FlowFile (depending on the value of the <Evaluation Mode> property) and does not bother searching " + + "for any value. When this strategy is chosen, the <Search Value> property is ignored."); public static final PropertyDescriptor SEARCH_VALUE = new PropertyDescriptor.Builder() .name("Regular Expression") @@ -108,7 +115,9 @@ public class ReplaceText extends AbstractProcessor { public static final PropertyDescriptor REPLACEMENT_VALUE = new PropertyDescriptor.Builder() .name("Replacement Value") .description("The value to insert using the 'Replacement Strategy'. Using \"Regex Replace\" back-references to Regular Expression capturing groups " - + "are supported, but back-references that reference capturing groups that do not exist in the regular expression will be treated as literal value.") + + "are supported, but back-references that reference capturing groups that do not exist in the regular expression will be treated as literal value. " + + "Back References may also be referenced using the Expression Language, as '$1', '$2', etc. The single-tick marks MUST be included, as these variables are " + + "not \"Standard\" attribute names (attribute names must be quoted unless they contain only numbers, letters, and _).") .required(true) .defaultValue(DEFAULT_REPLACEMENT_VALUE) .addValidator(Validator.VALID) @@ -124,11 +133,11 @@ public class ReplaceText extends AbstractProcessor { public static final PropertyDescriptor MAX_BUFFER_SIZE = new PropertyDescriptor.Builder() .name("Maximum Buffer Size") .description("Specifies the maximum amount of data to buffer (per file or per line, depending on the Evaluation Mode) in order to " - + "apply the regular expressions. If 'Entire Text' (in Evaluation Mode) is selected and the FlowFile is larger than this value, " + + "apply the replacement. If 'Entire Text' (in Evaluation Mode) is selected and the FlowFile is larger than this value, " + "the FlowFile will be routed to 'failure'. " + "In 'Line-by-Line' Mode, if a single line is larger than this value, the FlowFile will be routed to 'failure'. A default value " + "of 1 MB is provided, primarily for 'Entire Text' mode. In 'Line-by-Line' Mode, a value such as 8 KB or 16 KB is suggested. " - + "This value is ignored and the buffer is not used if 'Regular Expression' is set to '.*'") + + "This value is ignored if the <Replacement Strategy> property is set to one of: Append, Prepend, Always Replace") .required(true) .addValidator(StandardValidators.DATA_SIZE_VALIDATOR) .defaultValue("1 MB") @@ -136,7 +145,7 @@ public class ReplaceText extends AbstractProcessor { public static final PropertyDescriptor REPLACEMENT_STRATEGY = new PropertyDescriptor.Builder() .name("Replacement Strategy") .description("The strategy for how and what to replace within the FlowFile's text content.") - .allowableValues(PREPEND, APPEND, REGEX_REPLACE, LITERAL_REPLACE) + .allowableValues(PREPEND, APPEND, REGEX_REPLACE, LITERAL_REPLACE, ALWAYS_REPLACE) .defaultValue(REGEX_REPLACE.getValue()) .required(true) .build(); @@ -210,18 +219,6 @@ public class ReplaceText extends AbstractProcessor { return; } - final AttributeValueDecorator escapeBackRefDecorator = new AttributeValueDecorator() { - @Override - public String decorate(final String attributeValue) { - return attributeValue.replace("$", "\\$"); - } - }; - - final String regexValue = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions().getValue(); - final int numCapturingGroups = Pattern.compile(regexValue).matcher("").groupCount(); - - final boolean skipBuffer = ".*".equals(unsubstitutedRegex); - final Charset charset = Charset.forName(context.getProperty(CHARACTER_SET).getValue()); final int maxBufferSize = context.getProperty(MAX_BUFFER_SIZE).asDataSize(DataUnit.B).intValue(); @@ -233,82 +230,151 @@ public class ReplaceText extends AbstractProcessor { buffer = null; } + ReplacementStrategyExecutor replacementStrategyExecutor; + switch (replacementStrategy) { + case prependValue: + replacementStrategyExecutor = new PrependReplace(); + break; + case appendValue: + replacementStrategyExecutor = new AppendReplace(); + break; + case regexReplaceValue: + // for backward compatibility - if replacement regex is ".*" then we will simply always replace the content. + if (context.getProperty(SEARCH_VALUE).getValue().equals(".*")) { + replacementStrategyExecutor = new AlwaysReplace(); + } else { + replacementStrategyExecutor = new RegexReplace(buffer, context); + } + + break; + case literalReplaceValue: + replacementStrategyExecutor = new LiteralReplace(buffer); + break; + case alwaysReplace: + replacementStrategyExecutor = new AlwaysReplace(); + break; + default: + throw new AssertionError(); + } + for (FlowFile flowFile : flowFiles) { if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { - if (flowFile.getSize() > maxBufferSize && !skipBuffer) { + if (flowFile.getSize() > maxBufferSize && replacementStrategyExecutor.isAllDataBufferedForEntireText()) { session.transfer(flowFile, REL_FAILURE); continue; } } - String replacement; - if (!replacementStrategy.equals(regexReplaceValue)) { - replacement = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue(); - } else { - replacement = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile, escapeBackRefDecorator).getValue(); - final Matcher backRefMatcher = backReferencePattern.matcher(replacement); - while (backRefMatcher.find()) { - final String backRefNum = backRefMatcher.group(1); - if (backRefNum.startsWith("0")) { - continue; - } - final int originalBackRefIndex = Integer.parseInt(backRefNum); - int backRefIndex = originalBackRefIndex; - - // if we have a replacement value like $123, and we have less than 123 capturing groups, then - // we want to truncate the 3 and use capturing group 12; if we have less than 12 capturing groups, - // then we want to truncate the 2 and use capturing group 1; if we don't have a capturing group then - // we want to truncate the 1 and get 0. - while (backRefIndex > numCapturingGroups && backRefIndex >= 10) { - backRefIndex /= 10; - } + final StopWatch stopWatch = new StopWatch(true); - if (backRefIndex > numCapturingGroups) { - final StringBuilder sb = new StringBuilder(replacement.length() + 1); - final int groupStart = backRefMatcher.start(1); + flowFile = replacementStrategyExecutor.replace(flowFile, session, context, evaluateMode, charset, maxBufferSize); - sb.append(replacement.substring(0, groupStart - 1)); - sb.append("\\"); - sb.append(replacement.substring(groupStart - 1)); - replacement = sb.toString(); - } - } + logger.info("Transferred {} to 'success'", new Object[] {flowFile}); + session.getProvenanceReporter().modifyContent(flowFile, stopWatch.getElapsed(TimeUnit.MILLISECONDS)); + session.transfer(flowFile, REL_SUCCESS); + } + } + + + // If we find a back reference that is not valid, then we will treat it as a literal string. For example, if we have 3 capturing + // groups and the Replacement Value has the value is "I owe $8 to him", then we want to treat the $8 as a literal "$8", rather + // than attempting to use it as a back reference. + private static String escapeLiteralBackReferences(final String unescaped, final int numCapturingGroups) { + if (numCapturingGroups == 0) { + return unescaped; + } + + String value = unescaped; + final Matcher backRefMatcher = backReferencePattern.matcher(value); + while (backRefMatcher.find()) { + final String backRefNum = backRefMatcher.group(1); + if (backRefNum.startsWith("0")) { + continue; + } + final int originalBackRefIndex = Integer.parseInt(backRefNum); + int backRefIndex = originalBackRefIndex; + + // if we have a replacement value like $123, and we have less than 123 capturing groups, then + // we want to truncate the 3 and use capturing group 12; if we have less than 12 capturing groups, + // then we want to truncate the 2 and use capturing group 1; if we don't have a capturing group then + // we want to truncate the 1 and get 0. + while (backRefIndex > numCapturingGroups && backRefIndex >= 10) { + backRefIndex /= 10; } - ReplacementStrategyExecutor replacementStrategyExecutor; - switch (replacementStrategy) { - case prependValue: - replacementStrategyExecutor = new PrependReplace(); - break; - case appendValue: - replacementStrategyExecutor = new AppendReplace(); - break; - case regexReplaceValue: - replacementStrategyExecutor = new RegexReplace(buffer); - break; - case literalReplaceValue: - replacementStrategyExecutor = new LiteralReplace(buffer); - break; - default: - throw new AssertionError(); + if (backRefIndex > numCapturingGroups) { + final StringBuilder sb = new StringBuilder(value.length() + 1); + final int groupStart = backRefMatcher.start(1); + + sb.append(value.substring(0, groupStart - 1)); + sb.append("\\"); + sb.append(value.substring(groupStart - 1)); + value = sb.toString(); } + } - final StopWatch stopWatch = new StopWatch(true); + return value; + } - flowFile = replacementStrategyExecutor.replace(flowFile, session, context, replacement, evaluateMode, - charset, maxBufferSize, skipBuffer); + private static class AlwaysReplace implements ReplacementStrategyExecutor { + @Override + public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { - logger.info("Transferred {} to 'success'", new Object[] {flowFile}); - session.getProvenanceReporter().modifyContent(flowFile, stopWatch.getElapsed(TimeUnit.MILLISECONDS)); - session.transfer(flowFile, REL_SUCCESS); + final String replacementValue = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue(); + final StringBuilder lineEndingBuilder = new StringBuilder(2); + + if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { + flowFile = session.write(flowFile, new StreamCallback() { + @Override + public void process(final InputStream in, final OutputStream out) throws IOException { + out.write(replacementValue.getBytes(charset)); + } + }); + } else { + flowFile = session.write(flowFile, new StreamCallback() { + @Override + public void process(final InputStream in, final OutputStream out) throws IOException { + try (NLKBufferedReader br = new NLKBufferedReader(new InputStreamReader(in, charset), maxBufferSize); + BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(out, charset));) { + + String line; + while ((line = br.readLine()) != null) { + // We need to determine what line ending was used and use that after our replacement value. + lineEndingBuilder.setLength(0); + for (int i = line.length() - 1; i >= 0; i--) { + final char c = line.charAt(i); + if (c == '\r' || c == '\n') { + lineEndingBuilder.append(c); + } else { + break; + } + } + + bw.write(replacementValue); + + // Preserve original line endings. Reverse string because we iterated over original line ending in reverse order, appending to builder. + // So if builder has multiple characters, they are now reversed from the original string's ordering. + bw.write(lineEndingBuilder.reverse().toString()); + } + } + } + }); + } + + return flowFile; + } + + @Override + public boolean isAllDataBufferedForEntireText() { + return false; } } private static class PrependReplace implements ReplacementStrategyExecutor { - @Override - public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String replacementValue, final String evaluateMode, - final Charset charset, final int maxBufferSize, final boolean skipBuffer) { + public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { + final String replacementValue = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue(); + if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { flowFile = session.write(flowFile, new StreamCallback() { @Override @@ -334,13 +400,19 @@ public class ReplaceText extends AbstractProcessor { } return flowFile; } + + @Override + public boolean isAllDataBufferedForEntireText() { + return false; + } } private static class AppendReplace implements ReplacementStrategyExecutor { @Override - public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String replacementValue, final String evaluateMode, - final Charset charset, final int maxBufferSize, final boolean skipBuffer) { + public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { + final String replacementValue = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue(); + if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { flowFile = session.write(flowFile, new StreamCallback() { @Override @@ -387,81 +459,121 @@ public class ReplaceText extends AbstractProcessor { } return flowFile; } + + @Override + public boolean isAllDataBufferedForEntireText() { + return false; + } } private static class RegexReplace implements ReplacementStrategyExecutor { private final byte[] buffer; + private final int numCapturingGroups; + private final Map<String, String> additionalAttrs; + + private static final AttributeValueDecorator escapeBackRefDecorator = new AttributeValueDecorator() { + @Override + public String decorate(final String attributeValue) { + return attributeValue.replace("$", "\\$"); + } + }; - public RegexReplace(final byte[] buffer) { + public RegexReplace(final byte[] buffer, final ProcessContext context) { this.buffer = buffer; + + final String regexValue = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions().getValue(); + numCapturingGroups = Pattern.compile(regexValue).matcher("").groupCount(); + additionalAttrs = new HashMap<>(numCapturingGroups); } @Override - public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String replacementValue, final String evaluateMode, - final Charset charset, final int maxBufferSize, final boolean skipBuffer) { - final String replacementFinal = replacementValue.replaceAll("(\\$\\D)", "\\\\$1"); - - // always match; just overwrite value with the replacement value; this optimization prevents us - // from reading the file at all. - if (skipBuffer) { - if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { - flowFile = session.write(flowFile, new OutputStreamCallback() { - @Override - public void process(final OutputStream out) throws IOException { - out.write(replacementFinal.getBytes(charset)); - } - }); - } else { - flowFile = session.write(flowFile, new StreamCallback() { - @Override - public void process(final InputStream in, final OutputStream out) throws IOException { - try (NLKBufferedReader br = new NLKBufferedReader(new InputStreamReader(in, charset), maxBufferSize); - BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(out, charset));) { - while (null != br.readLine()) { - bw.write(replacementFinal); - } - } - } - }); + public FlowFile replace(final FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { + final AttributeValueDecorator quotedAttributeDecorator = new AttributeValueDecorator() { + @Override + public String decorate(final String attributeValue) { + return Pattern.quote(attributeValue); } - } else { - final AttributeValueDecorator quotedAttributeDecorator = new AttributeValueDecorator() { + }; + final String searchRegex = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue(); + final Pattern searchPattern = Pattern.compile(searchRegex); + + final int flowFileSize = (int) flowFile.getSize(); + FlowFile updatedFlowFile; + if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { + session.read(flowFile, new InputStreamCallback() { @Override - public String decorate(final String attributeValue) { - return Pattern.quote(attributeValue); + public void process(final InputStream in) throws IOException { + StreamUtils.fillBuffer(in, buffer, false); } - }; - final String searchRegex = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue(); + }); + + final String contentString = new String(buffer, 0, flowFileSize, charset); + additionalAttrs.clear(); + final Matcher matcher = searchPattern.matcher(contentString); + if (matcher.find()) { + for (int i = 1; i <= matcher.groupCount(); i++) { + final String groupValue = matcher.group(i); + additionalAttrs.put("$" + i, groupValue); + } + + String replacement = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile, additionalAttrs, escapeBackRefDecorator).getValue(); + replacement = escapeLiteralBackReferences(replacement, numCapturingGroups); + + // If we have a $ followed by anything other than a number, then escape it. E.g., $d becomes \$d so that it can be used as a literal in a regex. + final String replacementFinal = replacement.replaceAll("(\\$\\D)", "\\\\$1"); - final int flowFileSize = (int) flowFile.getSize(); - if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { - flowFile = session.write(flowFile, new StreamCallback() { + final String updatedValue = contentString.replaceAll(searchRegex, replacementFinal); + updatedFlowFile = session.write(flowFile, new OutputStreamCallback() { @Override - public void process(final InputStream in, final OutputStream out) throws IOException { - StreamUtils.fillBuffer(in, buffer, false); - final String contentString = new String(buffer, 0, flowFileSize, charset); - final String updatedValue = contentString.replaceAll(searchRegex, replacementFinal); + public void process(final OutputStream out) throws IOException { out.write(updatedValue.getBytes(charset)); } }); } else { - flowFile = session.write(flowFile, new StreamCallback() { - @Override - public void process(final InputStream in, final OutputStream out) throws IOException { - try (NLKBufferedReader br = new NLKBufferedReader(new InputStreamReader(in, charset), maxBufferSize); - BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(out, charset))) { - String oneLine; - while (null != (oneLine = br.readLine())) { + // If no match, just return the original. No need to write out any content. + return flowFile; + } + } else { + updatedFlowFile = session.write(flowFile, new StreamCallback() { + @Override + public void process(final InputStream in, final OutputStream out) throws IOException { + try (NLKBufferedReader br = new NLKBufferedReader(new InputStreamReader(in, charset), maxBufferSize); + BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(out, charset))) { + String oneLine; + while (null != (oneLine = br.readLine())) { + additionalAttrs.clear(); + final Matcher matcher = searchPattern.matcher(oneLine); + if (matcher.find()) { + for (int i = 1; i <= matcher.groupCount(); i++) { + final String groupValue = matcher.group(i); + additionalAttrs.put("$" + i, groupValue); + } + + String replacement = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile, additionalAttrs, escapeBackRefDecorator).getValue(); + replacement = escapeLiteralBackReferences(replacement, numCapturingGroups); + + // If we have a $ followed by anything other than a number, then escape it. E.g., $d becomes \$d so that it can be used as a literal in a regex. + final String replacementFinal = replacement.replaceAll("(\\$\\D)", "\\\\$1"); + final String updatedValue = oneLine.replaceAll(searchRegex, replacementFinal); bw.write(updatedValue); + } else { + // No match. Just write out the line as it was. + bw.write(oneLine); } } } - }); - } + } + }); } - return flowFile; + + return updatedFlowFile; + } + + @Override + public boolean isAllDataBufferedForEntireText() { + return true; } } @@ -473,8 +585,10 @@ public class ReplaceText extends AbstractProcessor { } @Override - public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String replacementValue, final String evaluateMode, - final Charset charset, final int maxBufferSize, final boolean skipBuffer) { + public FlowFile replace(FlowFile flowFile, final ProcessSession session, final ProcessContext context, final String evaluateMode, final Charset charset, final int maxBufferSize) { + + final String replacementValue = context.getProperty(REPLACEMENT_VALUE).evaluateAttributeExpressions(flowFile).getValue(); + final AttributeValueDecorator quotedAttributeDecorator = new AttributeValueDecorator() { @Override public String decorate(final String attributeValue) { @@ -484,7 +598,6 @@ public class ReplaceText extends AbstractProcessor { final String searchValue = context.getProperty(SEARCH_VALUE).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue(); - final int flowFileSize = (int) flowFile.getSize(); if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) { flowFile = session.write(flowFile, new StreamCallback() { @@ -515,9 +628,16 @@ public class ReplaceText extends AbstractProcessor { } return flowFile; } + + @Override + public boolean isAllDataBufferedForEntireText() { + return true; + } } private interface ReplacementStrategyExecutor { - FlowFile replace(FlowFile flowFile, ProcessSession session, ProcessContext context, String replacement, String evaluateMode, Charset charset, int maxBufferSize, boolean skipBuffer); + FlowFile replace(FlowFile flowFile, ProcessSession session, ProcessContext context, String evaluateMode, Charset charset, int maxBufferSize); + + boolean isAllDataBufferedForEntireText(); } } http://git-wip-us.apache.org/repos/asf/nifi/blob/f378ee90/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestReplaceText.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestReplaceText.java b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestReplaceText.java index 3a311a3..561a1e0 100644 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestReplaceText.java +++ b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestReplaceText.java @@ -914,13 +914,13 @@ public class TestReplaceText { final Map<String, String> attributes = new HashMap<>(); attributes.put("abc", "Good"); - runner.enqueue(translateNewLines(Paths.get("src/test/resources/TestReplaceTextLineByLine/testFile.txt")), attributes); + runner.enqueue(Paths.get("src/test/resources/TestReplaceTextLineByLine/testFile.txt"), attributes); runner.run(); runner.assertAllFlowFilesTransferred(ReplaceText.REL_SUCCESS, 1); final MockFlowFile out = runner.getFlowFilesForRelationship(ReplaceText.REL_SUCCESS).get(0); - out.assertContentEquals(translateNewLines(new File("src/test/resources/TestReplaceTextLineByLine/Good.txt"))); + out.assertContentEquals("Good\nGood\nGood\nGood\nGood\nGood\nGood\nGood\nGood\nGood\nGood"); } @Test @@ -939,7 +939,7 @@ public class TestReplaceText { runner.assertAllFlowFilesTransferred(ReplaceText.REL_SUCCESS, 1); final MockFlowFile out = runner.getFlowFilesForRelationship(ReplaceText.REL_SUCCESS).get(0); - out.assertContentEquals("GoodGoodGood"); + out.assertContentEquals("Good\r\nGood\r\nGood\r"); } @Test @@ -984,17 +984,108 @@ public class TestReplaceText { System.out.println(outContent); Assert.assertTrue(outContent.equals("attribute header\n\nabc.txt\n\ndata header\n\nHello\n\n\nfooter\n" + "attribute header\n\nabc.txt\n\ndata header\n\nWorld!\n\nfooter\n")); + } + + @Test + public void testCapturingGroupInExpressionLanguage() { + final TestRunner runner = TestRunners.newTestRunner(new ReplaceText()); + runner.setValidateExpressionUsage(false); + runner.setProperty(ReplaceText.EVALUATION_MODE, ReplaceText.LINE_BY_LINE); + runner.setProperty(ReplaceText.SEARCH_VALUE, "(.*?),(.*?),(\\d+.*)"); + runner.setProperty(ReplaceText.REPLACEMENT_VALUE, "$1,$2,${ '$3':toDate('ddMMMyyyy'):format('yyyy/MM/dd') }"); + + final String csvIn = + "2006,10-01-2004,10may2004\n" + + "2007,15-05-2006,10jun2005\r\n" + + "2009,8-8-2008,10aug2008"; + final String expectedCsvOut = + "2006,10-01-2004,2004/05/10\n" + + "2007,15-05-2006,2005/06/10\r\n" + + "2009,8-8-2008,2008/08/10"; + + runner.enqueue(csvIn.getBytes()); + + runner.run(); + + runner.assertAllFlowFilesTransferred(ReplaceText.REL_SUCCESS, 1); + final MockFlowFile out = runner.getFlowFilesForRelationship(ReplaceText.REL_SUCCESS).get(0); + out.assertContentEquals(expectedCsvOut); + } + + @Test + public void testCapturingGroupInExpressionLanguage2() { + final TestRunner runner = TestRunners.newTestRunner(new ReplaceText()); + runner.setValidateExpressionUsage(false); + runner.setProperty(ReplaceText.EVALUATION_MODE, ReplaceText.LINE_BY_LINE); + runner.setProperty(ReplaceText.SEARCH_VALUE, "(.*)/(.*?).jpg"); + runner.setProperty(ReplaceText.REPLACEMENT_VALUE, "$1/${ '$2':substring(0,1) }.png"); + + final String csvIn = + "1,2,3,https://123.jpg,[email protected]\n"; + + "3,2,1,https://321.jpg,[email protected]";; + final String expectedCsvOut = + "1,2,3,https://1.png,[email protected]\n"; + + "3,2,1,https://3.png,[email protected]";; + + runner.enqueue(csvIn.getBytes()); + + runner.run(); + + runner.assertAllFlowFilesTransferred(ReplaceText.REL_SUCCESS, 1); + final MockFlowFile out = runner.getFlowFilesForRelationship(ReplaceText.REL_SUCCESS).get(0); + out.assertContentEquals(expectedCsvOut); + } + @Test + public void testAlwaysReplaceEntireText() { + final TestRunner runner = TestRunners.newTestRunner(new ReplaceText()); + runner.setValidateExpressionUsage(false); + runner.setProperty(ReplaceText.EVALUATION_MODE, ReplaceText.ENTIRE_TEXT); + runner.setProperty(ReplaceText.REPLACEMENT_STRATEGY, ReplaceText.ALWAYS_REPLACE); + runner.setProperty(ReplaceText.SEARCH_VALUE, "i do not exist anywhere in the text"); + runner.setProperty(ReplaceText.REPLACEMENT_VALUE, "${filename}"); + + final Map<String, String> attributes = new HashMap<>(); + attributes.put("filename", "abc.txt"); + runner.enqueue("Hello\nWorld!".getBytes(), attributes); + + runner.run(); + + runner.assertAllFlowFilesTransferred(ReplaceText.REL_SUCCESS, 1); + final MockFlowFile out = runner.getFlowFilesForRelationship(ReplaceText.REL_SUCCESS).get(0); + out.assertContentEquals("abc.txt"); + } + + @Test + public void testAlwaysReplaceLineByLine() { + final TestRunner runner = TestRunners.newTestRunner(new ReplaceText()); + runner.setValidateExpressionUsage(false); + runner.setProperty(ReplaceText.EVALUATION_MODE, ReplaceText.LINE_BY_LINE); + runner.setProperty(ReplaceText.REPLACEMENT_STRATEGY, ReplaceText.ALWAYS_REPLACE); + runner.setProperty(ReplaceText.SEARCH_VALUE, "i do not exist anywhere in the text"); + runner.setProperty(ReplaceText.REPLACEMENT_VALUE, "${filename}"); + + final Map<String, String> attributes = new HashMap<>(); + attributes.put("filename", "abc.txt"); + runner.enqueue("Hello\nWorld!\r\ntoday!\n".getBytes(), attributes); + + runner.run(); + + runner.assertAllFlowFilesTransferred(ReplaceText.REL_SUCCESS, 1); + final MockFlowFile out = runner.getFlowFilesForRelationship(ReplaceText.REL_SUCCESS).get(0); + out.assertContentEquals("abc.txt\nabc.txt\r\nabc.txt\n"); } - private byte[] translateNewLines(final File file) throws IOException { + + + private String translateNewLines(final File file) throws IOException { return translateNewLines(file.toPath()); } - private byte[] translateNewLines(final Path path) throws IOException { + private String translateNewLines(final Path path) throws IOException { final byte[] data = Files.readAllBytes(path); final String text = new String(data, StandardCharsets.UTF_8); - return translateNewLines(text).getBytes(StandardCharsets.UTF_8); + return translateNewLines(text); } private String translateNewLines(final String text) { http://git-wip-us.apache.org/repos/asf/nifi/blob/f378ee90/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestReplaceTextLineByLine/Good.txt ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestReplaceTextLineByLine/Good.txt b/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestReplaceTextLineByLine/Good.txt deleted file mode 100755 index 6e90a7d..0000000 --- a/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/TestReplaceTextLineByLine/Good.txt +++ /dev/null @@ -1 +0,0 @@ -GoodGoodGoodGoodGoodGoodGoodGoodGoodGoodGood \ No newline at end of file
![https://123.jpg,[email protected]\n"](https://123.jpg,em...@mydomain.com\n"){kind=link}
![https://321.jpg,[email protected]"](https://321.jpg,other.em...@mydomain.com"){kind=link}
![https://1.png,[email protected]\n"](https://1.png,em...@mydomain.com\n"){kind=link}
![https://3.png,[email protected]"](https://3.png,other.em...@mydomain.com"){kind=link}