You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by mc...@apache.org on 2015/04/28 16:04:45 UTC
[11/50] [abbrv] incubator-nifi git commit: NIFI-271
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/d29a2d68/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutJMS.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutJMS.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutJMS.java
index b37471e..65bbb36 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutJMS.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutJMS.java
@@ -96,14 +96,14 @@ public class PutJMS extends AbstractProcessor {
public static final Charset UTF8 = Charset.forName("UTF-8");
public static final int DEFAULT_MESSAGE_PRIORITY = 4;
- public static final Relationship REL_SUCCESS = new Relationship.Builder().
- name("success").
- description("All FlowFiles that are sent to the JMS destination are routed to this relationship").
- build();
- public static final Relationship REL_FAILURE = new Relationship.Builder().
- name("failure").
- description("All FlowFiles that cannot be routed to the JMS destination are routed to this relationship").
- build();
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("All FlowFiles that are sent to the JMS destination are routed to this relationship")
+ .build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("All FlowFiles that cannot be routed to the JMS destination are routed to this relationship")
+ .build();
private final Queue<WrappedMessageProducer> producerQueue = new LinkedBlockingQueue<>();
private final List<PropertyDescriptor> properties;
@@ -156,10 +156,7 @@ public class PutJMS extends AbstractProcessor {
@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
final ProcessorLog logger = getLogger();
- final List<FlowFile> flowFiles = session.get(context.
- getProperty(BATCH_SIZE).
- asInteger().
- intValue());
+ final List<FlowFile> flowFiles = session.get(context.getProperty(BATCH_SIZE).asInteger().intValue());
if (flowFiles.isEmpty()) {
return;
}
@@ -167,14 +164,10 @@ public class PutJMS extends AbstractProcessor {
WrappedMessageProducer wrappedProducer = producerQueue.poll();
if (wrappedProducer == null) {
try {
- wrappedProducer = JmsFactory.
- createMessageProducer(context, true);
- logger.info("Connected to JMS server {}", new Object[]{context.
- getProperty(URL).
- getValue()});
+ wrappedProducer = JmsFactory.createMessageProducer(context, true);
+ logger.info("Connected to JMS server {}", new Object[]{context.getProperty(URL).getValue()});
} catch (final JMSException e) {
- logger.
- error("Failed to connect to JMS Server due to {}", new Object[]{e});
+ logger.error("Failed to connect to JMS Server due to {}", new Object[]{e});
session.transfer(flowFiles, REL_FAILURE);
context.yield();
return;
@@ -184,9 +177,7 @@ public class PutJMS extends AbstractProcessor {
final Session jmsSession = wrappedProducer.getSession();
final MessageProducer producer = wrappedProducer.getProducer();
- final int maxBufferSize = context.getProperty(MAX_BUFFER_SIZE).
- asDataSize(DataUnit.B).
- intValue();
+ final int maxBufferSize = context.getProperty(MAX_BUFFER_SIZE).asDataSize(DataUnit.B).intValue();
try {
final Set<FlowFile> successfulFlowFiles = new HashSet<>();
@@ -194,8 +185,7 @@ public class PutJMS extends AbstractProcessor {
for (FlowFile flowFile : flowFiles) {
if (flowFile.getSize() > maxBufferSize) {
session.transfer(flowFile, REL_FAILURE);
- logger.
- warn("Routing {} to failure because its size exceeds the configured max", new Object[]{flowFile});
+ logger.warn("Routing {} to failure because its size exceeds the configured max", new Object[]{flowFile});
continue;
}
@@ -208,29 +198,18 @@ public class PutJMS extends AbstractProcessor {
}
});
- final Long ttl = context.getProperty(MESSAGE_TTL).
- asTimePeriod(TimeUnit.MILLISECONDS);
+ final Long ttl = context.getProperty(MESSAGE_TTL).asTimePeriod(TimeUnit.MILLISECONDS);
- final String replyToQueueName = context.
- getProperty(REPLY_TO_QUEUE).
- evaluateAttributeExpressions(flowFile).
- getValue();
- final Destination replyToQueue = replyToQueueName == null ? null : JmsFactory.
- createQueue(context, replyToQueueName);
+ final String replyToQueueName = context.getProperty(REPLY_TO_QUEUE).evaluateAttributeExpressions(flowFile).getValue();
+ final Destination replyToQueue = replyToQueueName == null ? null : JmsFactory.createQueue(context, replyToQueueName);
int priority = DEFAULT_MESSAGE_PRIORITY;
try {
- final Integer priorityInt = context.
- getProperty(MESSAGE_PRIORITY).
- evaluateAttributeExpressions(flowFile).
- asInteger();
+ final Integer priorityInt = context.getProperty(MESSAGE_PRIORITY).evaluateAttributeExpressions(flowFile).asInteger();
priority = priorityInt == null ? priority : priorityInt;
} catch (final NumberFormatException e) {
- logger.
- warn("Invalid value for JMS Message Priority: {}; defaulting to priority of {}", new Object[]{
- context.getProperty(MESSAGE_PRIORITY).
- evaluateAttributeExpressions(flowFile).
- getValue(), DEFAULT_MESSAGE_PRIORITY});
+ logger.warn("Invalid value for JMS Message Priority: {}; defaulting to priority of {}",
+ new Object[]{context.getProperty(MESSAGE_PRIORITY).evaluateAttributeExpressions(flowFile).getValue(), DEFAULT_MESSAGE_PRIORITY});
}
try {
@@ -242,16 +221,14 @@ public class PutJMS extends AbstractProcessor {
}
producer.send(message);
} catch (final JMSException e) {
- logger.
- error("Failed to send {} to JMS Server due to {}", new Object[]{flowFile, e});
+ logger.error("Failed to send {} to JMS Server due to {}", new Object[]{flowFile, e});
session.transfer(flowFiles, REL_FAILURE);
context.yield();
try {
jmsSession.rollback();
} catch (final JMSException jmse) {
- logger.
- warn("Unable to roll back JMS Session due to {}", new Object[]{jmse});
+ logger.warn("Unable to roll back JMS Session due to {}", new Object[]{jmse});
}
wrappedProducer.close(logger);
@@ -259,22 +236,17 @@ public class PutJMS extends AbstractProcessor {
}
successfulFlowFiles.add(flowFile);
- session.getProvenanceReporter().
- send(flowFile, "jms://" + context.getProperty(URL).
- getValue());
+ session.getProvenanceReporter().send(flowFile, "jms://" + context.getProperty(URL).getValue());
}
try {
jmsSession.commit();
session.transfer(successfulFlowFiles, REL_SUCCESS);
- final String flowFileDescription = successfulFlowFiles.size() > 10 ? successfulFlowFiles.
- size() + " FlowFiles" : successfulFlowFiles.toString();
- logger.
- info("Sent {} to JMS Server and transferred to 'success'", new Object[]{flowFileDescription});
+ final String flowFileDescription = successfulFlowFiles.size() > 10 ? successfulFlowFiles.size() + " FlowFiles" : successfulFlowFiles.toString();
+ logger.info("Sent {} to JMS Server and transferred to 'success'", new Object[]{flowFileDescription});
} catch (JMSException e) {
- logger.
- error("Failed to commit JMS Session due to {}; rolling back session", new Object[]{e});
+ logger.error("Failed to commit JMS Session due to {}; rolling back session", new Object[]{e});
session.rollback();
wrappedProducer.close(logger);
}
@@ -289,22 +261,19 @@ public class PutJMS extends AbstractProcessor {
final FlowFile flowFile, final Destination replyToQueue, final Integer priority) throws JMSException {
final Message message;
- switch (context.getProperty(MESSAGE_TYPE).
- getValue()) {
+ switch (context.getProperty(MESSAGE_TYPE).getValue()) {
case MSG_TYPE_EMPTY: {
message = jmsSession.createTextMessage("");
break;
}
case MSG_TYPE_STREAM: {
- final StreamMessage streamMessage = jmsSession.
- createStreamMessage();
+ final StreamMessage streamMessage = jmsSession.createStreamMessage();
streamMessage.writeBytes(messageContent);
message = streamMessage;
break;
}
case MSG_TYPE_TEXT: {
- message = jmsSession.
- createTextMessage(new String(messageContent, UTF8));
+ message = jmsSession.createTextMessage(new String(messageContent, UTF8));
break;
}
case MSG_TYPE_MAP: {
@@ -313,8 +282,7 @@ public class PutJMS extends AbstractProcessor {
}
case MSG_TYPE_BYTE:
default: {
- final BytesMessage bytesMessage = jmsSession.
- createBytesMessage();
+ final BytesMessage bytesMessage = jmsSession.createBytesMessage();
bytesMessage.writeBytes(messageContent);
message = bytesMessage;
}
@@ -330,8 +298,7 @@ public class PutJMS extends AbstractProcessor {
message.setJMSPriority(priority);
}
- if (context.getProperty(ATTRIBUTES_TO_JMS_PROPS).
- asBoolean()) {
+ if (context.getProperty(ATTRIBUTES_TO_JMS_PROPS).asBoolean()) {
copyAttributesToJmsProps(flowFile, message);
}
@@ -339,35 +306,25 @@ public class PutJMS extends AbstractProcessor {
}
/**
- * Iterates through all of the flow file's metadata and for any metadata key
- * that starts with <code>jms.</code>, the value for the corresponding key
- * is written to the JMS message as a property. The name of this property is
- * equal to the key of the flow file's metadata minus the <code>jms.</code>.
- * For example, if the flowFile has a metadata entry:
+ * Iterates through all of the flow file's metadata and for any metadata key that starts with <code>jms.</code>, the value for the corresponding key is written to the JMS message as a property.
+ * The name of this property is equal to the key of the flow file's metadata minus the <code>jms.</code>. For example, if the flowFile has a metadata entry:
* <br /><br />
* <code>jms.count</code> = <code>8</code>
* <br /><br />
- * then the JMS message will have a String property added to it with the
- * property name <code>count</code> and value <code>8</code>.
+ * then the JMS message will have a String property added to it with the property name <code>count</code> and value <code>8</code>.
*
- * If the flow file also has a metadata key with the name
- * <code>jms.count.type</code>, then the value of that metadata entry will
- * determine the JMS property type to use for the value. For example, if the
- * flow file has the following properties:
+ * If the flow file also has a metadata key with the name <code>jms.count.type</code>, then the value of that metadata entry will determine the JMS property type to use for the value. For example,
+ * if the flow file has the following properties:
* <br /><br />
* <code>jms.count</code> = <code>8</code><br />
* <code>jms.count.type</code> = <code>integer</code>
* <br /><br />
- * Then <code>message</code> will have an INTEGER property added with the
- * value 8.
+ * Then <code>message</code> will have an INTEGER property added with the value 8.
* <br /><br/>
- * If the type is not valid for the given value (e.g.,
- * <code>jms.count.type</code> = <code>integer</code> and
- * <code>jms.count</code> = <code>hello</code>, then this JMS property will
- * not be added to <code>message</code>.
+ * If the type is not valid for the given value (e.g., <code>jms.count.type</code> = <code>integer</code> and <code>jms.count</code> = <code>hello</code>, then this JMS property will not be added
+ * to <code>message</code>.
*
- * @param flowFile The flow file whose metadata should be examined for JMS
- * properties.
+ * @param flowFile The flow file whose metadata should be examined for JMS properties.
* @param message The JMS message to which we want to add properties.
* @throws JMSException ex
*/
@@ -380,49 +337,37 @@ public class PutJMS extends AbstractProcessor {
final String value = entry.getValue();
if (key.toLowerCase().
- startsWith(ATTRIBUTE_PREFIX.toLowerCase())
- && !key.toLowerCase().
- endsWith(ATTRIBUTE_TYPE_SUFFIX.toLowerCase())) {
+ startsWith(ATTRIBUTE_PREFIX.toLowerCase()) && !key.toLowerCase().endsWith(ATTRIBUTE_TYPE_SUFFIX.toLowerCase())) {
- final String jmsPropName = key.substring(ATTRIBUTE_PREFIX.
- length());
+ final String jmsPropName = key.substring(ATTRIBUTE_PREFIX.length());
final String type = attributes.get(key + ATTRIBUTE_TYPE_SUFFIX);
try {
if (type == null || type.equalsIgnoreCase(PROP_TYPE_STRING)) {
message.setStringProperty(jmsPropName, value);
} else if (type.equalsIgnoreCase(PROP_TYPE_INTEGER)) {
- message.setIntProperty(jmsPropName, Integer.
- parseInt(value));
+ message.setIntProperty(jmsPropName, Integer.parseInt(value));
} else if (type.equalsIgnoreCase(PROP_TYPE_BOOLEAN)) {
- message.setBooleanProperty(jmsPropName, Boolean.
- parseBoolean(value));
+ message.setBooleanProperty(jmsPropName, Boolean.parseBoolean(value));
} else if (type.equalsIgnoreCase(PROP_TYPE_SHORT)) {
- message.setShortProperty(jmsPropName, Short.
- parseShort(value));
+ message.setShortProperty(jmsPropName, Short.parseShort(value));
} else if (type.equalsIgnoreCase(PROP_TYPE_LONG)) {
- message.setLongProperty(jmsPropName, Long.
- parseLong(value));
+ message.setLongProperty(jmsPropName, Long.parseLong(value));
} else if (type.equalsIgnoreCase(PROP_TYPE_BYTE)) {
- message.setByteProperty(jmsPropName, Byte.
- parseByte(value));
+ message.setByteProperty(jmsPropName, Byte.parseByte(value));
} else if (type.equalsIgnoreCase(PROP_TYPE_DOUBLE)) {
- message.setDoubleProperty(jmsPropName, Double.
- parseDouble(value));
+ message.setDoubleProperty(jmsPropName, Double.parseDouble(value));
} else if (type.equalsIgnoreCase(PROP_TYPE_FLOAT)) {
- message.setFloatProperty(jmsPropName, Float.
- parseFloat(value));
+ message.setFloatProperty(jmsPropName, Float.parseFloat(value));
} else if (type.equalsIgnoreCase(PROP_TYPE_OBJECT)) {
message.setObjectProperty(jmsPropName, value);
} else {
- logger.
- warn("Attribute key '{}' for {} has value '{}', but expected one of: integer, string, object, byte, double, float, long, short, boolean; not adding this property",
- new Object[]{key, flowFile, value});
+ logger.warn("Attribute key '{}' for {} has value '{}', but expected one of: integer, string, object, byte, double, float, long, short, boolean; not adding this property",
+ new Object[]{key, flowFile, value});
}
} catch (NumberFormatException e) {
- logger.
- warn("Attribute key '{}' for {} has value '{}', but attribute key '{}' has value '{}'. Not adding this JMS property",
- new Object[]{key, flowFile, value, key + ATTRIBUTE_TYPE_SUFFIX, PROP_TYPE_INTEGER});
+ logger.warn("Attribute key '{}' for {} has value '{}', but attribute key '{}' has value '{}'. Not adding this JMS property",
+ new Object[]{key, flowFile, value, key + ATTRIBUTE_TYPE_SUFFIX, PROP_TYPE_INTEGER});
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/d29a2d68/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java
index d061c33..acabe08 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceText.java
@@ -67,54 +67,53 @@ public class ReplaceText extends AbstractProcessor {
private final Pattern backReferencePattern = Pattern.compile("\\$(\\d+)");
private static final byte[] ZERO_BYTE_BUFFER = new byte[0];
// Properties
- public static final PropertyDescriptor REGEX = new PropertyDescriptor.Builder().
- name("Regular Expression").
- description("The Regular Expression to search for in the FlowFile content").
- required(true).
- addValidator(StandardValidators.
- createRegexValidator(0, Integer.MAX_VALUE, true)).
- expressionLanguageSupported(true).
- defaultValue("(.*)").
- build();
- public static final PropertyDescriptor REPLACEMENT_VALUE = new PropertyDescriptor.Builder().
- name("Replacement Value").
- description("The value to replace the regular expression with. 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.").
- required(true).
- defaultValue("$1").
- addValidator(Validator.VALID).
- expressionLanguageSupported(true).
- build();
- public static final PropertyDescriptor CHARACTER_SET = new PropertyDescriptor.Builder().
- name("Character Set").
- description("The Character Set in which the file is encoded").
- required(true).
- addValidator(StandardValidators.CHARACTER_SET_VALIDATOR).
- defaultValue("UTF-8").
- build();
- 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, 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 '.*'").
- required(true).
- addValidator(StandardValidators.DATA_SIZE_VALIDATOR).
- defaultValue("1 MB").
- build();
- public static final PropertyDescriptor EVALUATION_MODE = new PropertyDescriptor.Builder().
- name("Evaluation Mode").
- description("Evaluate the 'Regular Expression' against each line (Line-by-Line) or buffer the entire file into memory (Entire Text) and then evaluate the 'Regular Expression'.").
- allowableValues(LINE_BY_LINE, ENTIRE_TEXT).
- defaultValue(ENTIRE_TEXT).
- required(true).
- build();
+ public static final PropertyDescriptor REGEX = new PropertyDescriptor.Builder()
+ .name("Regular Expression")
+ .description("The Regular Expression to search for in the FlowFile content")
+ .required(true)
+ .addValidator(StandardValidators.createRegexValidator(0, Integer.MAX_VALUE, true))
+ .expressionLanguageSupported(true)
+ .defaultValue("(.*)")
+ .build();
+ public static final PropertyDescriptor REPLACEMENT_VALUE = new PropertyDescriptor.Builder()
+ .name("Replacement Value")
+ .description("The value to replace the regular expression with. 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.")
+ .required(true)
+ .defaultValue("$1")
+ .addValidator(Validator.VALID)
+ .expressionLanguageSupported(true)
+ .build();
+ public static final PropertyDescriptor CHARACTER_SET = new PropertyDescriptor.Builder()
+ .name("Character Set")
+ .description("The Character Set in which the file is encoded")
+ .required(true)
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .defaultValue("UTF-8")
+ .build();
+ 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, 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 '.*'")
+ .required(true)
+ .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
+ .defaultValue("1 MB")
+ .build();
+ public static final PropertyDescriptor EVALUATION_MODE = new PropertyDescriptor.Builder()
+ .name("Evaluation Mode")
+ .description("Evaluate the 'Regular Expression' against each line (Line-by-Line) or buffer the entire file into memory (Entire Text) and then evaluate the 'Regular Expression'.")
+ .allowableValues(LINE_BY_LINE, ENTIRE_TEXT)
+ .defaultValue(ENTIRE_TEXT)
+ .required(true)
+ .build();
// Relationships
- public static final Relationship REL_SUCCESS = new Relationship.Builder().
- name("success").
- description("FlowFiles that have been successfully updated are routed to this relationship, as well as FlowFiles whose content does not match the given Regular Expression").
- build();
- public static final Relationship REL_FAILURE = new Relationship.Builder().
- name("failure").
- description("FlowFiles that could not be updated are routed to this relationship").
- build();
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("FlowFiles that have been successfully updated are routed to this relationship, as well as FlowFiles whose content does not match the given Regular Expression")
+ .build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("FlowFiles that could not be updated are routed to this relationship")
+ .build();
//
private List<PropertyDescriptor> properties;
private Set<Relationship> relationships;
@@ -147,19 +146,15 @@ public class ReplaceText extends AbstractProcessor {
@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
- final List<FlowFile> flowFiles = session.get(FlowFileFilters.
- newSizeBasedFilter(1, DataUnit.MB, 100));
+ final List<FlowFile> flowFiles = session.get(FlowFileFilters.newSizeBasedFilter(1, DataUnit.MB, 100));
if (flowFiles.isEmpty()) {
return;
}
final ProcessorLog logger = getLogger();
- final String unsubstitutedRegex = context.getProperty(REGEX).
- getValue();
- String unsubstitutedReplacement = context.getProperty(REPLACEMENT_VALUE).
- getValue();
- if (unsubstitutedRegex.equals("(.*)") && unsubstitutedReplacement.
- equals("$1")) {
+ final String unsubstitutedRegex = context.getProperty(REGEX).getValue();
+ String unsubstitutedReplacement = context.getProperty(REPLACEMENT_VALUE).getValue();
+ if (unsubstitutedRegex.equals("(.*)") && unsubstitutedReplacement.equals("$1")) {
// This pattern says replace content with itself. We can highly optimize this process by simply transferring
// all FlowFiles to the 'success' relationship
session.transfer(flowFiles, REL_SUCCESS);
@@ -180,26 +175,17 @@ public class ReplaceText extends AbstractProcessor {
}
};
- final String regexValue = context.getProperty(REGEX).
- evaluateAttributeExpressions().
- getValue();
- final int numCapturingGroups = Pattern.compile(regexValue).
- matcher("").
- groupCount();
+ final String regexValue = context.getProperty(REGEX).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();
+ final Charset charset = Charset.forName(context.getProperty(CHARACTER_SET).getValue());
+ final int maxBufferSize = context.getProperty(MAX_BUFFER_SIZE).asDataSize(DataUnit.B).intValue();
final byte[] buffer = skipBuffer ? ZERO_BYTE_BUFFER : new byte[maxBufferSize];
- final String evaluateMode = context.getProperty(EVALUATION_MODE).
- getValue();
+ final String evaluateMode = context.getProperty(EVALUATION_MODE).getValue();
for (FlowFile flowFile : flowFiles) {
if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) {
@@ -209,11 +195,8 @@ public class ReplaceText extends AbstractProcessor {
}
}
- String replacement = context.getProperty(REPLACEMENT_VALUE).
- evaluateAttributeExpressions(flowFile, escapeBackRefDecorator).
- getValue();
- final Matcher backRefMatcher = backReferencePattern.
- matcher(replacement);
+ String 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")) {
@@ -231,8 +214,7 @@ public class ReplaceText extends AbstractProcessor {
}
if (backRefIndex > numCapturingGroups) {
- final StringBuilder sb = new StringBuilder(replacement.
- length() + 1);
+ final StringBuilder sb = new StringBuilder(replacement.length() + 1);
final int groupStart = backRefMatcher.start(1);
sb.append(replacement.substring(0, groupStart - 1));
@@ -250,14 +232,12 @@ public class ReplaceText extends AbstractProcessor {
if (skipBuffer) {
final StopWatch stopWatch = new StopWatch(true);
if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) {
- flowFile = session.
- write(flowFile, new OutputStreamCallback() {
- @Override
- public void process(final OutputStream out) throws IOException {
- out.
- write(replacementValue.getBytes(charset));
- }
- });
+ flowFile = session.write(flowFile, new OutputStreamCallback() {
+ @Override
+ public void process(final OutputStream out) throws IOException {
+ out.write(replacementValue.getBytes(charset));
+ }
+ });
} else {
flowFile = session.write(flowFile, new StreamCallback() {
@Override
@@ -271,19 +251,14 @@ public class ReplaceText extends AbstractProcessor {
}
});
}
- session.getProvenanceReporter().
- modifyContent(flowFile, stopWatch.
- getElapsed(TimeUnit.MILLISECONDS));
+ session.getProvenanceReporter().modifyContent(flowFile, stopWatch.getElapsed(TimeUnit.MILLISECONDS));
session.transfer(flowFile, REL_SUCCESS);
- logger.
- info("Transferred {} to 'success'", new Object[]{flowFile});
+ logger.info("Transferred {} to 'success'", new Object[]{flowFile});
continue;
}
final StopWatch stopWatch = new StopWatch(true);
- final String regex = context.getProperty(REGEX).
- evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).
- getValue();
+ final String regex = context.getProperty(REGEX).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue();
if (evaluateMode.equalsIgnoreCase(ENTIRE_TEXT)) {
final int flowFileSize = (int) flowFile.getSize();
@@ -292,8 +267,7 @@ public class ReplaceText extends AbstractProcessor {
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(regex, replacementValue);
+ final String updatedValue = contentString.replaceAll(regex, replacementValue);
out.write(updatedValue.getBytes(charset));
}
});
@@ -305,8 +279,7 @@ public class ReplaceText extends AbstractProcessor {
BufferedWriter bw = new BufferedWriter(new OutputStreamWriter(out, charset));) {
String oneLine;
while (null != (oneLine = br.readLine())) {
- final String updatedValue = oneLine.
- replaceAll(regex, replacementValue);
+ final String updatedValue = oneLine.replaceAll(regex, replacementValue);
bw.write(updatedValue);
}
}
@@ -315,9 +288,7 @@ public class ReplaceText extends AbstractProcessor {
}
logger.info("Transferred {} to 'success'", new Object[]{flowFile});
- session.getProvenanceReporter().
- modifyContent(flowFile, stopWatch.
- getElapsed(TimeUnit.MILLISECONDS));
+ session.getProvenanceReporter().modifyContent(flowFile, stopWatch.getElapsed(TimeUnit.MILLISECONDS));
session.transfer(flowFile, REL_SUCCESS);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/d29a2d68/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceTextWithMapping.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceTextWithMapping.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceTextWithMapping.java
index a8a2919..5be2b69 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceTextWithMapping.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ReplaceTextWithMapping.java
@@ -71,62 +71,60 @@ import org.apache.commons.lang3.StringUtils;
@CapabilityDescription("Updates the content of a FlowFile by evaluating a Regular Expression against it and replacing the section of the content that matches the Regular Expression with some alternate value provided in a mapping file.")
public class ReplaceTextWithMapping extends AbstractProcessor {
- public static final PropertyDescriptor REGEX = new PropertyDescriptor.Builder().
- name("Regular Expression").
- description("The Regular Expression to search for in the FlowFile content").
- required(true).
- addValidator(StandardValidators.
- createRegexValidator(0, Integer.MAX_VALUE, true)).
- expressionLanguageSupported(true).
- defaultValue("\\S+").
- build();
- public static final PropertyDescriptor MATCHING_GROUP_FOR_LOOKUP_KEY = new PropertyDescriptor.Builder().
- name("Matching Group").
- description("The number of the matching group of the provided regex to replace with the corresponding value from the mapping file (if it exists).").
- addValidator(StandardValidators.INTEGER_VALIDATOR).
- required(true).
- expressionLanguageSupported(true).
- defaultValue("0").
- build();
- public static final PropertyDescriptor MAPPING_FILE = new PropertyDescriptor.Builder().
- name("Mapping File").
- description("The name of the file (including the full path) containing the Mappings.").
- addValidator(StandardValidators.FILE_EXISTS_VALIDATOR).
- required(true).
- build();
- public static final PropertyDescriptor MAPPING_FILE_REFRESH_INTERVAL = new PropertyDescriptor.Builder().
- name("Mapping File Refresh Interval").
- description("The polling interval in seconds to check for updates to the mapping file. The default is 60s.").
- addValidator(StandardValidators.TIME_PERIOD_VALIDATOR).
- required(true).
- defaultValue("60s").
- build();
- public static final PropertyDescriptor CHARACTER_SET = new PropertyDescriptor.Builder().
- name("Character Set").
- description("The Character Set in which the file is encoded").
- required(true).
- addValidator(StandardValidators.CHARACTER_SET_VALIDATOR).
- defaultValue("UTF-8").
- build();
- 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) in order to apply the regular expressions. If a FlowFile is larger than this value, the FlowFile will be routed to 'failure'").
- required(true).
- addValidator(StandardValidators.DATA_SIZE_VALIDATOR).
- defaultValue("1 MB").
- build();
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder().
- name("success").
- description("FlowFiles that have been successfully updated are routed to this relationship, as well as FlowFiles whose content does not match the given Regular Expression").
- build();
- public static final Relationship REL_FAILURE = new Relationship.Builder().
- name("failure").
- description("FlowFiles that could not be updated are routed to this relationship").
- build();
-
- private final Pattern backReferencePattern = Pattern.
- compile("[^\\\\]\\$(\\d+)");
+ public static final PropertyDescriptor REGEX = new PropertyDescriptor.Builder()
+ .name("Regular Expression")
+ .description("The Regular Expression to search for in the FlowFile content")
+ .required(true)
+ .addValidator(StandardValidators.createRegexValidator(0, Integer.MAX_VALUE, true))
+ .expressionLanguageSupported(true)
+ .defaultValue("\\S+")
+ .build();
+ public static final PropertyDescriptor MATCHING_GROUP_FOR_LOOKUP_KEY = new PropertyDescriptor.Builder()
+ .name("Matching Group")
+ .description("The number of the matching group of the provided regex to replace with the corresponding value from the mapping file (if it exists).")
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .required(true)
+ .expressionLanguageSupported(true)
+ .defaultValue("0")
+ .build();
+ public static final PropertyDescriptor MAPPING_FILE = new PropertyDescriptor.Builder()
+ .name("Mapping File")
+ .description("The name of the file (including the full path) containing the Mappings.")
+ .addValidator(StandardValidators.FILE_EXISTS_VALIDATOR)
+ .required(true)
+ .build();
+ public static final PropertyDescriptor MAPPING_FILE_REFRESH_INTERVAL = new PropertyDescriptor.Builder()
+ .name("Mapping File Refresh Interval")
+ .description("The polling interval in seconds to check for updates to the mapping file. The default is 60s.")
+ .addValidator(StandardValidators.TIME_PERIOD_VALIDATOR)
+ .required(true)
+ .defaultValue("60s")
+ .build();
+ public static final PropertyDescriptor CHARACTER_SET = new PropertyDescriptor.Builder()
+ .name("Character Set")
+ .description("The Character Set in which the file is encoded")
+ .required(true)
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .defaultValue("UTF-8")
+ .build();
+ 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) in order to apply the regular expressions. If a FlowFile is larger than this value, the FlowFile will be routed to 'failure'")
+ .required(true)
+ .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
+ .defaultValue("1 MB")
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("FlowFiles that have been successfully updated are routed to this relationship, as well as FlowFiles whose content does not match the given Regular Expression")
+ .build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("FlowFiles that could not be updated are routed to this relationship")
+ .build();
+
+ private final Pattern backReferencePattern = Pattern.compile("[^\\\\]\\$(\\d+)");
private List<PropertyDescriptor> properties;
private Set<Relationship> relationships;
@@ -134,31 +132,23 @@ public class ReplaceTextWithMapping extends AbstractProcessor {
private final ReentrantLock processorLock = new ReentrantLock();
private final AtomicLong lastModified = new AtomicLong(0L);
final AtomicLong mappingTestTime = new AtomicLong(0);
- private final AtomicReference<ConfigurationState> configurationStateRef = new AtomicReference<>(
- new ConfigurationState(null));
+ private final AtomicReference<ConfigurationState> configurationStateRef = new AtomicReference<>(new ConfigurationState(null));
@Override
protected Collection<ValidationResult> customValidate(final ValidationContext context) {
- final List<ValidationResult> errors = new ArrayList<>(super.
- customValidate(context));
-
- final String regexValue = context.getProperty(REGEX).
- evaluateAttributeExpressions().
- getValue();
- final int numCapturingGroups = Pattern.compile(regexValue).
- matcher("").
- groupCount();
- final int groupToMatch = context.
- getProperty(MATCHING_GROUP_FOR_LOOKUP_KEY).
- evaluateAttributeExpressions().
- asInteger();
+ final List<ValidationResult> errors = new ArrayList<>(super.customValidate(context));
+
+ final String regexValue = context.getProperty(REGEX).evaluateAttributeExpressions().getValue();
+ final int numCapturingGroups = Pattern.compile(regexValue).matcher("").groupCount();
+ final int groupToMatch = context.getProperty(MATCHING_GROUP_FOR_LOOKUP_KEY).evaluateAttributeExpressions().asInteger();
if (groupToMatch > numCapturingGroups) {
- errors.add(new ValidationResult.Builder().
- subject("Insufficient Matching Groups").
- valid(false).
- explanation("The specified matching group does not exist for the regular expression provided").
- build());
+ errors.add(
+ new ValidationResult.Builder()
+ .subject("Insufficient Matching Groups")
+ .valid(false)
+ .explanation("The specified matching group does not exist for the regular expression provided")
+ .build());
}
return errors;
}
@@ -200,9 +190,7 @@ public class ReplaceTextWithMapping extends AbstractProcessor {
final ProcessorLog logger = getLogger();
- final int maxBufferSize = context.getProperty(MAX_BUFFER_SIZE).
- asDataSize(DataUnit.B).
- intValue();
+ final int maxBufferSize = context.getProperty(MAX_BUFFER_SIZE).asDataSize(DataUnit.B).intValue();
for (FlowFile flowFile : flowFiles) {
if (flowFile.getSize() > maxBufferSize) {
@@ -212,13 +200,10 @@ public class ReplaceTextWithMapping extends AbstractProcessor {
final StopWatch stopWatch = new StopWatch(true);
- flowFile = session.
- write(flowFile, new ReplaceTextCallback(context, flowFile, maxBufferSize));
+ flowFile = session.write(flowFile, new ReplaceTextCallback(context, flowFile, maxBufferSize));
logger.info("Transferred {} to 'success'", new Object[]{flowFile});
- session.getProvenanceReporter().
- modifyContent(flowFile, stopWatch.
- getElapsed(TimeUnit.MILLISECONDS));
+ session.getProvenanceReporter().modifyContent(flowFile, stopWatch.getElapsed(TimeUnit.MILLISECONDS));
session.transfer(flowFile, REL_SUCCESS);
}
}
@@ -252,42 +237,33 @@ public class ReplaceTextWithMapping extends AbstractProcessor {
// if not queried mapping file lastUpdate time in
// mapppingRefreshPeriodSecs, do so.
long currentTimeSecs = System.currentTimeMillis() / 1000;
- long mappingRefreshPeriodSecs = context.
- getProperty(MAPPING_FILE_REFRESH_INTERVAL).
- asTimePeriod(TimeUnit.SECONDS);
+ long mappingRefreshPeriodSecs = context.getProperty(MAPPING_FILE_REFRESH_INTERVAL).asTimePeriod(TimeUnit.SECONDS);
boolean retry = (currentTimeSecs > (mappingTestTime.get() + mappingRefreshPeriodSecs));
if (retry) {
mappingTestTime.set(System.currentTimeMillis() / 1000);
// see if the mapping file needs to be reloaded
- final String fileName = context.getProperty(MAPPING_FILE).
- getValue();
+ final String fileName = context.getProperty(MAPPING_FILE).getValue();
final File file = new File(fileName);
if (file.exists() && file.isFile() && file.canRead()) {
if (file.lastModified() > lastModified.get()) {
lastModified.getAndSet(file.lastModified());
try (FileInputStream is = new FileInputStream(file)) {
- logger.
- info("Reloading mapping file: {}", new Object[]{fileName});
+ logger.info("Reloading mapping file: {}", new Object[]{fileName});
final Map<String, String> mapping = loadMappingFile(is);
- final ConfigurationState newState = new ConfigurationState(
- mapping);
+ final ConfigurationState newState = new ConfigurationState(mapping);
configurationStateRef.set(newState);
} catch (IOException e) {
- logger.
- error("Error reading mapping file: {}", new Object[]{e.
- getMessage()});
+ logger.error("Error reading mapping file: {}", new Object[]{e.getMessage()});
}
}
} else {
- logger.
- error("Mapping file does not exist or is not readable: {}", new Object[]{fileName});
+ logger.error("Mapping file does not exist or is not readable: {}", new Object[]{fileName});
}
}
} catch (Exception e) {
- logger.error("Error loading mapping file: {}", new Object[]{e.
- getMessage()});
+ logger.error("Error loading mapping file: {}", new Object[]{e.getMessage()});
} finally {
processorLock.unlock();
}
@@ -354,34 +330,23 @@ public class ReplaceTextWithMapping extends AbstractProcessor {
};
private ReplaceTextCallback(ProcessContext context, FlowFile flowFile, int maxBufferSize) {
- this.regex = context.getProperty(REGEX).
- evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).
- getValue();
+ this.regex = context.getProperty(REGEX).evaluateAttributeExpressions(flowFile, quotedAttributeDecorator).getValue();
this.flowFile = flowFile;
- this.charset = Charset.forName(context.getProperty(CHARACTER_SET).
- getValue());
+ this.charset = Charset.forName(context.getProperty(CHARACTER_SET).getValue());
- final String regexValue = context.getProperty(REGEX).
- evaluateAttributeExpressions().
- getValue();
- this.numCapturingGroups = Pattern.compile(regexValue).
- matcher("").
- groupCount();
+ final String regexValue = context.getProperty(REGEX).evaluateAttributeExpressions().getValue();
+ this.numCapturingGroups = Pattern.compile(regexValue).matcher("").groupCount();
this.buffer = new byte[maxBufferSize];
- this.groupToMatch = context.
- getProperty(MATCHING_GROUP_FOR_LOOKUP_KEY).
- evaluateAttributeExpressions().
- asInteger();
+ this.groupToMatch = context.getProperty(MATCHING_GROUP_FOR_LOOKUP_KEY).evaluateAttributeExpressions().asInteger();
}
@Override
public void process(final InputStream in, final OutputStream out) throws IOException {
- final Map<String, String> mapping = configurationStateRef.get().
- getMapping();
+ final Map<String, String> mapping = configurationStateRef.get().getMapping();
StreamUtils.fillBuffer(in, buffer, false);
@@ -389,8 +354,7 @@ public class ReplaceTextWithMapping extends AbstractProcessor {
final String contentString = new String(buffer, 0, flowFileSize, charset);
- final Matcher matcher = Pattern.compile(regex).
- matcher(contentString);
+ final Matcher matcher = Pattern.compile(regex).matcher(contentString);
matcher.reset();
boolean result = matcher.find();
@@ -401,37 +365,26 @@ public class ReplaceTextWithMapping extends AbstractProcessor {
String rv = mapping.get(matched);
if (rv == null) {
- String replacement = matcher.group().
- replace("$", "\\$");
+ String replacement = matcher.group().replace("$", "\\$");
matcher.appendReplacement(sb, replacement);
} else {
String allRegexMatched = matcher.group(); //this is everything that matched the regex
- int scaledStart = matcher.start(groupToMatch) - matcher.
- start();
- int scaledEnd = scaledStart + matcher.
- group(groupToMatch).
- length();
+ int scaledStart = matcher.start(groupToMatch) - matcher.start();
+ int scaledEnd = scaledStart + matcher.group(groupToMatch).length();
StringBuilder replacementBuilder = new StringBuilder();
- replacementBuilder.append(allRegexMatched.
- substring(0, scaledStart).
- replace("$", "\\$"));
- replacementBuilder.
- append(fillReplacementValueBackReferences(rv, numCapturingGroups));
- replacementBuilder.append(allRegexMatched.
- substring(scaledEnd).
- replace("$", "\\$"));
-
- matcher.appendReplacement(sb, replacementBuilder.
- toString());
+ replacementBuilder.append(allRegexMatched.substring(0, scaledStart).replace("$", "\\$"));
+ replacementBuilder.append(fillReplacementValueBackReferences(rv, numCapturingGroups));
+ replacementBuilder.append(allRegexMatched.substring(scaledEnd).replace("$", "\\$"));
+
+ matcher.appendReplacement(sb, replacementBuilder.toString());
}
result = matcher.find();
} while (result);
matcher.appendTail(sb);
- out.write(sb.toString().
- getBytes(charset));
+ out.write(sb.toString().getBytes(charset));
return;
}
out.write(contentString.getBytes(charset));
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/d29a2d68/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnAttribute.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnAttribute.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnAttribute.java
index ff231d7..8b6a7b4 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnAttribute.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnAttribute.java
@@ -48,16 +48,10 @@ import org.apache.nifi.processor.util.StandardValidators;
/**
* <p>
- * This processor routes a FlowFile based on its flow file attributes by using
- * the Attribute Expression Language. The Expression Language is used by adding
- * Optional Properties to the processor. The name of the Property indicates the
- * name of the relationship to which a FlowFile will be routed if matched. The
- * value of the Property indicates an Attribute Expression Language Expression
- * that will be used to determine whether or not a given FlowFile will be routed
- * to the associated relationship. If multiple expressions match a FlowFile's
- * attributes, that FlowFile will be cloned and routed to each corresponding
- * relationship. If none of the supplied expressions matches for a given
- * FlowFile, that FlowFile will be routed to the 'unmatched' relationship.
+ * This processor routes a FlowFile based on its flow file attributes by using the Attribute Expression Language. The Expression Language is used by adding Optional Properties to the processor. The
+ * name of the Property indicates the name of the relationship to which a FlowFile will be routed if matched. The value of the Property indicates an Attribute Expression Language Expression that will
+ * be used to determine whether or not a given FlowFile will be routed to the associated relationship. If multiple expressions match a FlowFile's attributes, that FlowFile will be cloned and routed to
+ * each corresponding relationship. If none of the supplied expressions matches for a given FlowFile, that FlowFile will be routed to the 'unmatched' relationship.
* </p>
*
* @author unattributed
@@ -79,43 +73,34 @@ public class RouteOnAttribute extends AbstractProcessor {
private static final String routeAnyMatches = "Route to 'match' if any matches";
private static final String routePropertyNameValue = "Route to Property name";
- public static final AllowableValue ROUTE_PROPERTY_NAME = new AllowableValue(
- routePropertyNameValue,
- "Route to Property name",
- "A copy of the FlowFile will be routed to each relationship whose corresponding expression evaluates to 'true'"
- );
- public static final AllowableValue ROUTE_ALL_MATCH = new AllowableValue(
- routeAllMatchValue,
- "Route to 'matched' if all match",
- "Requires that all user-defined expressions evaluate to 'true' for the FlowFile to be considered a match"
- );
- public static final AllowableValue ROUTE_ANY_MATCHES = new AllowableValue(
- routeAnyMatches, // keep the word 'match' instead of 'matched' to maintain backward compatibility (there was a typo originally)
+ public static final AllowableValue ROUTE_PROPERTY_NAME = new AllowableValue(routePropertyNameValue, "Route to Property name",
+ "A copy of the FlowFile will be routed to each relationship whose corresponding expression evaluates to 'true'");
+ public static final AllowableValue ROUTE_ALL_MATCH = new AllowableValue(routeAllMatchValue, "Route to 'matched' if all match",
+ "Requires that all user-defined expressions evaluate to 'true' for the FlowFile to be considered a match");
+ public static final AllowableValue ROUTE_ANY_MATCHES = new AllowableValue(routeAnyMatches, // keep the word 'match' instead of 'matched' to maintain backward compatibility (there was a typo originally)
"Route to 'matched' if any matches",
- "Requires that at least one user-defined expression evaluate to 'true' for hte FlowFile to be considered a match"
- );
+ "Requires that at least one user-defined expression evaluate to 'true' for hte FlowFile to be considered a match");
- public static final PropertyDescriptor ROUTE_STRATEGY = new PropertyDescriptor.Builder().
- name("Routing Strategy").
- description("Specifies how to determine which relationship to use when evaluating the Expression Language").
- required(true).
- allowableValues(ROUTE_PROPERTY_NAME, ROUTE_ALL_MATCH, ROUTE_ANY_MATCHES).
- defaultValue(ROUTE_PROPERTY_NAME.getValue()).
- build();
+ public static final PropertyDescriptor ROUTE_STRATEGY = new PropertyDescriptor.Builder()
+ .name("Routing Strategy")
+ .description("Specifies how to determine which relationship to use when evaluating the Expression Language")
+ .required(true)
+ .allowableValues(ROUTE_PROPERTY_NAME, ROUTE_ALL_MATCH, ROUTE_ANY_MATCHES)
+ .defaultValue(ROUTE_PROPERTY_NAME.getValue())
+ .build();
public static final Relationship REL_NO_MATCH = new Relationship.Builder()
- .name("unmatched").
- description("FlowFiles that do not match any user-define expression will be routed here").
- build();
+ .name("unmatched")
+ .description("FlowFiles that do not match any user-define expression will be routed here")
+ .build();
public static final Relationship REL_MATCH = new Relationship.Builder()
- .name("matched").
- description("FlowFiles will be routed to 'match' if one or all Expressions match, depending on the configuration of the Routing Strategy property").
- build();
+ .name("matched")
+ .description("FlowFiles will be routed to 'match' if one or all Expressions match, depending on the configuration of the Routing Strategy property")
+ .build();
private AtomicReference<Set<Relationship>> relationships = new AtomicReference<>();
private List<PropertyDescriptor> properties;
- private volatile String configuredRouteStrategy = ROUTE_STRATEGY.
- getDefaultValue();
+ private volatile String configuredRouteStrategy = ROUTE_STRATEGY.getDefaultValue();
private volatile Set<String> dynamicPropertyNames = new HashSet<>();
@Override
@@ -142,13 +127,12 @@ public class RouteOnAttribute extends AbstractProcessor {
@Override
protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
return new PropertyDescriptor.Builder()
- .required(false).
- name(propertyDescriptorName).
- addValidator(StandardValidators.
- createAttributeExpressionLanguageValidator(ResultType.BOOLEAN, false)).
- dynamic(true).
- expressionLanguageSupported(true).
- build();
+ .required(false)
+ .name(propertyDescriptorName)
+ .addValidator(StandardValidators.createAttributeExpressionLanguageValidator(ResultType.BOOLEAN, false))
+ .dynamic(true)
+ .expressionLanguageSupported(true)
+ .build();
}
@Override
@@ -163,8 +147,7 @@ public class RouteOnAttribute extends AbstractProcessor {
newDynamicPropertyNames.add(descriptor.getName());
}
- this.dynamicPropertyNames = Collections.
- unmodifiableSet(newDynamicPropertyNames);
+ this.dynamicPropertyNames = Collections.unmodifiableSet(newDynamicPropertyNames);
}
// formulate the new set of Relationships
@@ -173,8 +156,7 @@ public class RouteOnAttribute extends AbstractProcessor {
final String routeStrategy = configuredRouteStrategy;
if (ROUTE_PROPERTY_NAME.equals(routeStrategy)) {
for (final String propName : allDynamicProps) {
- newRelationships.add(new Relationship.Builder().name(propName).
- build());
+ newRelationships.add(new Relationship.Builder().name(propName).build());
}
} else {
newRelationships.add(REL_MATCH);
@@ -193,32 +175,26 @@ public class RouteOnAttribute extends AbstractProcessor {
final ProcessorLog logger = getLogger();
final Map<Relationship, PropertyValue> propertyMap = new HashMap<>();
- for (final PropertyDescriptor descriptor : context.getProperties().
- keySet()) {
+ for (final PropertyDescriptor descriptor : context.getProperties().keySet()) {
if (!descriptor.isDynamic()) {
continue;
}
- propertyMap.put(new Relationship.Builder().
- name(descriptor.getName()).
- build(), context.getProperty(descriptor));
+ propertyMap.put(new Relationship.Builder().name(descriptor.getName()).build(), context.getProperty(descriptor));
}
final Set<Relationship> matchingRelationships = new HashSet<>();
- for (final Map.Entry<Relationship, PropertyValue> entry : propertyMap.
- entrySet()) {
+ for (final Map.Entry<Relationship, PropertyValue> entry : propertyMap.entrySet()) {
final PropertyValue value = entry.getValue();
- final boolean matches = value.evaluateAttributeExpressions(flowFile).
- asBoolean();
+ final boolean matches = value.evaluateAttributeExpressions(flowFile).asBoolean();
if (matches) {
matchingRelationships.add(entry.getKey());
}
}
final Set<Relationship> destinationRelationships = new HashSet<>();
- switch (context.getProperty(ROUTE_STRATEGY).
- getValue()) {
+ switch (context.getProperty(ROUTE_STRATEGY).getValue()) {
case routeAllMatchValue:
if (matchingRelationships.size() == propertyMap.size()) {
destinationRelationships.add(REL_MATCH);
@@ -241,52 +217,36 @@ public class RouteOnAttribute extends AbstractProcessor {
if (destinationRelationships.isEmpty()) {
logger.info(this + " routing " + flowFile + " to unmatched");
- flowFile = session.
- putAttribute(flowFile, ROUTE_ATTRIBUTE_KEY, REL_NO_MATCH.
- getName());
- session.getProvenanceReporter().
- route(flowFile, REL_NO_MATCH);
+ flowFile = session.putAttribute(flowFile, ROUTE_ATTRIBUTE_KEY, REL_NO_MATCH.getName());
+ session.getProvenanceReporter().route(flowFile, REL_NO_MATCH);
session.transfer(flowFile, REL_NO_MATCH);
} else {
- final Iterator<Relationship> relationshipNameIterator = destinationRelationships.
- iterator();
- final Relationship firstRelationship = relationshipNameIterator.
- next();
+ final Iterator<Relationship> relationshipNameIterator = destinationRelationships.iterator();
+ final Relationship firstRelationship = relationshipNameIterator.next();
final Map<Relationship, FlowFile> transferMap = new HashMap<>();
final Set<FlowFile> clones = new HashSet<>();
// make all the clones for any remaining relationships
while (relationshipNameIterator.hasNext()) {
- final Relationship relationship = relationshipNameIterator.
- next();
+ final Relationship relationship = relationshipNameIterator.next();
final FlowFile cloneFlowFile = session.clone(flowFile);
clones.add(cloneFlowFile);
transferMap.put(relationship, cloneFlowFile);
}
// now transfer any clones generated
- for (final Map.Entry<Relationship, FlowFile> entry : transferMap.
- entrySet()) {
- logger.info(this + " cloned " + flowFile + " into " + entry.
- getValue() + " and routing clone to relationship " + entry.
- getKey());
- FlowFile updatedFlowFile = session.
- putAttribute(entry.getValue(), ROUTE_ATTRIBUTE_KEY, entry.
- getKey().
- getName());
- session.getProvenanceReporter().
- route(updatedFlowFile, entry.getKey());
+ for (final Map.Entry<Relationship, FlowFile> entry : transferMap.entrySet()) {
+ logger.info(this + " cloned " + flowFile + " into " + entry.getValue() + " and routing clone to relationship " + entry.getKey());
+ FlowFile updatedFlowFile = session.putAttribute(entry.getValue(), ROUTE_ATTRIBUTE_KEY, entry.getKey().getName());
+ session.getProvenanceReporter().route(updatedFlowFile, entry.getKey());
session.transfer(updatedFlowFile, entry.getKey());
}
//now transfer the original flow file
logger.
info("Routing {} to {}", new Object[]{flowFile, firstRelationship});
- session.getProvenanceReporter().
- route(flowFile, firstRelationship);
- flowFile = session.
- putAttribute(flowFile, ROUTE_ATTRIBUTE_KEY, firstRelationship.
- getName());
+ session.getProvenanceReporter().route(flowFile, firstRelationship);
+ flowFile = session.putAttribute(flowFile, ROUTE_ATTRIBUTE_KEY, firstRelationship.getName());
session.transfer(flowFile, firstRelationship);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/d29a2d68/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnContent.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnContent.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnContent.java
index 8f1eb4e..937bc69 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnContent.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/RouteOnContent.java
@@ -70,34 +70,34 @@ public class RouteOnContent extends AbstractProcessor {
public static final String MATCH_ALL = "content must match exactly";
public static final String MATCH_SUBSEQUENCE = "content must contain match";
- public static final PropertyDescriptor BUFFER_SIZE = new PropertyDescriptor.Builder().
- name("Content Buffer Size").
- description("Specifies the maximum amount of data to buffer in order to apply the regular expressions. If the size of the FlowFile "
- + "exceeds this value, any amount of this value will be ignored").
- required(true).
- addValidator(StandardValidators.DATA_SIZE_VALIDATOR).
- defaultValue("1 MB").
- build();
- public static final PropertyDescriptor MATCH_REQUIREMENT = new PropertyDescriptor.Builder().
- name("Match Requirement").
- description("Specifies whether the entire content of the file must match the regular expression exactly, or if any part of the file "
- + "(up to Content Buffer Size) can contain the regular expression in order to be considered a match").
- required(true).
- allowableValues(MATCH_ALL, MATCH_SUBSEQUENCE).
- defaultValue(MATCH_ALL).
- build();
- public static final PropertyDescriptor CHARACTER_SET = new PropertyDescriptor.Builder().
- name("Character Set").
- description("The Character Set in which the file is encoded").
- required(true).
- addValidator(StandardValidators.CHARACTER_SET_VALIDATOR).
- defaultValue("UTF-8").
- build();
-
- public static final Relationship REL_NO_MATCH = new Relationship.Builder().
- name("unmatched").
- description("FlowFiles that do not match any of the user-supplied regular expressions will be routed to this relationship").
- build();
+ public static final PropertyDescriptor BUFFER_SIZE = new PropertyDescriptor.Builder()
+ .name("Content Buffer Size")
+ .description("Specifies the maximum amount of data to buffer in order to apply the regular expressions. If the size of the FlowFile "
+ + "exceeds this value, any amount of this value will be ignored")
+ .required(true)
+ .addValidator(StandardValidators.DATA_SIZE_VALIDATOR)
+ .defaultValue("1 MB")
+ .build();
+ public static final PropertyDescriptor MATCH_REQUIREMENT = new PropertyDescriptor.Builder()
+ .name("Match Requirement")
+ .description("Specifies whether the entire content of the file must match the regular expression exactly, or if any part of the file "
+ + "(up to Content Buffer Size) can contain the regular expression in order to be considered a match")
+ .required(true)
+ .allowableValues(MATCH_ALL, MATCH_SUBSEQUENCE)
+ .defaultValue(MATCH_ALL)
+ .build();
+ public static final PropertyDescriptor CHARACTER_SET = new PropertyDescriptor.Builder()
+ .name("Character Set")
+ .description("The Character Set in which the file is encoded")
+ .required(true)
+ .addValidator(StandardValidators.CHARACTER_SET_VALIDATOR)
+ .defaultValue("UTF-8")
+ .build();
+
+ public static final Relationship REL_NO_MATCH = new Relationship.Builder()
+ .name("unmatched")
+ .description("FlowFiles that do not match any of the user-supplied regular expressions will be routed to this relationship")
+ .build();
private final AtomicReference<Set<Relationship>> relationships = new AtomicReference<>();
private List<PropertyDescriptor> properties;
@@ -132,23 +132,19 @@ public class RouteOnContent extends AbstractProcessor {
}
return new PropertyDescriptor.Builder()
- .required(false).
- name(propertyDescriptorName).
- addValidator(StandardValidators.
- createRegexValidator(0, Integer.MAX_VALUE, true)).
- dynamic(true).
- expressionLanguageSupported(true).
- build();
+ .required(false)
+ .name(propertyDescriptorName)
+ .addValidator(StandardValidators.createRegexValidator(0, Integer.MAX_VALUE, true))
+ .dynamic(true)
+ .expressionLanguageSupported(true)
+ .build();
}
@Override
public void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue) {
if (descriptor.isDynamic()) {
- final Set<Relationship> relationships = new HashSet<>(this.relationships.
- get());
- final Relationship relationship = new Relationship.Builder().
- name(descriptor.getName()).
- build();
+ final Set<Relationship> relationships = new HashSet<>(this.relationships.get());
+ final Relationship relationship = new Relationship.Builder().name(descriptor.getName()).build();
if (newValue == null) {
relationships.remove(relationship);
@@ -170,20 +166,15 @@ public class RouteOnContent extends AbstractProcessor {
final AttributeValueDecorator quoteDecorator = new AttributeValueDecorator() {
@Override
public String decorate(final String attributeValue) {
- return (attributeValue == null) ? null : Pattern.
- quote(attributeValue);
+ return (attributeValue == null) ? null : Pattern.quote(attributeValue);
}
};
final Map<FlowFile, Set<Relationship>> flowFileDestinationMap = new HashMap<>();
final ProcessorLog logger = getLogger();
- final Charset charset = Charset.forName(context.
- getProperty(CHARACTER_SET).
- getValue());
- final byte[] buffer = new byte[context.getProperty(BUFFER_SIZE).
- asDataSize(DataUnit.B).
- intValue()];
+ final Charset charset = Charset.forName(context.getProperty(CHARACTER_SET).getValue());
+ final byte[] buffer = new byte[context.getProperty(BUFFER_SIZE).asDataSize(DataUnit.B).intValue()];
for (final FlowFile flowFile : flowFiles) {
final Set<Relationship> destinations = new HashSet<>();
flowFileDestinationMap.put(flowFile, destinations);
@@ -192,82 +183,58 @@ public class RouteOnContent extends AbstractProcessor {
session.read(flowFile, new InputStreamCallback() {
@Override
public void process(final InputStream in) throws IOException {
- bufferedByteCount.set(StreamUtils.
- fillBuffer(in, buffer, false));
+ bufferedByteCount.set(StreamUtils.fillBuffer(in, buffer, false));
}
});
- final String contentString = new String(buffer, 0, bufferedByteCount.
- get(), charset);
+ final String contentString = new String(buffer, 0, bufferedByteCount.get(), charset);
- for (final PropertyDescriptor descriptor : context.getProperties().
- keySet()) {
+ for (final PropertyDescriptor descriptor : context.getProperties().keySet()) {
if (!descriptor.isDynamic()) {
continue;
}
- final String regex = context.getProperty(descriptor).
- evaluateAttributeExpressions(flowFile, quoteDecorator).
- getValue();
+ final String regex = context.getProperty(descriptor).evaluateAttributeExpressions(flowFile, quoteDecorator).getValue();
final Pattern pattern = Pattern.compile(regex);
final boolean matches;
- if (context.getProperty(MATCH_REQUIREMENT).
- getValue().
- equalsIgnoreCase(MATCH_ALL)) {
- matches = pattern.matcher(contentString).
- matches();
+ if (context.getProperty(MATCH_REQUIREMENT).getValue().equalsIgnoreCase(MATCH_ALL)) {
+ matches = pattern.matcher(contentString).matches();
} else {
- matches = pattern.matcher(contentString).
- find();
+ matches = pattern.matcher(contentString).find();
}
if (matches) {
- final Relationship relationship = new Relationship.Builder().
- name(descriptor.getName()).
- build();
+ final Relationship relationship = new Relationship.Builder().name(descriptor.getName()).build();
destinations.add(relationship);
}
}
}
- for (final Map.Entry<FlowFile, Set<Relationship>> entry : flowFileDestinationMap.
- entrySet()) {
+ for (final Map.Entry<FlowFile, Set<Relationship>> entry : flowFileDestinationMap.entrySet()) {
FlowFile flowFile = entry.getKey();
final Set<Relationship> destinations = entry.getValue();
if (destinations.isEmpty()) {
- flowFile = session.
- putAttribute(flowFile, ROUTE_ATTRIBUTE_KEY, REL_NO_MATCH.
- getName());
+ flowFile = session.putAttribute(flowFile, ROUTE_ATTRIBUTE_KEY, REL_NO_MATCH.getName());
session.transfer(flowFile, REL_NO_MATCH);
- session.getProvenanceReporter().
- route(flowFile, REL_NO_MATCH);
+ session.getProvenanceReporter().route(flowFile, REL_NO_MATCH);
logger.info("Routing {} to 'unmatched'", new Object[]{flowFile});
} else {
- final Relationship firstRelationship = destinations.iterator().
- next();
+ final Relationship firstRelationship = destinations.iterator().next();
destinations.remove(firstRelationship);
for (final Relationship relationship : destinations) {
FlowFile clone = session.clone(flowFile);
- clone = session.
- putAttribute(clone, ROUTE_ATTRIBUTE_KEY, relationship.
- getName());
- session.getProvenanceReporter().
- route(clone, relationship);
+ clone = session.putAttribute(clone, ROUTE_ATTRIBUTE_KEY, relationship.getName());
+ session.getProvenanceReporter().route(clone, relationship);
session.transfer(clone, relationship);
- logger.
- info("Cloning {} to {} and routing clone to {}", new Object[]{flowFile, clone, relationship});
+ logger.info("Cloning {} to {} and routing clone to {}", new Object[]{flowFile, clone, relationship});
}
- flowFile = session.
- putAttribute(flowFile, ROUTE_ATTRIBUTE_KEY, firstRelationship.
- getName());
- session.getProvenanceReporter().
- route(flowFile, firstRelationship);
+ flowFile = session.putAttribute(flowFile, ROUTE_ATTRIBUTE_KEY, firstRelationship.getName());
+ session.getProvenanceReporter().route(flowFile, firstRelationship);
session.transfer(flowFile, firstRelationship);
- logger.
- info("Routing {} to {}", new Object[]{flowFile, firstRelationship});
+ logger.info("Routing {} to {}", new Object[]{flowFile, firstRelationship});
}
}
}