adamdebreceni commented on a change in pull request #1137: URL: https://github.com/apache/nifi-minifi-cpp/pull/1137#discussion_r689611731
########## File path: extensions/standard-processors/processors/AttributesToJSON.h ########## @@ -0,0 +1,106 @@ +/** + * @file AttributesToJSON.h + * AttributesToJSON class declaration + * + * 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. + */ +#pragma once + +#include <vector> +#include <string> +#include <set> +#include <unordered_set> +#include <memory> +#include <map> +#include <regex> + +#include "rapidjson/document.h" +#include "core/Processor.h" +#include "core/Property.h" +#include "core/logging/Logger.h" +#include "utils/Enum.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace processors { + +class AttributesToJSON : public core::Processor { + public: + // Supported Properties + static const core::Property AttributesList; + static const core::Property AttributesRegularExpression; + static const core::Property Destination; + static const core::Property IncludeCoreAttributes; + static const core::Property NullValue; + + // Supported Relationships + static core::Relationship Success; + + SMART_ENUM(WriteDestination, + (FLOWFILE_ATTRIBUTE, "flowfile-attribute"), + (FLOWFILE_CONTENT, "flowfile-content") + ) + + explicit AttributesToJSON(const std::string& name, const utils::Identifier& uuid = {}) + : core::Processor(name, uuid), + logger_(logging::LoggerFactory<AttributesToJSON>::getLogger()), + core_attributes_(core::SpecialFlowAttribute::getSpecialFlowAttributes()) { + } + + void initialize() override; + void onSchedule(core::ProcessContext *context, core::ProcessSessionFactory* sessionFactory) override; + void onTrigger(core::ProcessContext *context, core::ProcessSession *session) override; + + core::annotation::Input getInputRequirement() const override { + return core::annotation::Input::INPUT_REQUIRED; + } + + private: + class WriteCallback : public OutputStreamCallback { + public: + explicit WriteCallback(const std::string& json_data) : json_data_(json_data) {} + int64_t process(const std::shared_ptr<io::BaseStream>& stream) override { + const auto write_ret = stream->write(reinterpret_cast<const uint8_t*>(json_data_.data()), json_data_.length()); + return io::isError(write_ret) ? -1 : gsl::narrow<int64_t>(write_ret); + } + private: + std::string json_data_; + }; + + bool isCoreAttributeToBeFiltered(const std::string& attribute) const; + std::unordered_set<std::string> getAttributesToBeWritten(const std::map<std::string, std::string>& flowfile_attributes) const; + void addAttributeToJson(rapidjson::Document& document, const std::string& key, const std::string& value); + std::string buildAttributeJsonData(std::map<std::string, std::string>&& flowfile_attributes); + + std::shared_ptr<logging::Logger> logger_; + const std::unordered_set<std::string> core_attributes_; + std::vector<std::string> attribute_list_; + std::optional<std::regex> attributes_regular_expression_; + WriteDestination write_destination_; + bool include_core_attributes_ = true; + bool null_value_ = false; +}; + +REGISTER_RESOURCE(AttributesToJSON, "Generates a JSON representation of the input FlowFile Attributes. " Review comment: note: after the dynamic extension PR is merged these should migrate to the source files, no need to move it now, but remind me in case I forget to do so on rebase :) -- 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]
