lordgamez commented on code in PR #1987: URL: https://github.com/apache/nifi-minifi-cpp/pull/1987#discussion_r2472123752
########## core-framework/c-api-framework/src/core/ProcessorImpl.cpp: ########## @@ -0,0 +1,87 @@ +/** + * @file Processor.cpp + * Processor 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 "api/core/ProcessorImpl.h" + +#include <ctime> +#include <cctype> + +#include <memory> +#include <set> +#include <string> +#include <vector> + +#include "fmt/format.h" + +using namespace std::literals::chrono_literals; + +namespace org::apache::nifi::minifi::api::core { + +ProcessorImpl::ProcessorImpl(minifi::core::ProcessorMetadata metadata) + : metadata_(std::move(metadata)), + trigger_when_empty_(false), + // metrics_(std::make_shared<ProcessorMetricsImpl>(*this)), Review Comment: Comment can be removed ########## core-framework/c-api-framework/src/core/ProcessorImpl.cpp: ########## @@ -0,0 +1,87 @@ +/** + * @file Processor.cpp + * Processor class implementation Review Comment: This part can be removed, it does not match with the current filename anyway. ########## extensions/python/ExecutePythonProcessor.cpp: ########## @@ -38,9 +38,15 @@ namespace org::apache::nifi::minifi::extensions::python::processors { void ExecutePythonProcessor::initialize() { initializeScript(); - std::vector<core::PropertyReference> all_properties; - ranges::transform(python_properties_, std::back_inserter(all_properties), &core::Property::getReference); - setSupportedProperties(all_properties); + std::vector<core::Property> all_properties; + all_properties.reserve(Properties.size() + python_properties_.size()); + for (auto& property : Properties) { + all_properties.emplace_back(property); + } + for (auto& python_property : python_properties_) { + all_properties.emplace_back(python_property); + } + setSupportedProperties(gsl::make_span(all_properties)); Review Comment: Properties is currently empty and we only want to include supported properties that are defined in the python processors, why was this changed? ########## libminifi/include/utils/CProcessor.h: ########## @@ -0,0 +1,193 @@ +/** +* 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 <string> + +#include "minifi-cpp/core/Annotation.h" +#include "core/ProcessorMetrics.h" +#include "minifi-c/minifi-c.h" +#include "minifi-cpp/agent/agent_docs.h" +#include "minifi-cpp/core/ProcessContext.h" +#include "minifi-cpp/core/ProcessorApi.h" +#include "minifi-cpp/core/ProcessorDescriptor.h" +#include "minifi-cpp/core/ProcessorMetadata.h" + +namespace org::apache::nifi::minifi::utils { + +class CProcessor; + +class CProcessorMetricsWrapper : public minifi::core::ProcessorMetricsImpl { + public: + class CProcessorInfoProvider : public ProcessorMetricsImpl::ProcessorInfoProvider { + public: + explicit CProcessorInfoProvider(const CProcessor& source_processor): source_processor_(source_processor) {} + + std::string getProcessorType() const override; + std::string getName() const override; + minifi::utils::SmallString<36> getUUIDStr() const override; + + ~CProcessorInfoProvider() override = default; + + private: + const CProcessor& source_processor_; + }; + + explicit CProcessorMetricsWrapper(const CProcessor& source_processor) + : minifi::core::ProcessorMetricsImpl(std::make_unique<CProcessorInfoProvider>(source_processor)), + source_processor_(source_processor) { + } + + std::vector<minifi::state::response::SerializedResponseNode> serialize() override; + + std::vector<minifi::state::PublishedMetric> calculateMetrics() override; + + private: + const CProcessor& source_processor_; +}; + +struct CProcessorClassDescription { + std::string name; + std::vector<minifi::core::Property> class_properties; + std::vector<minifi::core::Relationship> class_relationships; + bool supports_dynamic_properties; + bool supports_dynamic_relationships; + minifi::core::annotation::Input input_requirement; + bool is_single_threaded; + + MinifiProcessorCallbacks callbacks; +}; + +class CProcessor : public minifi::core::ProcessorApi { + public: + CProcessor(CProcessorClassDescription class_description, minifi::core::ProcessorMetadata metadata) + : class_description_(std::move(class_description)), + metrics_(std::make_shared<CProcessorMetricsWrapper>(*this)) { + metadata_ = metadata; + MinifiProcessorMetadata c_metadata; + auto uuid_str = metadata.uuid.to_string(); + c_metadata.uuid = MinifiStringView{.data = uuid_str.data(), .length = gsl::narrow<uint32_t>(uuid_str.length())}; + c_metadata.name = MinifiStringView{.data = metadata.name.data(), .length = gsl::narrow<uint32_t>(metadata.name.length())}; + c_metadata.logger = reinterpret_cast<MinifiLogger>(&metadata_.logger); + impl_ = class_description_.callbacks.create(c_metadata); + } + CProcessor(CProcessorClassDescription class_description, minifi::core::ProcessorMetadata metadata, OWNED void* impl) + : class_description_(std::move(class_description)), + impl_(impl), + metadata_(metadata), + metrics_(std::make_shared<CProcessorMetricsWrapper>(*this)) {} + ~CProcessor() override { + class_description_.callbacks.destroy(impl_); + } + + bool isWorkAvailable() override { + return static_cast<bool>(class_description_.callbacks.isWorkAvailable(impl_)); + } + + void restore(const std::shared_ptr<minifi::core::FlowFile>& file) override { + class_description_.callbacks.restore(impl_, reinterpret_cast<OWNED MinifiFlowFile>(new std::shared_ptr<minifi::core::FlowFile>(file))); + } + + bool supportsDynamicProperties() const override { + return class_description_.supports_dynamic_properties; + } + + bool supportsDynamicRelationships() const override { + return class_description_.supports_dynamic_relationships; + } + + void initialize(minifi::core::ProcessorDescriptor& descriptor) override { + descriptor.setSupportedProperties(std::span(class_description_.class_properties)); + + std::vector<minifi::core::RelationshipDefinition> relationships; + for (auto& rel : class_description_.class_relationships) {relationships.push_back(rel.getDefinition());} + descriptor.setSupportedRelationships(relationships); + } + + bool isSingleThreaded() const override { + return class_description_.is_single_threaded; + } + + std::string getProcessorType() const override { + return class_description_.name; + } + + bool getTriggerWhenEmpty() const override { + return static_cast<bool>(class_description_.callbacks.getTriggerWhenEmpty(impl_)); + } + + void onTrigger(minifi::core::ProcessContext& process_context, minifi::core::ProcessSession& process_session) override { + std::optional<std::string> error; + auto status = class_description_.callbacks.onTrigger(impl_, reinterpret_cast<MinifiProcessContext>(&process_context), reinterpret_cast<MinifiProcessSession>(&process_session)); + if (status == MINIFI_PROCESSOR_YIELD) { + process_context.yield(); + return; + } + if (status != MINIFI_SUCCESS) { + throw minifi::Exception(minifi::ExceptionType::PROCESSOR_EXCEPTION, "Error while triggering processor"); + } + } + + void onSchedule(minifi::core::ProcessContext& process_context, minifi::core::ProcessSessionFactory& /*process_session_factory*/) override { + std::optional<std::string> error; + auto status = class_description_.callbacks.onSchedule(impl_, reinterpret_cast<MinifiProcessContext>(&process_context)); + if (status != MINIFI_SUCCESS) { + throw minifi::Exception(minifi::ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Error while scheduling processor"); + } + } + + void onUnSchedule() override { + class_description_.callbacks.onUnSchedule(impl_); + } + + void notifyStop() override { + class_description_.callbacks.onUnSchedule(impl_); + } + + minifi::core::annotation::Input getInputRequirement() const override { + return class_description_.input_requirement; + } + + gsl::not_null<std::shared_ptr<minifi::core::ProcessorMetrics>> getMetrics() const override { + return metrics_; + } + + std::vector<minifi::state::PublishedMetric> getCustomMetrics() const; + + void forEachLogger(const std::function<void(std::shared_ptr<minifi::core::logging::Logger>)>&) override { + // pass Review Comment: I suppose this comment is also unneeded ########## extensions/llamacpp/processors/DefaultLlamaContext.cpp: ########## @@ -88,7 +87,7 @@ std::optional<std::string> DefaultLlamaContext::applyTemplate(const std::vector< std::transform(messages.begin(), messages.end(), std::back_inserter(llama_messages), [](const LlamaChatMessage& msg) { return llama_chat_message{.role = msg.role.c_str(), .content = msg.content.c_str()}; }); std::string text; - text.resize(utils::configuration::DEFAULT_BUFFER_SIZE); + text.resize(4096); Review Comment: Why was the default buffer size constant removed here? ########## extensions/llamacpp/tests/RunLlamaCppInferenceTests.cpp: ########## @@ -15,12 +15,29 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -#include "unit/TestBase.h" -#include "unit/Catch.h" #include "RunLlamaCppInference.h" -#include "unit/SingleProcessorTestController.h" +#include "api/core/Resource.h" #include "minifi-cpp/core/FlowFile.h" -#include "unit/ProcessorUtils.h" +#include "unit/Catch.h" +#include "unit/SingleProcessorTestController.h" +#include "unit/TestBase.h" +#include "utils/CProcessor.h" +#include "unit/TestUtils.h" + +namespace org::apache::nifi::minifi::api::test::utils { + +template<typename T, typename ...Args> +std::unique_ptr<minifi::core::Processor> make_custom_processor(minifi::core::ProcessorMetadata metadata, Args&&... args) { // NOLINT(cppcoreguidelines-missing-std-forward) Review Comment: Should this be moved to a common utils file? I suppose this will be reused in the future for other processor tests too. ########## libminifi/src/core/ProcessSession.cpp: ########## @@ -223,14 +223,15 @@ void ProcessSessionImpl::penalize(const std::shared_ptr<core::FlowFile> &flow) { } void ProcessSessionImpl::transfer(const std::shared_ptr<core::FlowFile>& flow, const Relationship& relationship) { - logger_->log_debug("Transferring {} from {} to relationship {}", flow->getUUIDStr(), process_context_->getProcessor().getName(), relationship.getName()); - utils::Identifier uuid = flow->getUUID(); + auto flow_file = std::dynamic_pointer_cast<FlowFile>(flow); Review Comment: We should check at least with an assert that this is not null ########## libminifi/src/minifi-c.cpp: ########## @@ -0,0 +1,502 @@ +/** +* 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 "minifi-c/minifi-c.h" + +#include <memory> +#include <vector> + +#include "agent/agent_docs.h" +#include "core/ProcessorMetrics.h" +#include "minifi-cpp/core/Annotation.h" +#include "minifi-cpp/core/ClassLoader.h" +#include "minifi-cpp/core/ProcessContext.h" +#include "minifi-cpp/core/ProcessSession.h" +#include "minifi-cpp/core/ProcessorApi.h" +#include "minifi-cpp/core/ProcessorDescriptor.h" +#include "minifi-cpp/core/ProcessorFactory.h" +#include "minifi-cpp/core/ProcessorMetadata.h" +#include "minifi-cpp/core/PropertyValidator.h" +#include "minifi-cpp/core/logging/Logger.h" +#include "minifi-cpp/core/state/PublishedMetricProvider.h" +#include "minifi-cpp/Exception.h" +#include "minifi-cpp/core/extension/ExtensionManager.h" +#include "utils/PropertyErrors.h" +#include "minifi-cpp/agent/build_description.h" +#include "utils/CProcessor.h" + +namespace minifi = org::apache::nifi::minifi; + +namespace { + +std::string toString(MinifiStringView sv) { + return {sv.data, sv.length}; +} + +std::string_view toStringView(MinifiStringView sv) { + return {sv.data, sv.length}; +} + +minifi::core::annotation::Input toInputRequirement(MinifiInputRequirement req) { + switch (req) { + case MINIFI_INPUT_REQUIRED: return minifi::core::annotation::Input::INPUT_REQUIRED; + case MINIFI_INPUT_ALLOWED: return minifi::core::annotation::Input::INPUT_ALLOWED; + case MINIFI_INPUT_FORBIDDEN: return minifi::core::annotation::Input::INPUT_FORBIDDEN; + } + gsl_FailFast(); +} + +minifi::core::logging::LOG_LEVEL toLogLevel(MinifiLogLevel lvl) { + switch (lvl) { + case MINIFI_TRACE: return minifi::core::logging::trace; + case MINIFI_DEBUG: return minifi::core::logging::debug; + case MINIFI_INFO: return minifi::core::logging::info; + case MINIFI_WARNING: return minifi::core::logging::warn; + case MINIFI_ERROR: return minifi::core::logging::err; + case MINIFI_CRITICAL: return minifi::core::logging::critical; + case MINIFI_OFF: return minifi::core::logging::off; + } + gsl_FailFast(); +} + +minifi::core::Property createProperty(const MinifiProperty* property_description) { + gsl_Expects(property_description); + std::vector<std::string_view> allowed_values; + allowed_values.reserve(property_description->allowed_values_count); + for (size_t i = 0; i < property_description->allowed_values_count; ++i) { + allowed_values.push_back(toStringView(property_description->allowed_values_ptr[i])); + } + std::vector<std::string_view> allowed_types; + allowed_types.reserve(property_description->types_count); + for (size_t i = 0; i < property_description->types_count; ++i) { + allowed_types.push_back(toStringView(property_description->types_ptr[i])); + } + std::vector<std::string_view> dependent_properties; + dependent_properties.reserve(property_description->dependent_properties_count); + for (size_t i = 0; i < property_description->dependent_properties_count; ++i) { + dependent_properties.push_back(toStringView(property_description->dependent_properties_ptr[i])); + } + std::vector<std::pair<std::string_view, std::string_view>> exclusive_of_properties; + exclusive_of_properties.reserve(property_description->exclusive_of_properties_count); + for (size_t i = 0; i < property_description->exclusive_of_properties_count; ++i) { + exclusive_of_properties.push_back({toStringView(property_description->exclusive_of_property_names_ptr[i]), toStringView(property_description->exclusive_of_property_values_ptr[i])}); + } + std::optional<std::string_view> default_value; + if (property_description->default_value) { + default_value = toStringView(*property_description->default_value); + } + return minifi::core::Property{minifi::core::PropertyReference{ + toStringView(property_description->name), + toStringView(property_description->display_name), + toStringView(property_description->description), + static_cast<bool>(property_description->is_required), + static_cast<bool>(property_description->is_sensitive), + std::span(allowed_values), + std::span(allowed_types), + std::span(dependent_properties), + std::span(exclusive_of_properties), + default_value, + gsl::make_not_null(reinterpret_cast<const minifi::core::PropertyValidator*>(property_description->validator)), + static_cast<bool>(property_description->supports_expression_language) + }}; +} + +class CProcessorFactory : public minifi::core::ProcessorFactory { + public: + CProcessorFactory(std::string group_name, std::string class_name, minifi::utils::CProcessorClassDescription class_description) + : group_name_(std::move(group_name)), + class_name_(std::move(class_name)), + class_description_(std::move(class_description)) {} + std::unique_ptr<minifi::core::ProcessorApi> create(minifi::core::ProcessorMetadata metadata) override { + return std::make_unique<minifi::utils::CProcessor>(class_description_, metadata); + } + + [[nodiscard]] std::string getGroupName() const override { + return group_name_; + } + + [[nodiscard]] std::string getClassName() const override { + return class_name_; + } + + CProcessorFactory() = delete; + CProcessorFactory(const CProcessorFactory&) = delete; + CProcessorFactory& operator=(const CProcessorFactory&) = delete; + CProcessorFactory(CProcessorFactory&&) = delete; + CProcessorFactory& operator=(CProcessorFactory&&) = delete; + ~CProcessorFactory() override = default; + + private: + std::string group_name_; + std::string class_name_; + minifi::utils::CProcessorClassDescription class_description_; +}; + +class CExtension : public minifi::core::extension::Extension { + public: + CExtension(std::string name, MinifiBool(*initializer)(void*, MinifiConfigure), void* user_data): name_(std::move(name)), initializer_(initializer), user_data_(user_data) {} + bool initialize(const minifi::core::extension::ExtensionConfig& config) override { + gsl_Assert(initializer_); + return static_cast<bool>(initializer_(user_data_, reinterpret_cast<MinifiConfigure>(config.get()))); + } + + [[nodiscard]] const std::string& getName() const override { + return name_; + } + + private: + std::string name_; + MinifiBool(*initializer_)(void*, MinifiConfigure); + void* user_data_; +}; + +} // namespace + +namespace org::apache::nifi::minifi::utils { + +void useCProcessorClassDescription(const MinifiProcessorClassDescription* class_description, const std::function<void(minifi::ClassDescription, minifi::utils::CProcessorClassDescription)>& fn) { + std::vector<minifi::core::Property> properties; + properties.reserve(class_description->class_properties_count); + for (size_t i = 0; i < class_description->class_properties_count; ++i) { + properties.push_back(createProperty(&class_description->class_properties_ptr[i])); + } + std::vector<minifi::core::DynamicProperty> dynamic_properties; + dynamic_properties.reserve(class_description->dynamic_properties_count); + for (size_t i = 0; i < class_description->dynamic_properties_count; ++i) { + dynamic_properties.push_back(minifi::core::DynamicProperty{ + .name = toStringView(class_description->dynamic_properties_ptr[i].name), + .value = toStringView(class_description->dynamic_properties_ptr[i].value), + .description = toStringView(class_description->dynamic_properties_ptr[i].description), + .supports_expression_language = static_cast<bool>(class_description->dynamic_properties_ptr[i].supports_expression_language) + }); + } + std::vector<minifi::core::Relationship> relationships; + relationships.reserve(class_description->class_relationships_count); + for (size_t i = 0; i < class_description->class_relationships_count; ++i) { + relationships.push_back(minifi::core::Relationship{ + toString(class_description->class_relationships_ptr[i].name), + toString(class_description->class_relationships_ptr[i].description) + }); + } + std::vector<std::vector<minifi::core::RelationshipDefinition>> output_attribute_relationships; + std::vector<minifi::core::OutputAttributeReference> output_attributes; + for (size_t i = 0; i < class_description->output_attributes_count; ++i) { + minifi::core::OutputAttributeReference ref{minifi::core::OutputAttributeDefinition{"", {}, ""}}; + ref.name = toStringView(class_description->output_attributes_ptr[i].name); + ref.description = toStringView(class_description->output_attributes_ptr[i].description); + output_attribute_relationships.push_back({}); + for (size_t j = 0; j < class_description->output_attributes_ptr[i].relationships_count; ++j) { + output_attribute_relationships.back().push_back(minifi::core::RelationshipDefinition{ + .name = toStringView(class_description->output_attributes_ptr[i].relationships_ptr[j].name), + .description = toStringView(class_description->output_attributes_ptr[i].relationships_ptr[j].description) + }); + } + ref.relationships = std::span(output_attribute_relationships.back()); + output_attributes.push_back(ref); + } + + auto name_segments = minifi::utils::string::split(toStringView(class_description->full_name), "::"); + gsl_Assert(!name_segments.empty()); + + minifi::ClassDescription description{ + .type_ = minifi::ResourceType::Processor, + .short_name_ = name_segments.back(), + .full_name_ = minifi::utils::string::join(".", name_segments), + .description_ = toString(class_description->description), + .class_properties_ = properties, + .dynamic_properties_ = dynamic_properties, + .class_relationships_ = relationships, + .output_attributes_ = output_attributes, + .supports_dynamic_properties_ = static_cast<bool>(class_description->supports_dynamic_properties), + .supports_dynamic_relationships_ = static_cast<bool>(class_description->supports_dynamic_relationships), + .inputRequirement_ = minifi::core::annotation::toString(toInputRequirement(class_description->input_requirement)), + .isSingleThreaded_ = static_cast<bool>(class_description->is_single_threaded), + }; + + minifi::utils::CProcessorClassDescription c_class_description{ + .name = name_segments.back(), + .class_properties = properties, + .class_relationships = relationships, + .supports_dynamic_properties = description.supports_dynamic_properties_, + .supports_dynamic_relationships = description.supports_dynamic_relationships_, + .input_requirement = toInputRequirement(class_description->input_requirement), + .is_single_threaded = description.isSingleThreaded_, + + .callbacks = class_description->callbacks + }; + + fn(description, c_class_description); +} + +} // namespace org::apache::nifi::minifi::utils + +extern "C" { + +MinifiPropertyValidator MinifiGetStandardValidator(MinifiStandardPropertyValidator validator) { + switch (validator) { + case MINIFI_ALWAYS_VALID_VALIDATOR: return reinterpret_cast<MinifiPropertyValidator>(&minifi::core::StandardPropertyValidators::ALWAYS_VALID_VALIDATOR); + case MINIFI_NON_BLANK_VALIDATOR: return reinterpret_cast<MinifiPropertyValidator>(&minifi::core::StandardPropertyValidators::NON_BLANK_VALIDATOR); + case MINIFI_TIME_PERIOD_VALIDATOR: return reinterpret_cast<MinifiPropertyValidator>(&minifi::core::StandardPropertyValidators::TIME_PERIOD_VALIDATOR); + case MINIFI_BOOLEAN_VALIDATOR: return reinterpret_cast<MinifiPropertyValidator>(&minifi::core::StandardPropertyValidators::BOOLEAN_VALIDATOR); + case MINIFI_INTEGER_VALIDATOR: return reinterpret_cast<MinifiPropertyValidator>(&minifi::core::StandardPropertyValidators::INTEGER_VALIDATOR); + case MINIFI_UNSIGNED_INTEGER_VALIDATOR: return reinterpret_cast<MinifiPropertyValidator>(&minifi::core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR); + case MINIFI_DATA_SIZE_VALIDATOR: return reinterpret_cast<MinifiPropertyValidator>(&minifi::core::StandardPropertyValidators::DATA_SIZE_VALIDATOR); + case MINIFI_PORT_VALIDATOR: return reinterpret_cast<MinifiPropertyValidator>(&minifi::core::StandardPropertyValidators::PORT_VALIDATOR); + default: gsl_FailFast(); + } +} + +void MinifiRegisterProcessorClass(const MinifiProcessorClassDescription* class_description) { + gsl_Expects(class_description); + + auto module_name = toString(class_description->module_name); + minifi::BundleDetails bundle{ + .artifact = module_name, + .group = module_name, + .version = "1.0.0" Review Comment: We shouldn't keep this 1.0.0 hardcoded value, could the version be part of the class description API? -- 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]
