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:46 UTC
[12/50] [abbrv] incubator-nifi git commit: NIFI-271
NIFI-271
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/d29a2d68
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/d29a2d68
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/d29a2d68
Branch: refs/heads/NIFI-292
Commit: d29a2d688e437bae42c12115768cdb038b7406c5
Parents: 5481889
Author: joewitt <jo...@apache.org>
Authored: Mon Apr 27 11:54:36 2015 -0400
Committer: joewitt <jo...@apache.org>
Committed: Mon Apr 27 11:54:36 2015 -0400
----------------------------------------------------------------------
.../standard/AbstractJsonPathProcessor.java | 11 +-
.../standard/ConvertCharacterSet.java | 2 +-
.../nifi/processors/standard/HashAttribute.java | 34 +-
.../nifi/processors/standard/PutEmail.java | 302 +++++++---------
.../apache/nifi/processors/standard/PutFTP.java | 24 +-
.../nifi/processors/standard/PutFile.java | 275 ++++++--------
.../processors/standard/PutFileTransfer.java | 140 +++-----
.../apache/nifi/processors/standard/PutJMS.java | 161 +++------
.../nifi/processors/standard/ReplaceText.java | 169 ++++-----
.../standard/ReplaceTextWithMapping.java | 231 +++++-------
.../processors/standard/RouteOnAttribute.java | 136 +++----
.../processors/standard/RouteOnContent.java | 147 +++-----
.../nifi/processors/standard/ScanAttribute.java | 121 +++----
.../nifi/processors/standard/ScanContent.java | 86 ++---
.../processors/standard/SegmentContent.java | 54 ++-
.../nifi/processors/standard/SplitContent.java | 125 +++----
.../nifi/processors/standard/SplitJson.java | 77 ++--
.../nifi/processors/standard/SplitText.java | 142 ++++----
.../nifi/processors/standard/SplitXml.java | 70 ++--
.../nifi/processors/standard/TransformXml.java | 87 ++---
.../nifi/processors/standard/UnpackContent.java | 215 +++++------
.../servlets/ContentAcknowledgmentServlet.java | 55 +--
.../standard/servlets/ListenHTTPServlet.java | 142 +++-----
.../nifi/processors/standard/util/Bin.java | 22 +-
.../processors/standard/util/BinManager.java | 41 +--
.../standard/util/DocumentReaderCallback.java | 6 +-
.../processors/standard/util/FTPTransfer.java | 351 +++++++-----------
.../nifi/processors/standard/util/FTPUtils.java | 95 ++---
.../nifi/processors/standard/util/FileInfo.java | 3 +-
.../processors/standard/util/FileTransfer.java | 356 +++++++++----------
.../processors/standard/util/JmsFactory.java | 128 +++----
.../processors/standard/util/JmsProperties.java | 256 ++++++-------
.../util/JsonPathExpressionValidator.java | 27 +-
.../standard/util/NLKBufferedReader.java | 14 +-
.../processors/standard/util/SFTPTransfer.java | 351 +++++++-----------
.../processors/standard/util/SFTPUtils.java | 167 ++++-----
.../standard/util/UDPStreamConsumer.java | 25 +-
.../util/ValidatingBase32InputStream.java | 3 +-
.../util/ValidatingBase64InputStream.java | 3 +-
.../standard/util/WrappedMessageConsumer.java | 9 +-
.../standard/util/WrappedMessageProducer.java | 9 +-
.../src/test/java/TestIngestAndUpdate.java | 3 +-
.../processors/standard/CaptureServlet.java | 3 +-
.../standard/RESTServiceContentModified.java | 15 +-
.../standard/TestBase64EncodeContent.java | 42 +--
.../standard/TestCompressContent.java | 85 ++---
.../processors/standard/TestControlRate.java | 3 +-
.../standard/TestConvertCharacterSet.java | 13 +-
.../standard/TestDetectDuplicate.java | 33 +-
.../processors/standard/TestDistributeLoad.java | 19 +-
.../processors/standard/TestEncodeContent.java | 66 ++--
.../processors/standard/TestEncryptContent.java | 30 +-
.../standard/TestEvaluateJsonPath.java | 219 ++++--------
.../processors/standard/TestEvaluateXPath.java | 106 ++----
.../processors/standard/TestEvaluateXQuery.java | 312 +++++-----------
.../processors/standard/TestExecuteProcess.java | 18 +-
.../standard/TestExecuteStreamCommand.java | 135 +++----
.../processors/standard/TestExtractText.java | 81 ++---
.../nifi/processors/standard/TestGetFile.java | 63 ++--
.../nifi/processors/standard/TestGetHTTP.java | 57 +--
.../processors/standard/TestGetJMSQueue.java | 63 ++--
.../standard/TestHandleHttpRequest.java | 19 +-
.../standard/TestHandleHttpResponse.java | 81 ++---
.../processors/standard/TestHashAttribute.java | 8 +-
.../processors/standard/TestHashContent.java | 5 +-
.../standard/TestIdentifyMimeType.java | 16 +-
.../processors/standard/TestInvokeHTTP.java | 137 ++-----
.../processors/standard/TestJmsConsumer.java | 88 ++---
.../nifi/processors/standard/TestListenUDP.java | 39 +-
.../processors/standard/TestMergeContent.java | 176 +++------
.../processors/standard/TestModifyBytes.java | 82 ++---
.../standard/TestMonitorActivity.java | 84 ++---
.../nifi/processors/standard/TestPostHTTP.java | 102 ++----
.../nifi/processors/standard/TestPutEmail.java | 17 +-
.../processors/standard/TestReplaceText.java | 81 ++---
.../standard/TestReplaceTextLineByLine.java | 204 ++++-------
.../standard/TestReplaceTextWithMapping.java | 316 +++++-----------
.../standard/TestRouteOnAttribute.java | 65 ++--
.../processors/standard/TestRouteOnContent.java | 19 +-
.../processors/standard/TestScanAttribute.java | 16 +-
.../processors/standard/TestScanContent.java | 34 +-
.../processors/standard/TestSegmentContent.java | 14 +-
.../nifi/processors/standard/TestServer.java | 21 +-
.../processors/standard/TestSplitContent.java | 196 ++++------
.../nifi/processors/standard/TestSplitJson.java | 91 ++---
.../nifi/processors/standard/TestSplitText.java | 80 ++---
.../nifi/processors/standard/TestSplitXml.java | 4 +-
.../processors/standard/TestTransformXml.java | 45 +--
.../processors/standard/TestUnpackContent.java | 131 +++----
.../processors/standard/TestValidateXml.java | 4 +-
.../standard/UserAgentTestingServlet.java | 1 -
91 files changed, 2933 insertions(+), 5281 deletions(-)
----------------------------------------------------------------------
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/AbstractJsonPathProcessor.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractJsonPathProcessor.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractJsonPathProcessor.java
index 9e77dab..d03240e 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractJsonPathProcessor.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/AbstractJsonPathProcessor.java
@@ -42,8 +42,7 @@ import java.util.Map;
import java.util.Objects;
/**
- * Provides common functionality used for processors interacting and
- * manipulating JSON data via JsonPath.
+ * Provides common functionality used for processors interacting and manipulating JSON data via JsonPath.
*
* @see <a href="http://json.org">http://json.org</a>
* @see
@@ -90,9 +89,8 @@ public abstract class AbstractJsonPathProcessor extends AbstractProcessor {
}
/**
- * Determines the context by which JsonSmartJsonProvider would treat the
- * value. {@link java.util.Map} and {@link java.util.List} objects can be
- * rendered as JSON elements, everything else is treated as a scalar.
+ * Determines the context by which JsonSmartJsonProvider would treat the value. {@link java.util.Map} and {@link java.util.List} objects can be rendered as JSON elements, everything else is
+ * treated as a scalar.
*
* @param obj item to be inspected if it is a scalar or a JSON element
* @return false, if the object is a supported type; true otherwise
@@ -131,8 +129,7 @@ public abstract class AbstractJsonPathProcessor extends AbstractProcessor {
abstract void cacheComputedValue(String subject, String input, JsonPath computedJsonPath);
/**
- * A hook for implementing classes to determine if a cached value is
- * stale for a compiled JsonPath represented by either a validation
+ * A hook for implementing classes to determine if a cached value is stale for a compiled JsonPath represented by either a validation
*/
abstract boolean isStale(String subject, String input);
}
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/ConvertCharacterSet.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertCharacterSet.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertCharacterSet.java
index c8d22d3..ec61370 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertCharacterSet.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/ConvertCharacterSet.java
@@ -159,7 +159,7 @@ public class ConvertCharacterSet extends AbstractProcessor {
});
session.getProvenanceReporter().modifyContent(flowFile, stopWatch.getElapsed(TimeUnit.MILLISECONDS));
- logger.info("successfully converted characters from {} to {} for {}",
+ logger.info("successfully converted characters from {} to {} for {}",
new Object[]{context.getProperty(INPUT_CHARSET).getValue(), context.getProperty(OUTPUT_CHARSET).getValue(), flowFile});
session.transfer(flowFile, REL_SUCCESS);
} catch (final Exception e) {
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/HashAttribute.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HashAttribute.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HashAttribute.java
index 9187aad..314f1c7 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HashAttribute.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/HashAttribute.java
@@ -50,30 +50,21 @@ import org.apache.nifi.processor.util.StandardValidators;
/**
* <p>
- * This processor identifies groups of user-specified flowfile attributes and
- * assigns a unique hash value to each group, recording this hash value in the
- * flowfile's attributes using a user-specified attribute key. The groups are
- * identified dynamically and preserved across application restarts. </p>
+ * This processor identifies groups of user-specified flowfile attributes and assigns a unique hash value to each group, recording this hash value in the flowfile's attributes using a user-specified
+ * attribute key. The groups are identified dynamically and preserved across application restarts. </p>
*
* <p>
- * The user must supply optional processor properties during runtime to
- * correctly configure this processor. The optional property key will be used as
- * the flowfile attribute key for attribute inspection. The value must be a
- * valid regular expression. This regular expression is evaluated against the
- * flowfile attribute values. If the regular expression contains a capturing
- * group, the value of that group will be used when comparing flow file
- * attributes. Otherwise, the original flow file attribute's value will be used
- * if and only if the value matches the given regular expression. </p>
+ * The user must supply optional processor properties during runtime to correctly configure this processor. The optional property key will be used as the flowfile attribute key for attribute
+ * inspection. The value must be a valid regular expression. This regular expression is evaluated against the flowfile attribute values. If the regular expression contains a capturing group, the value
+ * of that group will be used when comparing flow file attributes. Otherwise, the original flow file attribute's value will be used if and only if the value matches the given regular expression. </p>
*
* <p>
- * If a flowfile does not have an attribute entry for one or more processor
- * configured values, then the flowfile is routed to failure. </p>
+ * If a flowfile does not have an attribute entry for one or more processor configured values, then the flowfile is routed to failure. </p>
*
* <p>
* An example hash value identification:
*
- * Assume Processor Configured with Two Properties ("MDKey1" = ".*" and "MDKey2"
- * = "(.).*").
+ * Assume Processor Configured with Two Properties ("MDKey1" = ".*" and "MDKey2" = "(.).*").
*
* FlowFile 1 has the following attributes: MDKey1 = a MDKey2 = b
*
@@ -89,17 +80,12 @@ import org.apache.nifi.processor.util.StandardValidators;
*
* FlowFile 4 has the following attribute: MDKey1 = a MDKey2 = bad
*
- * and will be assigned to group 1 (because the value of MDKey1 has the regular
- * expression ".*" applied to it, and that evaluates to the same as MDKey1
- * attribute of the first flow file. Similarly, the capturing group for the
- * MDKey2 property indicates that only the first character of the MDKey2
- * attribute must match, and the first character of MDKey2 for Flow File 1 and
- * Flow File 4 are both 'b'.)
+ * and will be assigned to group 1 (because the value of MDKey1 has the regular expression ".*" applied to it, and that evaluates to the same as MDKey1 attribute of the first flow file. Similarly, the
+ * capturing group for the MDKey2 property indicates that only the first character of the MDKey2 attribute must match, and the first character of MDKey2 for Flow File 1 and Flow File 4 are both 'b'.)
*
* FlowFile 5 has the following attributes: MDKey1 = a
*
- * and will route to failure because it does not have MDKey2 entry in its
- * attribute
+ * and will route to failure because it does not have MDKey2 entry in its attribute
* </p>
*
* <p>
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/PutEmail.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutEmail.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutEmail.java
index 8cad06f..8efc563 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutEmail.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutEmail.java
@@ -124,96 +124,95 @@ public class PutEmail extends AbstractProcessor {
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
.defaultValue("javax.net.ssl.SSLSocketFactory")
.build();
- public static final PropertyDescriptor HEADER_XMAILER = new PropertyDescriptor.Builder().
- name("SMTP X-Mailer Header").
- description("X-Mailer used in the header of the outgoing email").
- required(true).
- expressionLanguageSupported(true).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- defaultValue("NiFi").
- build();
- public static final PropertyDescriptor CONTENT_TYPE = new PropertyDescriptor.Builder().
- name("Content Type").
- description("Mime Type used to interpret the contents of the email, such as text/plain or text/html").
- required(true).
- expressionLanguageSupported(true).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- defaultValue("text/plain").
- build();
- public static final PropertyDescriptor FROM = new PropertyDescriptor.Builder().
- name("From").
- description("Specifies the Email address to use as the sender").
- required(true).
- expressionLanguageSupported(true).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- build();
+ public static final PropertyDescriptor HEADER_XMAILER = new PropertyDescriptor.Builder()
+ .name("SMTP X-Mailer Header")
+ .description("X-Mailer used in the header of the outgoing email")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .defaultValue("NiFi")
+ .build();
+ public static final PropertyDescriptor CONTENT_TYPE = new PropertyDescriptor.Builder()
+ .name("Content Type")
+ .description("Mime Type used to interpret the contents of the email, such as text/plain or text/html")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .defaultValue("text/plain")
+ .build();
+ public static final PropertyDescriptor FROM = new PropertyDescriptor.Builder()
+ .name("From")
+ .description("Specifies the Email address to use as the sender")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
public static final PropertyDescriptor TO = new PropertyDescriptor.Builder()
- .name("To").
- description("The recipients to include in the To-Line of the email").
- required(false).
- expressionLanguageSupported(true).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- build();
+ .name("To")
+ .description("The recipients to include in the To-Line of the email")
+ .required(false)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
public static final PropertyDescriptor CC = new PropertyDescriptor.Builder()
- .name("CC").
- description("The recipients to include in the CC-Line of the email").
- required(false).
- expressionLanguageSupported(true).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- build();
- public static final PropertyDescriptor BCC = new PropertyDescriptor.Builder().
- name("BCC").
- description("The recipients to include in the BCC-Line of the email").
- required(false).
- expressionLanguageSupported(true).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- build();
- public static final PropertyDescriptor SUBJECT = new PropertyDescriptor.Builder().
- name("Subject").
- description("The email subject").
- required(true).
- expressionLanguageSupported(true).
- defaultValue("Message from NiFi").
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- build();
- public static final PropertyDescriptor MESSAGE = new PropertyDescriptor.Builder().
- name("Message").
- description("The body of the email message").
- required(true).
- expressionLanguageSupported(true).
- defaultValue("").
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- build();
- public static final PropertyDescriptor ATTACH_FILE = new PropertyDescriptor.Builder().
- name("Attach File").
- description("Specifies whether or not the FlowFile content should be attached to the email").
- required(true).
- allowableValues("true", "false").
- defaultValue("false").
- build();
- public static final PropertyDescriptor INCLUDE_ALL_ATTRIBUTES = new PropertyDescriptor.Builder().
- name("Include All Attributes In Message").
- description("Specifies whether or not all FlowFile attributes should be recorded in the body of the email message").
- required(true).
- allowableValues("true", "false").
- defaultValue("false").
- build();
-
- public static final Relationship REL_SUCCESS = new Relationship.Builder().
- name("success").
- description("FlowFiles that are successfully sent will be routed to this relationship").
- build();
- public static final Relationship REL_FAILURE = new Relationship.Builder().
- name("failure").
- description("FlowFiles that fail to send will be routed to this relationship").
- build();
+ .name("CC")
+ .description("The recipients to include in the CC-Line of the email")
+ .required(false)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor BCC = new PropertyDescriptor.Builder()
+ .name("BCC")
+ .description("The recipients to include in the BCC-Line of the email")
+ .required(false)
+ .expressionLanguageSupported(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor SUBJECT = new PropertyDescriptor.Builder()
+ .name("Subject")
+ .description("The email subject")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .defaultValue("Message from NiFi")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor MESSAGE = new PropertyDescriptor.Builder()
+ .name("Message")
+ .description("The body of the email message")
+ .required(true)
+ .expressionLanguageSupported(true)
+ .defaultValue("")
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor ATTACH_FILE = new PropertyDescriptor.Builder()
+ .name("Attach File")
+ .description("Specifies whether or not the FlowFile content should be attached to the email")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+ public static final PropertyDescriptor INCLUDE_ALL_ATTRIBUTES = new PropertyDescriptor.Builder()
+ .name("Include All Attributes In Message")
+ .description("Specifies whether or not all FlowFile attributes should be recorded in the body of the email message")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("false")
+ .build();
+
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("FlowFiles that are successfully sent will be routed to this relationship")
+ .build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("FlowFiles that fail to send will be routed to this relationship")
+ .build();
private List<PropertyDescriptor> properties;
private Set<Relationship> relationships;
/**
- * Mapping of the mail properties to the NiFi PropertyDescriptors that will
- * be evaluated at runtime
+ * Mapping of the mail properties to the NiFi PropertyDescriptors that will be evaluated at runtime
*/
private static final Map<String, PropertyDescriptor> propertyToContext = new HashMap<>();
@@ -221,8 +220,7 @@ public class PutEmail extends AbstractProcessor {
propertyToContext.put("mail.smtp.host", SMTP_HOSTNAME);
propertyToContext.put("mail.smtp.port", SMTP_PORT);
propertyToContext.put("mail.smtp.socketFactory.port", SMTP_PORT);
- propertyToContext.
- put("mail.smtp.socketFactory.class", SMTP_SOCKET_FACTORY);
+ propertyToContext.put("mail.smtp.socketFactory.class", SMTP_SOCKET_FACTORY);
propertyToContext.put("mail.smtp.auth", SMTP_AUTH);
propertyToContext.put("mail.smtp.starttls.enable", SMTP_TLS);
propertyToContext.put("mail.smtp.user", SMTP_USERNAME);
@@ -269,21 +267,15 @@ public class PutEmail extends AbstractProcessor {
@Override
protected Collection<ValidationResult> customValidate(final ValidationContext context) {
- final List<ValidationResult> errors = new ArrayList<>(super.
- customValidate(context));
+ final List<ValidationResult> errors = new ArrayList<>(super.customValidate(context));
- final String to = context.getProperty(TO).
- getValue();
- final String cc = context.getProperty(CC).
- getValue();
- final String bcc = context.getProperty(BCC).
- getValue();
+ final String to = context.getProperty(TO).getValue();
+ final String cc = context.getProperty(CC).getValue();
+ final String bcc = context.getProperty(BCC).getValue();
if (to == null && cc == null && bcc == null) {
errors.add(new ValidationResult.Builder().subject("To, CC, BCC").
- valid(false).
- explanation("Must specify at least one To/CC/BCC address").
- build());
+ valid(false).explanation("Must specify at least one To/CC/BCC address").build());
}
return errors;
@@ -296,8 +288,7 @@ public class PutEmail extends AbstractProcessor {
return;
}
- final Properties properties = this.
- getMailPropertiesFromFlowFile(context, flowFile);
+ final Properties properties = this.getMailPropertiesFromFlowFile(context, flowFile);
final Session mailSession = this.createMailSession(properties);
@@ -305,71 +296,46 @@ public class PutEmail extends AbstractProcessor {
final ProcessorLog logger = getLogger();
try {
- message.setFrom(InternetAddress.parse(context.getProperty(FROM).
- evaluateAttributeExpressions(flowFile).
- getValue())[0]);
-
- final InternetAddress[] toAddresses = toInetAddresses(context.
- getProperty(TO).
- evaluateAttributeExpressions(flowFile).
- getValue());
+ message.setFrom(InternetAddress.parse(context.getProperty(FROM).evaluateAttributeExpressions(flowFile).getValue())[0]);
+
+ final InternetAddress[] toAddresses = toInetAddresses(context.getProperty(TO).evaluateAttributeExpressions(flowFile).getValue());
message.setRecipients(RecipientType.TO, toAddresses);
- final InternetAddress[] ccAddresses = toInetAddresses(context.
- getProperty(CC).
- evaluateAttributeExpressions(flowFile).
- getValue());
+ final InternetAddress[] ccAddresses = toInetAddresses(context.getProperty(CC).evaluateAttributeExpressions(flowFile).getValue());
message.setRecipients(RecipientType.CC, ccAddresses);
- final InternetAddress[] bccAddresses = toInetAddresses(context.
- getProperty(BCC).
- evaluateAttributeExpressions(flowFile).
- getValue());
+ final InternetAddress[] bccAddresses = toInetAddresses(context.getProperty(BCC).evaluateAttributeExpressions(flowFile).getValue());
message.setRecipients(RecipientType.BCC, bccAddresses);
- message.setHeader("X-Mailer", context.getProperty(HEADER_XMAILER).
- evaluateAttributeExpressions(flowFile).
- getValue());
- message.setSubject(context.getProperty(SUBJECT).
- evaluateAttributeExpressions(flowFile).
- getValue());
- String messageText = context.getProperty(MESSAGE).
- evaluateAttributeExpressions(flowFile).
- getValue();
-
- if (context.getProperty(INCLUDE_ALL_ATTRIBUTES).
- asBoolean()) {
+ message.setHeader("X-Mailer", context.getProperty(HEADER_XMAILER).evaluateAttributeExpressions(flowFile).getValue());
+ message.setSubject(context.getProperty(SUBJECT).evaluateAttributeExpressions(flowFile).getValue());
+ String messageText = context.getProperty(MESSAGE).evaluateAttributeExpressions(flowFile).getValue();
+
+ if (context.getProperty(INCLUDE_ALL_ATTRIBUTES).asBoolean()) {
messageText = formatAttributes(flowFile, messageText);
}
- String contentType = context.getProperty(CONTENT_TYPE).
- evaluateAttributeExpressions(flowFile).
- getValue();
+ String contentType = context.getProperty(CONTENT_TYPE).evaluateAttributeExpressions(flowFile).getValue();
message.setContent(messageText, contentType);
message.setSentDate(new Date());
- if (context.getProperty(ATTACH_FILE).
- asBoolean()) {
+ if (context.getProperty(ATTACH_FILE).asBoolean()) {
final MimeBodyPart mimeText = new PreencodedMimeBodyPart("base64");
- mimeText.
- setDataHandler(new DataHandler(new ByteArrayDataSource(Base64.
- encodeBase64(messageText.
- getBytes("UTF-8")), "text/plain; charset=\"utf-8\"")));
+ mimeText.setDataHandler(new DataHandler(new ByteArrayDataSource(
+ Base64.encodeBase64(messageText.getBytes("UTF-8")), "text/plain; charset=\"utf-8\"")));
final MimeBodyPart mimeFile = new MimeBodyPart();
session.read(flowFile, new InputStreamCallback() {
@Override
public void process(final InputStream stream) throws IOException {
try {
- mimeFile.
- setDataHandler(new DataHandler(new ByteArrayDataSource(stream, "application/octet-stream")));
+ mimeFile.setDataHandler(new DataHandler(new ByteArrayDataSource(stream, "application/octet-stream")));
} catch (final Exception e) {
throw new IOException(e);
}
}
});
- mimeFile.setFileName(flowFile.
- getAttribute(CoreAttributes.FILENAME.key()));
+ mimeFile.setFileName(flowFile.getAttribute(CoreAttributes.FILENAME.key()));
MimeMultipart multipart = new MimeMultipart();
multipart.addBodyPart(mimeText);
multipart.addBodyPart(mimeFile);
@@ -378,24 +344,18 @@ public class PutEmail extends AbstractProcessor {
Transport.send(message);
- session.getProvenanceReporter().
- send(flowFile, "mailto:" + message.getAllRecipients()[0].
- toString());
+ session.getProvenanceReporter().send(flowFile, "mailto:" + message.getAllRecipients()[0].toString());
session.transfer(flowFile, REL_SUCCESS);
- logger.
- info("Sent email as a result of receiving {}", new Object[]{flowFile});
+ logger.info("Sent email as a result of receiving {}", new Object[]{flowFile});
} catch (final ProcessException | MessagingException | IOException e) {
context.yield();
- logger.
- error("Failed to send email for {}: {}; routing to failure", new Object[]{flowFile, e});
+ logger.error("Failed to send email for {}: {}; routing to failure", new Object[]{flowFile, e});
session.transfer(flowFile, REL_FAILURE);
}
}
/**
- * Based on the input properties, determine whether an authenticate or
- * unauthenticated session should be used. If authenticated, creates a
- * Password Authenticator for use in sending the email.
+ * Based on the input properties, determine whether an authenticate or unauthenticated session should be used. If authenticated, creates a Password Authenticator for use in sending the email.
*
* @param properties mail properties
* @return session
@@ -407,22 +367,18 @@ public class PutEmail extends AbstractProcessor {
/*
* Conditionally create a password authenticator if the 'auth' parameter is set.
*/
- final Session mailSession = auth ? Session.
- getInstance(properties, new Authenticator() {
- @Override
- public PasswordAuthentication getPasswordAuthentication() {
- String username = properties.
- getProperty("mail.smtp.user"),
- password = properties.getProperty("mail.smtp.password");
- return new PasswordAuthentication(username, password);
- }
- }) : Session.getInstance(properties); // without auth
+ final Session mailSession = auth ? Session.getInstance(properties, new Authenticator() {
+ @Override
+ public PasswordAuthentication getPasswordAuthentication() {
+ String username = properties.getProperty("mail.smtp.user"), password = properties.getProperty("mail.smtp.password");
+ return new PasswordAuthentication(username, password);
+ }
+ }) : Session.getInstance(properties); // without auth
return mailSession;
}
/**
- * Uses the mapping of javax.mail properties to NiFi PropertyDescriptors to
- * build the required Properties object to be used for sending this email
+ * Uses the mapping of javax.mail properties to NiFi PropertyDescriptors to build the required Properties object to be used for sending this email
*
* @param context context
* @param flowFile flowFile
@@ -438,14 +394,11 @@ public class PutEmail extends AbstractProcessor {
entrySet()) {
// Evaluate the property descriptor against the flow file
- String flowFileValue = context.getProperty(entry.getValue()).
- evaluateAttributeExpressions(flowFile).
- getValue();
+ String flowFileValue = context.getProperty(entry.getValue()).evaluateAttributeExpressions(flowFile).getValue();
String property = entry.getKey();
- logger.
- debug("Evaluated Mail Property: {} with Value: {}", new Object[]{property, flowFileValue});
+ logger.debug("Evaluated Mail Property: {} with Value: {}", new Object[]{property, flowFileValue});
// Nullable values are not allowed, so filter out
if (null != flowFileValue) {
@@ -464,19 +417,12 @@ public class PutEmail extends AbstractProcessor {
StringBuilder message = new StringBuilder(messagePrepend);
message.append(BODY_SEPARATOR);
message.append("\nStandard FlowFile Metadata:");
- message.append(String.
- format("\n\t%1$s = '%2$s'", "id", flowFile.getId()));
- message.append(String.
- format("\n\t%1$s = '%2$s'", "entryDate", new Date(flowFile.
- getEntryDate())));
- message.append(String.format("\n\t%1$s = '%2$s'", "fileSize", flowFile.
- getSize()));
+ message.append(String.format("\n\t%1$s = '%2$s'", "id", flowFile.getId()));
+ message.append(String.format("\n\t%1$s = '%2$s'", "entryDate", new Date(flowFile.getEntryDate())));
+ message.append(String.format("\n\t%1$s = '%2$s'", "fileSize", flowFile.getSize()));
message.append("\nFlowFile Attributes:");
- for (Entry<String, String> attribute : flowFile.getAttributes().
- entrySet()) {
- message.append(String.
- format("\n\t%1$s = '%2$s'", attribute.getKey(), attribute.
- getValue()));
+ for (Entry<String, String> attribute : flowFile.getAttributes().entrySet()) {
+ message.append(String.format("\n\t%1$s = '%2$s'", attribute.getKey(), attribute.getValue()));
}
message.append("\n");
return message.toString();
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/PutFTP.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFTP.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFTP.java
index 6786bf0..051cb07 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFTP.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFTP.java
@@ -104,8 +104,7 @@ public class PutFTP extends PutFileTransfer<FTPTransfer> {
@Override
protected void beforePut(final FlowFile flowFile, final ProcessContext context, final FTPTransfer transfer) throws IOException {
- transfer.
- sendCommands(getCommands(preSendDescriptorRef.get(), context, flowFile), flowFile);
+ transfer.sendCommands(getCommands(preSendDescriptorRef.get(), context, flowFile), flowFile);
}
@Override
@@ -122,10 +121,10 @@ public class PutFTP extends PutFileTransfer<FTPTransfer> {
@Override
protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
return new PropertyDescriptor.Builder()
- .name(propertyDescriptorName).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- dynamic(true).
- build();
+ .name(propertyDescriptorName)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .dynamic(true)
+ .build();
}
@OnScheduled
@@ -133,8 +132,7 @@ public class PutFTP extends PutFileTransfer<FTPTransfer> {
final Map<Integer, PropertyDescriptor> preDescriptors = new TreeMap<>();
final Map<Integer, PropertyDescriptor> postDescriptors = new TreeMap<>();
- for (final PropertyDescriptor descriptor : context.getProperties().
- keySet()) {
+ for (final PropertyDescriptor descriptor : context.getProperties().keySet()) {
final String name = descriptor.getName();
final Matcher preMatcher = PRE_SEND_CMD_PATTERN.matcher(name);
if (preMatcher.matches()) {
@@ -149,10 +147,8 @@ public class PutFTP extends PutFileTransfer<FTPTransfer> {
}
}
- final List<PropertyDescriptor> preDescriptorList = new ArrayList<>(preDescriptors.
- values());
- final List<PropertyDescriptor> postDescriptorList = new ArrayList<>(postDescriptors.
- values());
+ final List<PropertyDescriptor> preDescriptorList = new ArrayList<>(preDescriptors.values());
+ final List<PropertyDescriptor> postDescriptorList = new ArrayList<>(postDescriptors.values());
this.preSendDescriptorRef.set(preDescriptorList);
this.postSendDescriptorRef.set(postDescriptorList);
}
@@ -160,9 +156,7 @@ public class PutFTP extends PutFileTransfer<FTPTransfer> {
private List<String> getCommands(final List<PropertyDescriptor> descriptors, final ProcessContext context, final FlowFile flowFile) {
final List<String> cmds = new ArrayList<>();
for (final PropertyDescriptor descriptor : descriptors) {
- cmds.add(context.getProperty(descriptor).
- evaluateAttributeExpressions(flowFile).
- getValue());
+ cmds.add(context.getProperty(descriptor).evaluateAttributeExpressions(flowFile).getValue());
}
return cmds;
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/PutFile.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFile.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFile.java
index ce03491..3bbe093 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFile.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFile.java
@@ -64,76 +64,76 @@ public class PutFile extends AbstractProcessor {
public static final String FILE_MODIFY_DATE_ATTRIBUTE = "file.lastModifiedTime";
public static final String FILE_MODIFY_DATE_ATTR_FORMAT = "yyyy-MM-dd'T'HH:mm:ssZ";
- public static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder().
- name("Directory").
- description("The directory to which files should be written. You may use expression language such as /aa/bb/${path}").
- required(true).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- expressionLanguageSupported(true).
- build();
- public static final PropertyDescriptor MAX_DESTINATION_FILES = new PropertyDescriptor.Builder().
- name("Maximum File Count").
- description("Specifies the maximum number of files that can exist in the output directory").
- required(false).
- addValidator(StandardValidators.INTEGER_VALIDATOR).
- build();
- public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder().
- name("Conflict Resolution Strategy").
- description("Indicates what should happen when a file with the same name already exists in the output directory").
- required(true).
- defaultValue(FAIL_RESOLUTION).
- allowableValues(REPLACE_RESOLUTION, IGNORE_RESOLUTION, FAIL_RESOLUTION).
- build();
- public static final PropertyDescriptor CHANGE_LAST_MODIFIED_TIME = new PropertyDescriptor.Builder().
- name("Last Modified Time").
- description("Sets the lastModifiedTime on the output file to the value of this attribute. Format must be yyyy-MM-dd'T'HH:mm:ssZ. "
- + "You may also use expression language such as ${file.lastModifiedTime}.").
- required(false).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- expressionLanguageSupported(true).
- build();
- public static final PropertyDescriptor CHANGE_PERMISSIONS = new PropertyDescriptor.Builder().
- name("Permissions").
- description("Sets the permissions on the output file to the value of this attribute. Format must be either UNIX rwxrwxrwx with a - in "
+ public static final PropertyDescriptor DIRECTORY = new PropertyDescriptor.Builder()
+ .name("Directory")
+ .description("The directory to which files should be written. You may use expression language such as /aa/bb/${path}")
+ .required(true)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ public static final PropertyDescriptor MAX_DESTINATION_FILES = new PropertyDescriptor.Builder()
+ .name("Maximum File Count")
+ .description("Specifies the maximum number of files that can exist in the output directory")
+ .required(false)
+ .addValidator(StandardValidators.INTEGER_VALIDATOR)
+ .build();
+ public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
+ .name("Conflict Resolution Strategy")
+ .description("Indicates what should happen when a file with the same name already exists in the output directory")
+ .required(true)
+ .defaultValue(FAIL_RESOLUTION)
+ .allowableValues(REPLACE_RESOLUTION, IGNORE_RESOLUTION, FAIL_RESOLUTION)
+ .build();
+ public static final PropertyDescriptor CHANGE_LAST_MODIFIED_TIME = new PropertyDescriptor.Builder()
+ .name("Last Modified Time")
+ .description("Sets the lastModifiedTime on the output file to the value of this attribute. Format must be yyyy-MM-dd'T'HH:mm:ssZ. "
+ + "You may also use expression language such as ${file.lastModifiedTime}.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ public static final PropertyDescriptor CHANGE_PERMISSIONS = new PropertyDescriptor.Builder()
+ .name("Permissions")
+ .description("Sets the permissions on the output file to the value of this attribute. Format must be either UNIX rwxrwxrwx with a - in "
+ "place of denied permissions (e.g. rw-r--r--) or an octal number (e.g. 644). You may also use expression language such as "
- + "${file.permissions}.").
- required(false).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- expressionLanguageSupported(true).
- build();
- public static final PropertyDescriptor CHANGE_OWNER = new PropertyDescriptor.Builder().
- name("Owner").
- description("Sets the owner on the output file to the value of this attribute. You may also use expression language such as "
- + "${file.owner}.").
- required(false).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- expressionLanguageSupported(true).
- build();
- public static final PropertyDescriptor CHANGE_GROUP = new PropertyDescriptor.Builder().
- name("Group").
- description("Sets the group on the output file to the value of this attribute. You may also use expression language such "
- + "as ${file.group}.").
- required(false).
- addValidator(StandardValidators.NON_EMPTY_VALIDATOR).
- expressionLanguageSupported(true).
- build();
- public static final PropertyDescriptor CREATE_DIRS = new PropertyDescriptor.Builder().
- name("Create Missing Directories").
- description("If true, then missing destination directories will be created. If false, flowfiles are penalized and sent to failure.").
- required(true).
- allowableValues("true", "false").
- defaultValue("true").
- build();
+ + "${file.permissions}.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ public static final PropertyDescriptor CHANGE_OWNER = new PropertyDescriptor.Builder()
+ .name("Owner")
+ .description("Sets the owner on the output file to the value of this attribute. You may also use expression language such as "
+ + "${file.owner}.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ public static final PropertyDescriptor CHANGE_GROUP = new PropertyDescriptor.Builder()
+ .name("Group")
+ .description("Sets the group on the output file to the value of this attribute. You may also use expression language such "
+ + "as ${file.group}.")
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .build();
+ public static final PropertyDescriptor CREATE_DIRS = new PropertyDescriptor.Builder()
+ .name("Create Missing Directories")
+ .description("If true, then missing destination directories will be created. If false, flowfiles are penalized and sent to failure.")
+ .required(true)
+ .allowableValues("true", "false")
+ .defaultValue("true")
+ .build();
public static final int MAX_FILE_LOCK_ATTEMPTS = 10;
- public static final Relationship REL_SUCCESS = new Relationship.Builder().
- name("success").
- description("Files that have been successfully written to the output directory are transferred to this relationship").
- build();
- public static final Relationship REL_FAILURE = new Relationship.Builder().
- name("failure").
- description("Files that could not be written to the output directory for some reason are transferred to this relationship").
- build();
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("Files that have been successfully written to the output directory are transferred to this relationship")
+ .build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("Files that could not be written to the output directory for some reason are transferred to this relationship")
+ .build();
private List<PropertyDescriptor> properties;
private Set<Relationship> relationships;
@@ -177,35 +177,25 @@ public class PutFile extends AbstractProcessor {
}
final StopWatch stopWatch = new StopWatch(true);
- final Path configuredRootDirPath = Paths.get(context.
- getProperty(DIRECTORY).
- evaluateAttributeExpressions(flowFile).
- getValue());
- final String conflictResponse = context.getProperty(CONFLICT_RESOLUTION).
- getValue();
- final Integer maxDestinationFiles = context.
- getProperty(MAX_DESTINATION_FILES).
- asInteger();
+ final Path configuredRootDirPath = Paths.get(context.getProperty(DIRECTORY).evaluateAttributeExpressions(flowFile).getValue());
+ final String conflictResponse = context.getProperty(CONFLICT_RESOLUTION).getValue();
+ final Integer maxDestinationFiles = context.getProperty(MAX_DESTINATION_FILES).asInteger();
final ProcessorLog logger = getLogger();
Path tempDotCopyFile = null;
try {
final Path rootDirPath = configuredRootDirPath;
- final Path tempCopyFile = rootDirPath.resolve("." + flowFile.
- getAttribute(CoreAttributes.FILENAME.key()));
- final Path copyFile = rootDirPath.resolve(flowFile.
- getAttribute(CoreAttributes.FILENAME.key()));
+ final Path tempCopyFile = rootDirPath.resolve("." + flowFile.getAttribute(CoreAttributes.FILENAME.key()));
+ final Path copyFile = rootDirPath.resolve(flowFile.getAttribute(CoreAttributes.FILENAME.key()));
if (!Files.exists(rootDirPath)) {
- if (context.getProperty(CREATE_DIRS).
- asBoolean()) {
+ if (context.getProperty(CREATE_DIRS).asBoolean()) {
Files.createDirectories(rootDirPath);
} else {
flowFile = session.penalize(flowFile);
session.transfer(flowFile, REL_FAILURE);
- logger.
- error("Penalizing {} and routing to 'failure' because the output directory {} does not exist and Processor is "
- + "configured not to create missing directories", new Object[]{flowFile, rootDirPath});
+ logger.error("Penalizing {} and routing to 'failure' because the output directory {} does not exist and Processor is "
+ + "configured not to create missing directories", new Object[]{flowFile, rootDirPath});
return;
}
}
@@ -216,14 +206,12 @@ public class PutFile extends AbstractProcessor {
final Path finalCopyFileDir = finalCopyFile.getParent();
if (Files.exists(finalCopyFileDir) && maxDestinationFiles != null) { // check if too many files already
- final int numFiles = finalCopyFileDir.toFile().
- list().length;
+ final int numFiles = finalCopyFileDir.toFile().list().length;
if (numFiles >= maxDestinationFiles) {
flowFile = session.penalize(flowFile);
- logger.
- info("Penalizing {} and routing to 'failure' because the output directory {} has {} files, which exceeds the "
- + "configured maximum number of files", new Object[]{flowFile, finalCopyFileDir, numFiles});
+ logger.info("Penalizing {} and routing to 'failure' because the output directory {} has {} files, which exceeds the "
+ + "configured maximum number of files", new Object[]{flowFile, finalCopyFileDir, numFiles});
session.transfer(flowFile, REL_FAILURE);
return;
}
@@ -233,18 +221,15 @@ public class PutFile extends AbstractProcessor {
switch (conflictResponse) {
case REPLACE_RESOLUTION:
Files.delete(finalCopyFile);
- logger.
- info("Deleted {} as configured in order to replace with the contents of {}", new Object[]{finalCopyFile, flowFile});
+ logger.info("Deleted {} as configured in order to replace with the contents of {}", new Object[]{finalCopyFile, flowFile});
break;
case IGNORE_RESOLUTION:
session.transfer(flowFile, REL_SUCCESS);
- logger.
- info("Transferring {} to success because file with same name already exists", new Object[]{flowFile});
+ logger.info("Transferring {} to success because file with same name already exists", new Object[]{flowFile});
return;
case FAIL_RESOLUTION:
flowFile = session.penalize(flowFile);
- logger.
- info("Penalizing {} and routing to failure as configured because file with the same name already exists", new Object[]{flowFile});
+ logger.info("Penalizing {} and routing to failure as configured because file with the same name already exists", new Object[]{flowFile});
session.transfer(flowFile, REL_FAILURE);
return;
default:
@@ -254,82 +239,53 @@ public class PutFile extends AbstractProcessor {
session.exportTo(flowFile, dotCopyFile, false);
- final String lastModifiedTime = context.
- getProperty(CHANGE_LAST_MODIFIED_TIME).
- evaluateAttributeExpressions(flowFile).
- getValue();
- if (lastModifiedTime != null && !lastModifiedTime.trim().
- isEmpty()) {
+ final String lastModifiedTime = context.getProperty(CHANGE_LAST_MODIFIED_TIME).evaluateAttributeExpressions(flowFile).getValue();
+ if (lastModifiedTime != null && !lastModifiedTime.trim().isEmpty()) {
try {
final DateFormat formatter = new SimpleDateFormat(FILE_MODIFY_DATE_ATTR_FORMAT, Locale.US);
- final Date fileModifyTime = formatter.
- parse(lastModifiedTime);
- dotCopyFile.toFile().
- setLastModified(fileModifyTime.getTime());
+ final Date fileModifyTime = formatter.parse(lastModifiedTime);
+ dotCopyFile.toFile().setLastModified(fileModifyTime.getTime());
} catch (Exception e) {
- logger.
- warn("Could not set file lastModifiedTime to {} because {}", new Object[]{lastModifiedTime, e});
+ logger.warn("Could not set file lastModifiedTime to {} because {}", new Object[]{lastModifiedTime, e});
}
}
- final String permissions = context.getProperty(CHANGE_PERMISSIONS).
- evaluateAttributeExpressions(flowFile).
- getValue();
- if (permissions != null && !permissions.trim().
- isEmpty()) {
+ final String permissions = context.getProperty(CHANGE_PERMISSIONS).evaluateAttributeExpressions(flowFile).getValue();
+ if (permissions != null && !permissions.trim().isEmpty()) {
try {
String perms = stringPermissions(permissions);
if (!perms.isEmpty()) {
- Files.
- setPosixFilePermissions(dotCopyFile, PosixFilePermissions.
- fromString(perms));
+ Files.setPosixFilePermissions(dotCopyFile, PosixFilePermissions.fromString(perms));
}
} catch (Exception e) {
- logger.
- warn("Could not set file permissions to {} because {}", new Object[]{permissions, e});
+ logger.warn("Could not set file permissions to {} because {}", new Object[]{permissions, e});
}
}
- final String owner = context.getProperty(CHANGE_OWNER).
- evaluateAttributeExpressions(flowFile).
- getValue();
- if (owner != null && !owner.trim().
- isEmpty()) {
+ final String owner = context.getProperty(CHANGE_OWNER).evaluateAttributeExpressions(flowFile).getValue();
+ if (owner != null && !owner.trim().isEmpty()) {
try {
- UserPrincipalLookupService lookupService = dotCopyFile.
- getFileSystem().
- getUserPrincipalLookupService();
- Files.setOwner(dotCopyFile, lookupService.
- lookupPrincipalByName(owner));
+ UserPrincipalLookupService lookupService = dotCopyFile.getFileSystem().getUserPrincipalLookupService();
+ Files.setOwner(dotCopyFile, lookupService.lookupPrincipalByName(owner));
} catch (Exception e) {
- logger.
- warn("Could not set file owner to {} because {}", new Object[]{owner, e});
+ logger.warn("Could not set file owner to {} because {}", new Object[]{owner, e});
}
}
- final String group = context.getProperty(CHANGE_GROUP).
- evaluateAttributeExpressions(flowFile).
- getValue();
- if (group != null && !group.trim().
- isEmpty()) {
+ final String group = context.getProperty(CHANGE_GROUP).evaluateAttributeExpressions(flowFile).getValue();
+ if (group != null && !group.trim().isEmpty()) {
try {
- UserPrincipalLookupService lookupService = dotCopyFile.
- getFileSystem().
- getUserPrincipalLookupService();
- PosixFileAttributeView view = Files.
- getFileAttributeView(dotCopyFile, PosixFileAttributeView.class);
- view.setGroup(lookupService.
- lookupPrincipalByGroupName(group));
+ UserPrincipalLookupService lookupService = dotCopyFile.getFileSystem().getUserPrincipalLookupService();
+ PosixFileAttributeView view = Files.getFileAttributeView(dotCopyFile, PosixFileAttributeView.class);
+ view.setGroup(lookupService.lookupPrincipalByGroupName(group));
} catch (Exception e) {
- logger.
- warn("Could not set file group to {} because {}", new Object[]{group, e});
+ logger.warn("Could not set file group to {} because {}", new Object[]{group, e});
}
}
boolean renamed = false;
for (int i = 0; i < 10; i++) { // try rename up to 10 times.
- if (dotCopyFile.toFile().
- renameTo(finalCopyFile.toFile())) {
+ if (dotCopyFile.toFile().renameTo(finalCopyFile.toFile())) {
renamed = true;
break;// rename was successful
}
@@ -337,36 +293,27 @@ public class PutFile extends AbstractProcessor {
}
if (!renamed) {
- if (Files.exists(dotCopyFile) && dotCopyFile.toFile().
- delete()) {
- logger.
- debug("Deleted dot copy file {}", new Object[]{dotCopyFile});
+ if (Files.exists(dotCopyFile) && dotCopyFile.toFile().delete()) {
+ logger.debug("Deleted dot copy file {}", new Object[]{dotCopyFile});
}
throw new ProcessException("Could not rename: " + dotCopyFile);
} else {
- logger.
- info("Produced copy of {} at location {}", new Object[]{flowFile, finalCopyFile});
+ logger.info("Produced copy of {} at location {}", new Object[]{flowFile, finalCopyFile});
}
- session.getProvenanceReporter().
- send(flowFile, finalCopyFile.toFile().
- toURI().
- toString(), stopWatch.
- getElapsed(TimeUnit.MILLISECONDS));
+ session.getProvenanceReporter().send(flowFile, finalCopyFile.toFile().toURI().toString(), stopWatch.getElapsed(TimeUnit.MILLISECONDS));
session.transfer(flowFile, REL_SUCCESS);
} catch (final Throwable t) {
if (tempDotCopyFile != null) {
try {
Files.deleteIfExists(tempDotCopyFile);
} catch (final Exception e) {
- logger.
- error("Unable to remove temporary file {} due to {}", new Object[]{tempDotCopyFile, e});
+ logger.error("Unable to remove temporary file {} due to {}", new Object[]{tempDotCopyFile, e});
}
}
flowFile = session.penalize(flowFile);
- logger.
- error("Penalizing {} and transferring to failure due to {}", new Object[]{flowFile, t});
+ logger.error("Penalizing {} and transferring to failure due to {}", new Object[]{flowFile, t});
session.transfer(flowFile, REL_FAILURE);
}
}
@@ -375,11 +322,9 @@ public class PutFile extends AbstractProcessor {
String permissions = "";
final Pattern rwxPattern = Pattern.compile("^[rwx-]{9}$");
final Pattern numPattern = Pattern.compile("\\d+");
- if (rwxPattern.matcher(perms).
- matches()) {
+ if (rwxPattern.matcher(perms).matches()) {
permissions = perms;
- } else if (numPattern.matcher(perms).
- matches()) {
+ } else if (numPattern.matcher(perms).matches()) {
try {
int number = Integer.parseInt(perms, 8);
StringBuilder permBuilder = new StringBuilder();
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/PutFileTransfer.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFileTransfer.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFileTransfer.java
index 893aee9..b60d07f 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFileTransfer.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/PutFileTransfer.java
@@ -48,18 +48,18 @@ import java.util.concurrent.TimeUnit;
*/
public abstract class PutFileTransfer<T extends FileTransfer> extends AbstractProcessor {
- public static final Relationship REL_SUCCESS = new Relationship.Builder().
- name("success").
- description("FlowFiles that are successfully sent will be routed to success").
- build();
- public static final Relationship REL_FAILURE = new Relationship.Builder().
- name("failure").
- description("FlowFiles that failed to send to the remote system; failure is usually looped back to this processor").
- build();
- public static final Relationship REL_REJECT = new Relationship.Builder().
- name("reject").
- description("FlowFiles that were rejected by the destination system").
- build();
+ public static final Relationship REL_SUCCESS = new Relationship.Builder()
+ .name("success")
+ .description("FlowFiles that are successfully sent will be routed to success")
+ .build();
+ public static final Relationship REL_FAILURE = new Relationship.Builder()
+ .name("failure")
+ .description("FlowFiles that failed to send to the remote system; failure is usually looped back to this processor")
+ .build();
+ public static final Relationship REL_REJECT = new Relationship.Builder()
+ .name("reject")
+ .description("FlowFiles that were rejected by the destination system")
+ .build();
private final Set<Relationship> relationships;
@@ -95,43 +95,27 @@ public abstract class PutFileTransfer<T extends FileTransfer> extends AbstractPr
}
final ProcessorLog logger = getLogger();
- final String hostname = context.getProperty(FileTransfer.HOSTNAME).
- evaluateAttributeExpressions(flowFile).
- getValue();
+ final String hostname = context.getProperty(FileTransfer.HOSTNAME).evaluateAttributeExpressions(flowFile).getValue();
- final int maxNumberOfFiles = context.
- getProperty(FileTransfer.BATCH_SIZE).
- asInteger();
+ final int maxNumberOfFiles = context.getProperty(FileTransfer.BATCH_SIZE).asInteger();
int fileCount = 0;
try (final T transfer = getFileTransfer(context)) {
do {
- final String rootPath = context.
- getProperty(FileTransfer.REMOTE_PATH).
- evaluateAttributeExpressions(flowFile).
- getValue();
+ final String rootPath = context.getProperty(FileTransfer.REMOTE_PATH).evaluateAttributeExpressions(flowFile).getValue();
final String workingDirPath;
if (rootPath == null) {
workingDirPath = null;
} else {
File workingDirectory = new File(rootPath);
- if (!workingDirectory.getPath().
- startsWith("/") && !workingDirectory.getPath().
- startsWith("\\")) {
- workingDirectory = new File(transfer.
- getHomeDirectory(flowFile), workingDirectory.
- getPath());
+ if (!workingDirectory.getPath().startsWith("/") && !workingDirectory.getPath().startsWith("\\")) {
+ workingDirectory = new File(transfer.getHomeDirectory(flowFile), workingDirectory.getPath());
}
- workingDirPath = workingDirectory.getPath().
- replace("\\", "/");
+ workingDirPath = workingDirectory.getPath().replace("\\", "/");
}
- final boolean rejectZeroByteFiles = context.
- getProperty(FileTransfer.REJECT_ZERO_BYTE).
- asBoolean();
- final ConflictResult conflictResult = identifyAndResolveConflictFile(context.
- getProperty(FileTransfer.CONFLICT_RESOLUTION).
- getValue(),
- transfer, workingDirPath, flowFile, rejectZeroByteFiles, logger);
+ final boolean rejectZeroByteFiles = context.getProperty(FileTransfer.REJECT_ZERO_BYTE).asBoolean();
+ final ConflictResult conflictResult
+ = identifyAndResolveConflictFile(context.getProperty(FileTransfer.CONFLICT_RESOLUTION).getValue(), transfer, workingDirPath, flowFile, rejectZeroByteFiles, logger);
if (conflictResult.isTransfer()) {
final StopWatch stopWatch = new StopWatch();
@@ -144,37 +128,28 @@ public abstract class PutFileTransfer<T extends FileTransfer> extends AbstractPr
@Override
public void process(final InputStream in) throws IOException {
try (final InputStream bufferedIn = new BufferedInputStream(in)) {
- if (workingDirPath != null && context.
- getProperty(SFTPTransfer.CREATE_DIRECTORY).
- asBoolean()) {
- transfer.
- ensureDirectoryExists(flowFileToTransfer, new File(workingDirPath));
+ if (workingDirPath != null && context.getProperty(SFTPTransfer.CREATE_DIRECTORY).asBoolean()) {
+ transfer.ensureDirectoryExists(flowFileToTransfer, new File(workingDirPath));
}
- fullPathRef.set(transfer.
- put(flowFileToTransfer, workingDirPath, conflictResult.
- getFileName(), bufferedIn));
+ fullPathRef.set(transfer.put(flowFileToTransfer, workingDirPath, conflictResult.getFileName(), bufferedIn));
}
}
});
afterPut(flowFile, context, transfer);
stopWatch.stop();
- final String dataRate = stopWatch.
- calculateDataRate(flowFile.getSize());
- final long millis = stopWatch.
- getDuration(TimeUnit.MILLISECONDS);
- logger.
- info("Successfully transfered {} to {} on remote host {} in {} milliseconds at a rate of {}",
- new Object[]{flowFile, fullPathRef.get(), hostname, millis, dataRate});
+ final String dataRate = stopWatch.calculateDataRate(flowFile.getSize());
+ final long millis = stopWatch.getDuration(TimeUnit.MILLISECONDS);
+ logger.info("Successfully transfered {} to {} on remote host {} in {} milliseconds at a rate of {}",
+ new Object[]{flowFile, fullPathRef.get(), hostname, millis, dataRate});
String fullPathWithSlash = fullPathRef.get();
if (!fullPathWithSlash.startsWith("/")) {
fullPathWithSlash = "/" + fullPathWithSlash;
}
final String destinationUri = transfer.getProtocolName() + "://" + hostname + fullPathWithSlash;
- session.getProvenanceReporter().
- send(flowFile, destinationUri, millis);
+ session.getProvenanceReporter().send(flowFile, destinationUri, millis);
}
if (conflictResult.isPenalize()) {
@@ -183,28 +158,23 @@ public abstract class PutFileTransfer<T extends FileTransfer> extends AbstractPr
session.transfer(flowFile, conflictResult.getRelationship());
session.commit();
- } while (isScheduled() && (getRelationships().
- size() == context.getAvailableRelationships().
- size()) && (++fileCount < maxNumberOfFiles) && ((flowFile = session.
- get()) != null));
+ } while (isScheduled()
+ && (getRelationships().size() == context.getAvailableRelationships().size())
+ && (++fileCount < maxNumberOfFiles)
+ && ((flowFile = session.get()) != null));
} catch (final IOException e) {
context.yield();
- logger.
- error("Unable to transfer {} to remote host {} due to {}", new Object[]{flowFile, hostname, e});
+ logger.error("Unable to transfer {} to remote host {} due to {}", new Object[]{flowFile, hostname, e});
flowFile = session.penalize(flowFile);
session.transfer(flowFile, REL_FAILURE);
} catch (final FlowFileAccessException e) {
context.yield();
- logger.
- error("Unable to transfer {} to remote host {} due to {}", new Object[]{flowFile, hostname, e.
- getCause()});
+ logger.error("Unable to transfer {} to remote host {} due to {}", new Object[]{flowFile, hostname, e.getCause()});
flowFile = session.penalize(flowFile);
session.transfer(flowFile, REL_FAILURE);
} catch (final ProcessException e) {
context.yield();
- logger.
- error("Unable to transfer {} to remote host {} due to {}: {}; routing to failure", new Object[]{flowFile, hostname, e, e.
- getCause()});
+ logger.error("Unable to transfer {} to remote host {} due to {}: {}; routing to failure", new Object[]{flowFile, hostname, e, e.getCause()});
flowFile = session.penalize(flowFile);
session.transfer(flowFile, REL_FAILURE);
}
@@ -222,62 +192,53 @@ public abstract class PutFileTransfer<T extends FileTransfer> extends AbstractPr
if (rejectZeroByteFiles) {
final long sizeInBytes = flowFile.getSize();
if (sizeInBytes == 0) {
- logger.
- warn("Rejecting {} because it is zero bytes", new Object[]{flowFile});
+ logger.warn("Rejecting {} because it is zero bytes", new Object[]{flowFile});
return new ConflictResult(REL_REJECT, false, fileName, true);
}
}
//Second, check if the user doesn't care about detecting naming conflicts ahead of time
- if (conflictResolutionType.
- equalsIgnoreCase(FileTransfer.CONFLICT_RESOLUTION_NONE)) {
+ if (conflictResolutionType.equalsIgnoreCase(FileTransfer.CONFLICT_RESOLUTION_NONE)) {
return new ConflictResult(destinationRelationship, transferFile, fileName, penalizeFile);
}
- final FileInfo remoteFileInfo = transfer.
- getRemoteFileInfo(flowFile, path, fileName);
+ final FileInfo remoteFileInfo = transfer.getRemoteFileInfo(flowFile, path, fileName);
if (remoteFileInfo == null) {
return new ConflictResult(destinationRelationship, transferFile, fileName, penalizeFile);
}
if (remoteFileInfo.isDirectory()) {
- logger.
- info("Resolving conflict by rejecting {} due to conflicting filename with a directory or file already on remote server", new Object[]{flowFile});
+ logger.info("Resolving conflict by rejecting {} due to conflicting filename with a directory or file already on remote server", new Object[]{flowFile});
return new ConflictResult(REL_REJECT, false, fileName, false);
}
- logger.
- info("Discovered a filename conflict on the remote server for {} so handling using configured Conflict Resolution of {}",
- new Object[]{flowFile, conflictResolutionType});
+ logger.info("Discovered a filename conflict on the remote server for {} so handling using configured Conflict Resolution of {}",
+ new Object[]{flowFile, conflictResolutionType});
switch (conflictResolutionType.toUpperCase()) {
case FileTransfer.CONFLICT_RESOLUTION_REJECT:
destinationRelationship = REL_REJECT;
transferFile = false;
penalizeFile = false;
- logger.
- info("Resolving conflict by rejecting {} due to conflicting filename with a directory or file already on remote server", new Object[]{flowFile});
+ logger.info("Resolving conflict by rejecting {} due to conflicting filename with a directory or file already on remote server", new Object[]{flowFile});
break;
case FileTransfer.CONFLICT_RESOLUTION_REPLACE:
transfer.deleteFile(path, fileName);
destinationRelationship = REL_SUCCESS;
transferFile = true;
penalizeFile = false;
- logger.
- info("Resolving filename conflict for {} with remote server by deleting remote file and replacing with flow file", new Object[]{flowFile});
+ logger.info("Resolving filename conflict for {} with remote server by deleting remote file and replacing with flow file", new Object[]{flowFile});
break;
case FileTransfer.CONFLICT_RESOLUTION_RENAME:
boolean uniqueNameGenerated = false;
for (int i = 1; i < 100 && !uniqueNameGenerated; i++) {
String possibleFileName = i + "." + fileName;
- final FileInfo renamedFileInfo = transfer.
- getRemoteFileInfo(flowFile, path, possibleFileName);
+ final FileInfo renamedFileInfo = transfer.getRemoteFileInfo(flowFile, path, possibleFileName);
uniqueNameGenerated = (renamedFileInfo == null);
if (uniqueNameGenerated) {
fileName = possibleFileName;
- logger.
- info("Attempting to resolve filename conflict for {} on the remote server by using a newly generated filename of: {}", new Object[]{flowFile, fileName});
+ logger.info("Attempting to resolve filename conflict for {} on the remote server by using a newly generated filename of: {}", new Object[]{flowFile, fileName});
destinationRelationship = REL_SUCCESS;
transferFile = true;
penalizeFile = false;
@@ -288,23 +249,20 @@ public abstract class PutFileTransfer<T extends FileTransfer> extends AbstractPr
destinationRelationship = REL_REJECT;
transferFile = false;
penalizeFile = false;
- logger.
- info("Could not determine a unique name after 99 attempts for. Switching resolution mode to REJECT for " + flowFile);
+ logger.info("Could not determine a unique name after 99 attempts for. Switching resolution mode to REJECT for " + flowFile);
}
break;
case FileTransfer.CONFLICT_RESOLUTION_IGNORE:
destinationRelationship = REL_SUCCESS;
transferFile = false;
penalizeFile = false;
- logger.
- info("Resolving conflict for {} by not transferring file and and still considering the process a success.", new Object[]{flowFile});
+ logger.info("Resolving conflict for {} by not transferring file and and still considering the process a success.", new Object[]{flowFile});
break;
case FileTransfer.CONFLICT_RESOLUTION_FAIL:
destinationRelationship = REL_FAILURE;
transferFile = false;
penalizeFile = true;
- logger.
- info("Resolved filename conflict for {} as configured by routing to FAILURE relationship.", new Object[]{flowFile});
+ logger.info("Resolved filename conflict for {} as configured by routing to FAILURE relationship.", new Object[]{flowFile});
default:
break;
}