You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by da...@apache.org on 2015/06/27 01:32:29 UTC
[16/35] incubator-nifi git commit: NIFI-378: updated documentation to
explain contract of MergeContent's Defragment strategy a bit more clearly
NIFI-378: updated documentation to explain contract of MergeContent's Defragment strategy a bit more clearly
Project: http://git-wip-us.apache.org/repos/asf/incubator-nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-nifi/commit/ddad70ba
Tree: http://git-wip-us.apache.org/repos/asf/incubator-nifi/tree/ddad70ba
Diff: http://git-wip-us.apache.org/repos/asf/incubator-nifi/diff/ddad70ba
Branch: refs/heads/NIFI-632
Commit: ddad70ba009a875e088020ae1ea8f29caeb529f9
Parents: 5d8bfa7
Author: Mark Payne <ma...@hotmail.com>
Authored: Mon Jun 22 12:42:22 2015 -0400
Committer: Mark Payne <ma...@hotmail.com>
Committed: Mon Jun 22 12:42:22 2015 -0400
----------------------------------------------------------------------
.../nifi/processors/standard/MergeContent.java | 57 +++++++++++---------
1 file changed, 31 insertions(+), 26 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-nifi/blob/ddad70ba/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
----------------------------------------------------------------------
diff --git a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
index 2883a75..65f4124 100644
--- a/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
+++ b/nifi/nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/MergeContent.java
@@ -82,31 +82,34 @@ import org.apache.nifi.util.ObjectHolder;
+ "created from FlowFiles in different connections. This processor updates the mime.type attribute as appropriate.")
@ReadsAttributes({
@ReadsAttribute(attribute = "fragment.identifier", description = "Applicable only if the <Merge Strategy> property is set to Defragment. "
- + "All FlowFiles with the same value for this attribute will be bundled together"),
- @ReadsAttribute(attribute = "fragment.index", description = "Applicable only if the <Merge Strategy> property is set to Defragment. This "
- + "attribute must be present on all FlowFiles with the same value for the fragment.identifier attribute and must be a unique integer "
- + "between 0 and the value of the fragment.count attribute. This attribute indicates the order in which the fragments should be assembled"),
+ + "All FlowFiles with the same value for this attribute will be bundled together."),
+ @ReadsAttribute(attribute = "fragment.index", description = "Applicable only if the <Merge Strategy> property is set to Defragment. "
+ + "This attribute indicates the order in which the fragments should be assembled. This "
+ + "attribute must be present on all FlowFiles when using the Defragment Merge Strategy and must be a unique (i.e., unique across all "
+ + "FlowFiles that have the same value for the \"fragment.identifier\" attribute) integer "
+ + "between 0 and the value of the fragment.count attribute. If two or more FlowFiles have the same value for the "
+ + "\"fragment.identifier\" attribute and the same value for the \"fragment.index\" attribute, the behavior of this Processor is undefined."),
@ReadsAttribute(attribute = "fragment.count", description = "Applicable only if the <Merge Strategy> property is set to Defragment. This "
- + "attribute must be present on all FlowFiles with the same value for the fragment.identifier attribute. All FlowFiles in the same "
- + "bundle must have the same value for this attribute. The value of this attribute indicates how many FlowFiles should be expected "
- + "in the given bundle"),
+ + "attribute must be present on all FlowFiles with the same value for the fragment.identifier attribute. All FlowFiles in the same "
+ + "bundle must have the same value for this attribute. The value of this attribute indicates how many FlowFiles should be expected "
+ + "in the given bundle."),
@ReadsAttribute(attribute = "segment.original.filename", description = "Applicable only if the <Merge Strategy> property is set to Defragment. "
- + "This attribute must be present on all FlowFiles with the same value for the fragment.identifier attribute. All FlowFiles in the same "
- + "bundle must have the same value for this attribute. The value of this attribute will be used for the filename of the completed merged "
- + "FlowFile"),
+ + "This attribute must be present on all FlowFiles with the same value for the fragment.identifier attribute. All FlowFiles in the same "
+ + "bundle must have the same value for this attribute. The value of this attribute will be used for the filename of the completed merged "
+ + "FlowFile."),
@ReadsAttribute(attribute = "tar.permissions", description = "Applicable only if the <Merge Format> property is set to TAR. The value of this "
- + "attribute must be 3 characters; each character must be in the range 0 to 7 (inclusive) and indicates the file permissions that should "
- + "be used for the FlowFile's TAR entry. If this attribute is missing or has an invalid value, the default value of 644 will be used")})
+ + "attribute must be 3 characters; each character must be in the range 0 to 7 (inclusive) and indicates the file permissions that should "
+ + "be used for the FlowFile's TAR entry. If this attribute is missing or has an invalid value, the default value of 644 will be used") })
@WritesAttributes({
@WritesAttribute(attribute = "filename", description = "When more than 1 file is merged, the filename comes from the segment.original.filename "
- + "attribute. If that attribute does not exist in the source FlowFiles, then the filename is set to the number of nanoseconds matching "
- + "system time. Then a filename extension may be applied:"
- + "if Merge Format is TAR, then the filename will be appended with .tar, "
- + "if Merge Format is ZIP, then the filename will be appended with .zip, "
- + "if Merge Format is FlowFileStream, then the filename will be appended with .pkg"),
+ + "attribute. If that attribute does not exist in the source FlowFiles, then the filename is set to the number of nanoseconds matching "
+ + "system time. Then a filename extension may be applied:"
+ + "if Merge Format is TAR, then the filename will be appended with .tar, "
+ + "if Merge Format is ZIP, then the filename will be appended with .zip, "
+ + "if Merge Format is FlowFileStream, then the filename will be appended with .pkg"),
@WritesAttribute(attribute = "merge.count", description = "The number of FlowFiles that were merged into this bundle"),
@WritesAttribute(attribute = "merge.bin.age", description = "The age of the bin, in milliseconds, when it was merged and output. Effectively "
- + "this is the greatest amount of time that any FlowFile in this bundle remained waiting in this processor before it was output")})
+ + "this is the greatest amount of time that any FlowFile in this bundle remained waiting in this processor before it was output") })
@SeeAlso(SegmentContent.class)
public class MergeContent extends BinFiles {
@@ -131,7 +134,9 @@ public class MergeContent extends BinFiles {
"Defragment",
"Combines fragments that are associated by attributes back into a single cohesive FlowFile. If using this strategy, all FlowFiles must "
+ "have the attributes <fragment.identifier>, <fragment.count>, and <fragment.index> or alternatively (for backward compatibility "
- + "purposes) <segment.identifier>, <segment.count>, and <segment.index>");
+ + "purposes) <segment.identifier>, <segment.count>, and <segment.index>. All FlowFiles with the same value for \"fragment.identifier\" "
+ + "will be grouped together. All FlowFiles in this group must have the same value for the \"fragment.count\" attribute. All FlowFiles "
+ + "in this group must have a unique value for the \"fragment.index\" attribute between 0 and the value of the \"fragment.count\" attribute.");
public static final AllowableValue DELIMITER_STRATEGY_FILENAME = new AllowableValue(
"Filename", "Filename", "The values of Header, Footer, and Demarcator will be retrieved from the contents of a file");
@@ -307,7 +312,7 @@ public class MergeContent extends BinFiles {
@Override
protected Collection<ValidationResult> additionalCustomValidation(ValidationContext context) {
- Collection<ValidationResult> results = new ArrayList<>();
+ final Collection<ValidationResult> results = new ArrayList<>();
final String delimiterStrategy = context.getProperty(DELIMITER_STRATEGY).getValue();
if(DELIMITER_STRATEGY_FILENAME.equals(delimiterStrategy)) {
@@ -353,7 +358,7 @@ public class MergeContent extends BinFiles {
@Override
protected String getGroupId(final ProcessContext context, final FlowFile flowFile) {
final String correlationAttributeName = context.getProperty(CORRELATION_ATTRIBUTE_NAME).getValue();
- String groupId = (correlationAttributeName == null) ? null : flowFile.getAttribute(correlationAttributeName);
+ String groupId = correlationAttributeName == null ? null : flowFile.getAttribute(correlationAttributeName);
// when MERGE_STRATEGY is Defragment and correlationAttributeName is null then bin by fragment.identifier
if (groupId == null && MERGE_STRATEGY_DEFRAGMENT.equals(context.getProperty(MERGE_STRATEGY).getValue())) {
@@ -442,7 +447,7 @@ public class MergeContent extends BinFiles {
bundle = session.putAllAttributes(bundle, bundleAttributes);
- final String inputDescription = (binCopy.size() < 10) ? binCopy.toString() : binCopy.size() + " FlowFiles";
+ final String inputDescription = binCopy.size() < 10 ? binCopy.toString() : binCopy.size() + " FlowFiles";
getLogger().info("Merged {} into {}", new Object[]{inputDescription, bundle});
session.transfer(bundle, REL_MERGED);
@@ -640,18 +645,18 @@ public class MergeContent extends BinFiles {
}
if (".".equals(path.getName(0).toString())) {
- path = (path.getNameCount() == 1) ? null : path.subpath(1, path.getNameCount());
+ path = path.getNameCount() == 1 ? null : path.subpath(1, path.getNameCount());
}
- return (path == null) ? "" : path.toString() + "/";
+ return path == null ? "" : path.toString() + "/";
}
private String createFilename(final List<FlowFileSessionWrapper> wrappers) {
if (wrappers.size() == 1) {
return wrappers.get(0).getFlowFile().getAttribute(CoreAttributes.FILENAME.key());
} else {
- FlowFile ff = wrappers.get(0).getFlowFile();
- String origFilename = ff.getAttribute(SEGMENT_ORIGINAL_FILENAME);
+ final FlowFile ff = wrappers.get(0).getFlowFile();
+ final String origFilename = ff.getAttribute(SEGMENT_ORIGINAL_FILENAME);
if (origFilename != null) {
return origFilename;
} else {