hunyadi-dev commented on a change in pull request #866:
URL: https://github.com/apache/nifi-minifi-cpp/pull/866#discussion_r468533549
##########
File path: extensions/libarchive/MergeContent.cpp
##########
@@ -252,36 +255,43 @@ bool MergeContent::processBin(core::ProcessContext
*context, core::ProcessSessio
}
std::unique_ptr<MergeBin> mergeBin;
- if (mergeFormat_ == MERGE_FORMAT_CONCAT_VALUE || mergeFormat_ ==
MERGE_FORMAT_TAR_VALUE
- || mergeFormat_ == MERGE_FORMAT_ZIP_VALUE) {
- if (mergeFormat_ == MERGE_FORMAT_CONCAT_VALUE)
- mergeBin = std::unique_ptr < MergeBin > (new BinaryConcatenationMerge());
- else if (mergeFormat_ == MERGE_FORMAT_TAR_VALUE)
- mergeBin = std::unique_ptr < MergeBin > (new TarMerge());
- else if (mergeFormat_ == MERGE_FORMAT_ZIP_VALUE)
- mergeBin = std::unique_ptr < MergeBin > (new ZipMerge());
- else
- return false;
-
- std::shared_ptr<core::FlowFile> mergeFlow;
- try {
- mergeFlow = mergeBin->merge(context, session, bin->getFlowFile(),
this->headerContent_, this->footerContent_, this->demarcatorContent_);
- } catch (...) {
- logger_->log_error("Merge Content merge catch exception");
- return false;
- }
- session->putAttribute(mergeFlow, BinFiles::FRAGMENT_COUNT_ATTRIBUTE,
std::to_string(bin->getSize()));
- // we successfully merge the flow
- session->transfer(mergeFlow, Merge);
- std::deque<std::shared_ptr<core::FlowFile>> &flows = bin->getFlowFile();
- for (auto flow : flows) {
- session->transfer(flow, Original);
- }
- logger_->log_info("Merge FlowFile record UUID %s, payload length %d",
mergeFlow->getUUIDStr(), mergeFlow->getSize());
- } else {
+ if (mergeFormat_ == MERGE_FORMAT_CONCAT_VALUE)
+ mergeBin = utils::make_unique<BinaryConcatenationMerge>();
+ else if (mergeFormat_ == MERGE_FORMAT_TAR_VALUE)
+ mergeBin = utils::make_unique<TarMerge>();
+ else if (mergeFormat_ == MERGE_FORMAT_ZIP_VALUE)
+ mergeBin = utils::make_unique<ZipMerge>();
+ else {
logger_->log_error("Merge format not supported %s", mergeFormat_);
return false;
}
+
+ std::shared_ptr<core::FlowFile> mergeFlow;
+ try {
+ mergeFlow = mergeBin->merge(context, session, bin->getFlowFile(),
this->headerContent_, this->footerContent_, this->demarcatorContent_);
+ } catch (...) {
+ logger_->log_error("Merge Content merge catch exception");
+ return false;
+ }
+ session->putAttribute(mergeFlow, BinFiles::FRAGMENT_COUNT_ATTRIBUTE,
std::to_string(bin->getSize()));
+
+ if (attributeStrategy_ == ATTRIBUTE_STRATEGY_KEEP_COMMON)
+
KeepOnlymergedAttributesMerger(bin->getFlowFile()).mergeAttributes(session,
mergeFlow);
+ else if (attributeStrategy_ == ATTRIBUTE_STRATEGY_KEEP_ALL_UNIQUE)
+ KeepAllUniqueAttributesMerger(bin->getFlowFile()).mergeAttributes(session,
mergeFlow);
+ else {
+ logger_->log_error("Attribute strategy not supported %s",
attributeStrategy_);
+ throw minifi::Exception(ExceptionType::PROCESSOR_EXCEPTION, "Invalid
attribute strategy: " + attributeStrategy_);
Review comment:
I agree with this behaviour, but just in case, can you check, what the
NiFi guys do in similar cases?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
