lordgamez commented on a change in pull request #1137:
URL: https://github.com/apache/nifi-minifi-cpp/pull/1137#discussion_r686579921
##########
File path: docker/test/integration/features/attributes_to_json.feature
##########
@@ -0,0 +1,15 @@
+Feature: Writing attribute data using AttributesToJSON processor
+ Background:
+ Given the content of "/tmp/output" is monitored
+
+ Scenario: Write selected attribute data to file
+ Given a GetFile processor with the "Input Directory" property set to
"/tmp/input"
+ And a file with filename "test_file.log" and content "test_data%" is
present in "/tmp/input"
Review comment:
I think this line was just copied from a test where `%` was testing
special characters, but it is not relevant here. Removed in
dc9f6aa438e986a8ae6d1bacb85ca4fc677cb175
##########
File path: extensions/standard-processors/processors/AttributesToJSON.cpp
##########
@@ -0,0 +1,174 @@
+/**
+ * @file AttributesToJSON.cpp
+ * AttributesToJSON class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include "AttributesToJSON.h"
+
+#include <unordered_set>
+
+#include "rapidjson/writer.h"
+#include "utils/StringUtils.h"
+#include "utils/ProcessorConfigUtils.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace processors {
+
+const core::Property AttributesToJSON::AttributesList(
+ core::PropertyBuilder::createProperty("Attributes List")
+ ->withDescription("Comma separated list of attributes to be included in
the resulting JSON. "
+ "If this value is left empty then all existing
Attributes will be included. This list of attributes is case sensitive. "
+ "If an attribute specified in the list is not found it
will be be emitted to the resulting JSON with an empty string or NULL value.")
+ ->build());
+
+const core::Property AttributesToJSON::AttributesRegularExpression(
+ core::PropertyBuilder::createProperty("Attributes Regular Expression")
+ ->withDescription("Regular expression that will be evaluated against the
flow file attributes to select the matching attributes. "
+ "Both the matching attributes and the selected
attributes from the Attributes List property will be written in the resulting
JSON.")
+ ->build());
+
+const core::Property AttributesToJSON::Destination(
+ core::PropertyBuilder::createProperty("Destination")
+ ->withDescription("Control if JSON value is written as a new flowfile
attribute 'JSONAttributes' or written in the flowfile content. "
+ "Writing to flowfile content will overwrite any existing
flowfile content.")
+ ->isRequired(true)
+
->withDefaultValue<std::string>(toString(WriteDestination::FLOWFILE_ATTRIBUTE))
+ ->withAllowableValues<std::string>(WriteDestination::values())
+ ->build());
+
+const core::Property AttributesToJSON::IncludeCoreAttributes(
+ core::PropertyBuilder::createProperty("Include Core Attributes")
+ ->withDescription("Determines if the FlowFile core attributes which are
contained in every FlowFile should be included in the final JSON value
generated.")
+ ->isRequired(true)
+ ->withDefaultValue<bool>(true)
+ ->build());
+
+const core::Property AttributesToJSON::NullValue(
+ core::PropertyBuilder::createProperty("Null Value")
+ ->withDescription("If true a non existing or empty attribute will be NULL
in the resulting JSON. If false an empty string will be placed in the JSON.")
+ ->isRequired(true)
+ ->withDefaultValue<bool>(false)
+ ->build());
+
+core::Relationship AttributesToJSON::Success("success", "All FlowFiles
received are routed to success");
+
+void AttributesToJSON::initialize() {
+ setSupportedProperties({
+ AttributesList,
+ AttributesRegularExpression,
+ Destination,
+ IncludeCoreAttributes,
+ NullValue
+ });
+ setSupportedRelationships({Success});
+}
+
+void AttributesToJSON::onSchedule(core::ProcessContext* context,
core::ProcessSessionFactory* /*sessionFactory*/) {
+ std::string value;
+ if (context->getProperty(AttributesList.getName(), value) && !value.empty())
{
+ attribute_list_ = utils::StringUtils::splitAndTrimRemovingEmpty(value,
",");
+ }
+ if (context->getProperty(AttributesRegularExpression.getName(), value) &&
!value.empty()) {
+ attributes_regular_expression_ = std::regex(value);
+ }
+ write_destination_ =
WriteDestination::parse(utils::parsePropertyWithAllowableValuesOrThrow(*context,
Destination.getName(), WriteDestination::values()).c_str());
+ context->getProperty(IncludeCoreAttributes.getName(),
include_core_attributes_);
+ context->getProperty(NullValue.getName(), null_value_);
+}
+
+bool AttributesToJSON::isCoreAttributeToBeFiltered(const std::string&
attribute) const {
+ return !include_core_attributes_ && core_attributes_.find(attribute) !=
core_attributes_.end();
+}
+
+std::unordered_set<std::string>
AttributesToJSON::getAttributesToBeWritten(const std::map<std::string,
std::string>& flowfile_attributes) const {
+ std::unordered_set<std::string> attributes;
+
+ for (const auto& attribute : attribute_list_) {
+ if (!isCoreAttributeToBeFiltered(attribute)) {
+ attributes.insert(attribute);
+ }
+ }
+
+ if (attributes_regular_expression_) {
+ for (const auto& [key, value] : flowfile_attributes) {
+ if (!isCoreAttributeToBeFiltered(key) && std::regex_search(key,
attributes_regular_expression_.value())) {
Review comment:
You are right, updated the code and the tests in
dc9f6aa438e986a8ae6d1bacb85ca4fc677cb175
##########
File path: extensions/standard-processors/tests/unit/AttributesToJSONTests.cpp
##########
@@ -0,0 +1,293 @@
+/**
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <string>
+#include <vector>
+
+#include "rapidjson/writer.h"
+#include "rapidjson/stringbuffer.h"
+#include "TestBase.h"
+#include "utils/TestUtils.h"
+#include "AttributesToJSON.h"
+#include "GetFile.h"
+#include "PutFile.h"
+#include "UpdateAttribute.h"
+#include "LogAttribute.h"
+
+namespace {
+
+class AttributesToJSONTestFixture {
+ public:
+ const std::string TEST_FILE_CONTENT = "test_content";
+ const std::string TEST_FILE_NAME = "tstFile.ext";
+
+ AttributesToJSONTestFixture() {
+ LogTestController::getInstance().setTrace<TestPlan>();
+
LogTestController::getInstance().setDebug<minifi::processors::AttributesToJSON>();
+ LogTestController::getInstance().setDebug<minifi::processors::GetFile>();
+ LogTestController::getInstance().setDebug<minifi::processors::PutFile>();
+
LogTestController::getInstance().setDebug<minifi::processors::UpdateAttribute>();
+
LogTestController::getInstance().setDebug<minifi::processors::LogAttribute>();
+
+ dir_ = test_controller_.createTempDirectory();
+
+ plan_ = test_controller_.createPlan();
+ getfile_ = plan_->addProcessor("GetFile", "GetFile");
+ update_attribute_ = plan_->addProcessor("UpdateAttribute",
"UpdateAttribute", core::Relationship("success", "description"), true);
+ attribute_to_json_ = plan_->addProcessor("AttributesToJSON",
"AttributesToJSON", core::Relationship("success", "description"), true);
+ logattribute_ = plan_->addProcessor("LogAttribute", "LogAttribute",
core::Relationship("success", "description"), true);
+ putfile_ = plan_->addProcessor("PutFile", "PutFile",
core::Relationship("success", "description"), true);
+
+ plan_->setProperty(getfile_,
org::apache::nifi::minifi::processors::GetFile::Directory.getName(), dir_);
+ plan_->setProperty(putfile_,
org::apache::nifi::minifi::processors::PutFile::Directory.getName(), dir_);
+
+ update_attribute_->setDynamicProperty("my_attribute", "my_value");
+ update_attribute_->setDynamicProperty("other_attribute", "other_value");
+ update_attribute_->setDynamicProperty("empty_attribute", "");
+
+ utils::putFileToDir(dir_, TEST_FILE_NAME, TEST_FILE_CONTENT);
+ }
+
+ void assertJSONAttributesFromLog(const std::unordered_map<std::string,
std::optional<std::string>>& expected_attributes) {
+ auto match =
LogTestController::getInstance().matchesRegex("key:JSONAttributes
value:(.*)").value();
+ assertAttributes(expected_attributes, match[1].str());
Review comment:
Updated in dc9f6aa438e986a8ae6d1bacb85ca4fc677cb175
##########
File path: libminifi/test/TestBase.h
##########
@@ -164,6 +165,31 @@ class LogTestController {
return found;
}
+ std::optional<std::smatch> matchesRegex(const std::string ®ex_str,
+ std::chrono::seconds timeout = std::chrono::seconds(3),
+ std::chrono::milliseconds sleep_interval =
std::chrono::milliseconds(200)) {
+ if (regex_str.length() == 0) {
+ return std::nullopt;
+ }
+ auto start = std::chrono::system_clock::now();
+ bool found = false;
+ bool timed_out = false;
+ std::regex matcher_regex(regex_str);
+ std::smatch match;
+ do {
+ std::string str = log_output.str();
+ found = std::regex_search(str, match, matcher_regex);
+ auto now = std::chrono::system_clock::now();
+ timed_out = std::chrono::duration_cast<std::chrono::milliseconds>(now -
start) > std::chrono::duration_cast<std::chrono::milliseconds>(timeout);
Review comment:
Updated in dc9f6aa438e986a8ae6d1bacb85ca4fc677cb175
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
