This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new 7cc7b265f MINIFICPP-2764 Controller Service CPP API (#2152)
7cc7b265f is described below
commit 7cc7b265f65afec70850f664d9a70dee6c0cfe6c
Author: Martin Zink <[email protected]>
AuthorDate: Thu May 7 07:32:03 2026 +0200
MINIFICPP-2764 Controller Service CPP API (#2152)
* useProcessorClassDescription -> useProcessorClassDefinition
* add explanatory comments to ControllerServiceImpl.h
* notifyStop -> disable in CPP API aswell
---
Extensions.md | 4 +-
...ProcessContext.h => ControllerServiceContext.h} | 17 +++--
.../include/api/core/ControllerServiceImpl.h | 63 +++++++++++++++++++
.../include/api/core/ProcessContext.h | 3 +-
.../include/api/core/ProcessorImpl.h | 7 +--
.../cpp-extension-lib/include/api/core/Resource.h | 72 +++++++++++++++++-----
.../include/api/utils/ProcessorConfigUtils.h | 24 ++++++++
.../libtest/CProcessorTestUtils.h | 30 +++++++--
...essContext.cpp => ControllerServiceContext.cpp} | 23 ++++---
.../src/core/ControllerServiceImpl.cpp | 65 +++++++++++++++++++
.../cpp-extension-lib/src/core/ProcessContext.cpp | 13 ++++
.../llamacpp/processors/ExtensionInitializer.cpp | 4 +-
libminifi/include/utils/CControllerService.h | 9 ++-
libminifi/src/minifi-c.cpp | 2 +-
.../extension-verification-test/CApiExtension.cpp | 4 +-
libminifi/test/libtest/unit/TestBase.cpp | 64 ++++++++++++-------
libminifi/test/libtest/unit/TestBase.h | 3 +-
17 files changed, 325 insertions(+), 82 deletions(-)
diff --git a/Extensions.md b/Extensions.md
index 51077009f..81df822a1 100644
--- a/Extensions.md
+++ b/Extensions.md
@@ -48,8 +48,8 @@ extern "C" void MinifiInitExtension(MinifiExtensionContext*
extension_context) {
.user_data = nullptr
};
auto* extension = MinifiRegisterExtension(extension_context,
&extension_definition);
-
minifi::api::core::useProcessorClassDescription<minifi::extensions::llamacpp::processors::RunLlamaCppInference>([&]
(const MinifiProcessorClassDefinition& description) {
- MinifiRegisterProcessor(extension, &description);
+
minifi::api::core::useProcessorClassDefinition<minifi::extensions::llamacpp::processors::RunLlamaCppInference>([&]
(const MinifiProcessorClassDefinition& definition) {
+ MinifiRegisterProcessor(extension, &definition);
});
}
```
diff --git
a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
b/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceContext.h
similarity index 61%
copy from
extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
copy to
extension-framework/cpp-extension-lib/include/api/core/ControllerServiceContext.h
index 442f9bbf4..561bef0ac 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
+++
b/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceContext.h
@@ -1,5 +1,5 @@
/**
-* Licensed to the Apache Software Foundation (ASF) under one or more
+ * 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
@@ -22,23 +22,20 @@
#include "minifi-c.h"
#include "minifi-cpp/core/PropertyDefinition.h"
-#include "api/core/FlowFile.h"
namespace org::apache::nifi::minifi::api::core {
-class ProcessContext {
+class ControllerServiceContext {
public:
- explicit ProcessContext(MinifiProcessContext* impl): impl_(impl) {}
+ explicit ControllerServiceContext(MinifiControllerServiceContext* impl) :
impl_(impl) {}
- std::expected<std::string, std::error_code> getProperty(std::string_view
name, const FlowFile* flow_file = nullptr) const;
- std::expected<std::string, std::error_code> getProperty(const
minifi::core::PropertyReference& property_reference, const FlowFile* flow_file
= nullptr) const {
- return getProperty(property_reference.name, flow_file);
+ [[nodiscard]] std::expected<std::string, std::error_code>
getProperty(std::string_view name) const;
+ [[nodiscard]] std::expected<std::string, std::error_code> getProperty(const
minifi::core::PropertyReference& property_reference) const {
+ return getProperty(property_reference.name);
}
- bool hasNonEmptyProperty(std::string_view name) const;
-
private:
- MinifiProcessContext* impl_;
+ MinifiControllerServiceContext* impl_;
};
} // namespace org::apache::nifi::minifi::api::core
diff --git
a/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceImpl.h
b/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceImpl.h
new file mode 100644
index 000000000..b4b747384
--- /dev/null
+++
b/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceImpl.h
@@ -0,0 +1,63 @@
+/**
+ * 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 <memory>
+#include <string>
+
+#include "ControllerServiceContext.h"
+#include "minifi-cpp/core/ControllerServiceMetadata.h"
+#include "minifi-cpp/utils/Id.h"
+#include "utils/SmallString.h"
+
+namespace org::apache::nifi::minifi::api {
+
+class Connection;
+
+namespace core {
+
+/// Helper class that makes implementing the stable API easier
+/// It is used in conjunction with useControllerServiceClassDefinition
+class ControllerServiceImpl {
+ public:
+ explicit ControllerServiceImpl(minifi::core::ControllerServiceMetadata
metadata);
+
+ ControllerServiceImpl(const ControllerServiceImpl&) = delete;
+ ControllerServiceImpl(ControllerServiceImpl&&) = delete;
+ ControllerServiceImpl& operator=(const ControllerServiceImpl&) = delete;
+ ControllerServiceImpl& operator=(ControllerServiceImpl&&) = delete;
+
+ virtual ~ControllerServiceImpl();
+
+ MinifiStatus enable(ControllerServiceContext&);
+ void disable();
+
+ [[nodiscard]] std::string getName() const;
+ [[nodiscard]] minifi::utils::Identifier getUUID() const;
+ [[nodiscard]] minifi::utils::SmallString<36> getUUIDStr() const;
+
+ protected:
+ virtual MinifiStatus enableImpl(api::core::ControllerServiceContext&) = 0;
+ virtual void disableImpl() {}
+
+ minifi::core::ControllerServiceMetadata metadata_;
+
+ std::shared_ptr<minifi::core::logging::Logger> logger_;
+};
+
+} // namespace core
+} // namespace org::apache::nifi::minifi::api
diff --git
a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
b/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
index 442f9bbf4..6f012d719 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
@@ -34,8 +34,9 @@ class ProcessContext {
std::expected<std::string, std::error_code> getProperty(const
minifi::core::PropertyReference& property_reference, const FlowFile* flow_file
= nullptr) const {
return getProperty(property_reference.name, flow_file);
}
+ [[nodiscard]] std::expected<MinifiControllerService*, std::error_code>
getControllerService(std::string_view controller_service_name, std::string_view
controller_service_class) const;
- bool hasNonEmptyProperty(std::string_view name) const;
+ [[nodiscard]] bool hasNonEmptyProperty(std::string_view name) const;
private:
MinifiProcessContext* impl_;
diff --git
a/extension-framework/cpp-extension-lib/include/api/core/ProcessorImpl.h
b/extension-framework/cpp-extension-lib/include/api/core/ProcessorImpl.h
index 2a5c90d52..f5869715e 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessorImpl.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessorImpl.h
@@ -17,9 +17,7 @@
#pragma once
#include <atomic>
-#include <condition_variable>
#include <memory>
-#include <mutex>
#include <string>
#include <vector>
@@ -53,7 +51,7 @@ class ProcessorImpl {
virtual ~ProcessorImpl();
- void setTriggerWhenEmpty(bool trigger_when_empty) {
+ void setTriggerWhenEmpty(const bool trigger_when_empty) {
trigger_when_empty_ = trigger_when_empty;
}
@@ -87,9 +85,6 @@ class ProcessorImpl {
std::atomic<bool> trigger_when_empty_;
std::shared_ptr<minifi::core::logging::Logger> logger_;
-
- private:
- mutable std::mutex mutex_;
};
} // namespace core
diff --git a/extension-framework/cpp-extension-lib/include/api/core/Resource.h
b/extension-framework/cpp-extension-lib/include/api/core/Resource.h
index ef934af95..8187605e5 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/Resource.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/Resource.h
@@ -26,19 +26,21 @@
#define WIN32_LEAN_AND_MEAN 1
#endif
-#include "minifi-c.h"
-#include "core/ClassName.h"
-#include "api/utils/minifi-c-utils.h"
+#include "ControllerServiceContext.h"
+#include "FlowFile.h"
#include "ProcessContext.h"
#include "ProcessSession.h"
-#include "FlowFile.h"
-#include "minifi-cpp/core/ProcessorMetadata.h"
+#include "api/utils/minifi-c-utils.h"
+#include "core/ClassName.h"
#include "logging/Logger.h"
+#include "minifi-c.h"
+#include "minifi-cpp/core/ControllerServiceMetadata.h"
+#include "minifi-cpp/core/ProcessorMetadata.h"
namespace org::apache::nifi::minifi::api::core {
template<typename Class, typename Fn>
-void useProcessorClassDescription(Fn&& fn) {
+void useProcessorClassDefinition(Fn&& fn) {
std::vector<std::vector<MinifiStringView>> string_vector_cache;
const auto full_name = minifi::core::className<Class>();
@@ -76,7 +78,7 @@ void useProcessorClassDescription(Fn&& fn) {
attribute_relationships_cache.push_back(std::move(rel_cache));
}
- MinifiProcessorClassDefinition description{
+ MinifiProcessorClassDefinition definition{
.full_name = utils::toStringView(full_name),
.description = utils::toStringView(Class::Description),
.class_properties_count = gsl::narrow<uint32_t>(class_properties.size()),
@@ -94,11 +96,12 @@ void useProcessorClassDescription(Fn&& fn) {
.callbacks = MinifiProcessorCallbacks{
.create = [] (MinifiProcessorMetadata metadata) -> MINIFI_OWNED void* {
- return new Class{minifi::core::ProcessorMetadata{
- .uuid =
minifi::utils::Identifier::parse(std::string{metadata.uuid.data,
metadata.uuid.length}).value(),
- .name = std::string{metadata.name.data, metadata.name.length},
- .logger = std::make_shared<logging::Logger>(metadata.logger)
- }};
+ try {
+ return new Class{minifi::core::ProcessorMetadata{
+ .uuid =
minifi::utils::Identifier::parse(std::string{metadata.uuid.data,
metadata.uuid.length}).value(),
+ .name = std::string{metadata.name.data, metadata.name.length},
+ .logger = std::make_shared<logging::Logger>(metadata.logger)}};
+ } catch (...) { return nullptr; }
},
.destroy = [] (MINIFI_OWNED void* self) -> void {
delete static_cast<Class*>(self);
@@ -124,7 +127,9 @@ void useProcessorClassDescription(Fn&& fn) {
}
},
.onUnSchedule = [] (void* self) -> void {
- static_cast<Class*>(self)->onUnSchedule();
+ try {
+ static_cast<Class*>(self)->onUnSchedule();
+ } catch (...) {}
},
.calculateMetrics = [] (void* self) -> MINIFI_OWNED
MinifiPublishedMetrics* {
auto metrics = static_cast<Class*>(self)->calculateMetrics();
@@ -139,7 +144,46 @@ void useProcessorClassDescription(Fn&& fn) {
}
};
- fn(description);
+ fn(definition);
+}
+
+template<typename Class, typename Fn>
+void useControllerServiceClassDefinition(Fn&& fn) {
+ std::vector<std::vector<MinifiStringView>> string_vector_cache;
+
+ const auto full_name = minifi::core::className<Class>();
+
+ std::vector<MinifiPropertyDefinition> class_properties =
utils::toProperties(Class::Properties, string_vector_cache);
+
+ MinifiControllerServiceClassDefinition definition{.full_name =
utils::toStringView(full_name),
+ .description = utils::toStringView(Class::Description),
+ .class_properties_count = gsl::narrow<uint32_t>(class_properties.size()),
+ .class_properties_ptr = class_properties.data(),
+
+ .callbacks = MinifiControllerServiceCallbacks{
+ .create = [](MinifiControllerServiceMetadata metadata) ->
MINIFI_OWNED void* {
+ try {
+ return new Class{minifi::core::ControllerServiceMetadata{
+ .uuid =
minifi::utils::Identifier::parse(std::string{metadata.uuid.data,
metadata.uuid.length}).value(),
+ .name = std::string{metadata.name.data,
metadata.name.length},
+ .logger =
std::make_shared<logging::Logger>(metadata.logger)}};
+ } catch (...) { return nullptr; }
+ },
+ .destroy = [](MINIFI_OWNED void* self) -> void { delete
static_cast<Class*>(self); },
+ .enable = [](void* self, MinifiControllerServiceContext* context) ->
MinifiStatus {
+ ControllerServiceContext context_wrapper(context);
+ try {
+ return static_cast<Class*>(self)->enable(context_wrapper);
+ } catch (...) { return MINIFI_STATUS_UNKNOWN_ERROR; }
+ },
+ .disable = [](void* self) -> void {
+ try {
+ static_cast<Class*>(self)->disable();
+ } catch (...) {}
+ },
+ }};
+
+ fn(definition);
}
} // namespace org::apache::nifi::minifi::api::core
diff --git
a/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
b/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
index 6a14699d7..a73dd7549 100644
---
a/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
+++
b/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
@@ -167,4 +167,28 @@ std::optional<T> parseOptionalEnumProperty(const
core::ProcessContext& context,
return result.value();
}
+template<typename ControllerServiceType>
+ControllerServiceType* parseOptionalControllerService(const
core::ProcessContext& context, const minifi::core::PropertyReference& prop) {
+ const auto controller_service_name = context.getProperty(prop.name);
+ if (!controller_service_name || controller_service_name->empty()) {
+ return nullptr;
+ }
+
+ auto service = context.getControllerService(*controller_service_name,
minifi::core::className<ControllerServiceType>());
+ if (!service) {
+ return nullptr;
+ }
+
+ return reinterpret_cast<ControllerServiceType*>(*service);
+}
+
+template<typename ControllerServiceType>
+gsl::not_null<ControllerServiceType*> parseControllerService(const
core::ProcessContext& context, const minifi::core::PropertyReference& prop) {
+ auto controller_service =
parseOptionalControllerService<ControllerServiceType>(context, prop);
+ if (!controller_service) {
+ throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("Required
controller service property '{}' is missing", prop.name));
+ }
+ return gsl::make_not_null(controller_service);
+}
+
} // namespace org::apache::nifi::minifi::api::utils
diff --git
a/extension-framework/cpp-extension-lib/libtest/CProcessorTestUtils.h
b/extension-framework/cpp-extension-lib/libtest/CProcessorTestUtils.h
index 142f4709e..2defdff5c 100644
--- a/extension-framework/cpp-extension-lib/libtest/CProcessorTestUtils.h
+++ b/extension-framework/cpp-extension-lib/libtest/CProcessorTestUtils.h
@@ -1,5 +1,5 @@
/**
-*
+ *
* 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.
@@ -17,22 +17,40 @@
*/
#pragma once
+#include "api/core/Resource.h"
#include "core/Processor.h"
+#include "core/controller/ControllerService.h"
+#include "minifi-cpp/core/ControllerServiceMetadata.h"
#include "minifi-cpp/core/ProcessorMetadata.h"
-#include "api/core/Resource.h"
+#include "utils/CControllerService.h"
#include "utils/CProcessor.h"
+#include "minifi-cpp/agent/agent_docs.h"
namespace org::apache::nifi::minifi::test::utils {
-template<typename T, typename ...Args>
-std::unique_ptr<minifi::core::Processor>
make_custom_c_processor(minifi::core::ProcessorMetadata metadata, Args&&...
args) { // NOLINT(cppcoreguidelines-missing-std-forward)
+template<typename T, typename... Args>
+std::unique_ptr<minifi::core::Processor>
make_custom_c_processor(minifi::core::ProcessorMetadata metadata,
+ Args&&... args) { // NOLINT(cppcoreguidelines-missing-std-forward)
std::unique_ptr<minifi::core::ProcessorApi> processor_impl;
- minifi::api::core::useProcessorClassDescription<T>([&] (const
MinifiProcessorClassDefinition& description) {
- minifi::utils::useCProcessorClassDescription(description, [&] (const
auto&, auto c_description) {
+ minifi::api::core::useProcessorClassDefinition<T>([&](const
MinifiProcessorClassDefinition& definition) {
+ minifi::utils::useCProcessorClassDescription(definition, [&](const auto&,
auto c_description) {
processor_impl =
std::make_unique<minifi::utils::CProcessor>(std::move(c_description), metadata,
new T(metadata, std::forward<Args>(args)...));
});
});
return std::make_unique<minifi::core::Processor>(metadata.name,
metadata.uuid, std::move(processor_impl));
}
+template<typename T, typename... Args>
+std::shared_ptr<minifi::core::controller::ControllerService>
make_custom_c_controller_service(minifi::core::ControllerServiceMetadata
metadata,
+ Args&&... args) { // NOLINT(cppcoreguidelines-missing-std-forward)
+ std::unique_ptr<minifi::core::controller::ControllerServiceApi>
controller_service_impl;
+ minifi::api::core::useControllerServiceClassDefinition<T>([&](const
MinifiControllerServiceClassDefinition& definition) {
+ minifi::utils::useCControllerServiceClassDescription(definition, [&](const
auto&, auto c_description) {
+ controller_service_impl =
std::make_unique<minifi::utils::CControllerService>(std::move(c_description),
+ metadata,
+ new T(metadata, std::forward<Args>(args)...));
+ });
+ });
+ return
std::make_shared<minifi::core::controller::ControllerService>(metadata.name,
metadata.uuid, std::move(controller_service_impl));
+}
} // namespace org::apache::nifi::minifi::test::utils
diff --git a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
b/extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp
similarity index 63%
copy from extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
copy to
extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp
index 958274d7d..089112c04 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
+++
b/extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp
@@ -1,5 +1,5 @@
/**
-* Licensed to the Apache Software Foundation (ASF) under one or more
+ * 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
@@ -15,27 +15,26 @@
* limitations under the License.
*/
-#include "api/core/ProcessContext.h"
+#include "api/core/ControllerServiceContext.h"
#include "api/utils/minifi-c-utils.h"
-#include "api/core/FlowFile.h"
namespace org::apache::nifi::minifi::api::core {
-std::expected<std::string, std::error_code>
ProcessContext::getProperty(std::string_view name, const FlowFile* flow_file)
const {
- std::optional<std::string> value;
- MinifiStatus status = MinifiProcessContextGetProperty(impl_,
utils::toStringView(name), flow_file ? flow_file->get() : MINIFI_NULL,
- [] (void* data, MinifiStringView result) {
+std::expected<std::string, std::error_code>
ControllerServiceContext::getProperty(const std::string_view name) const {
+ std::optional<std::string> value = std::nullopt;
+ const MinifiStatus status = MinifiControllerServiceContextGetProperty(impl_,
utils::toStringView(name),
+ [] (void* data, const MinifiStringView result) {
(*static_cast<std::optional<std::string>*>(data)) =
std::string(result.data, result.length);
}, &value);
- if (!value) {
+ if (status != MINIFI_STATUS_SUCCESS) {
return std::unexpected{utils::make_error_code(status)};
}
- return value.value();
-}
-bool ProcessContext::hasNonEmptyProperty(std::string_view name) const {
- return MinifiProcessContextHasNonEmptyProperty(impl_,
utils::toStringView(name));
+ if (!value) {
+ return
std::unexpected{utils::make_error_code(MINIFI_STATUS_UNKNOWN_ERROR)};
+ }
+ return value.value();
}
} // namespace org::apache::nifi::minifi::api::core
diff --git
a/extension-framework/cpp-extension-lib/src/core/ControllerServiceImpl.cpp
b/extension-framework/cpp-extension-lib/src/core/ControllerServiceImpl.cpp
new file mode 100644
index 000000000..920dabf97
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/src/core/ControllerServiceImpl.cpp
@@ -0,0 +1,65 @@
+/**
+ * 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/ControllerServiceImpl.h"
+
+#include <memory>
+#include <string>
+
+#include "fmt/format.h"
+namespace org::apache::nifi::minifi::api::core {
+
+ControllerServiceImpl::ControllerServiceImpl(minifi::core::ControllerServiceMetadata
metadata)
+ : metadata_(std::move(metadata)),
+ logger_(metadata_.logger) {
+ logger_->log_debug("ControllerService {} created with uuid {}", getName(),
getUUIDStr());
+}
+
+ControllerServiceImpl::~ControllerServiceImpl() {
+ logger_->log_debug("Destroying controller service {} with uuid {}",
getName(), getUUIDStr());
+}
+
+MinifiStatus ControllerServiceImpl::enable(ControllerServiceContext& ctx) {
+ try {
+ return enableImpl(ctx);
+ } catch (const std::exception& e) {
+ logger_->log_error("{}", e.what());
+ throw;
+ }
+}
+
+void ControllerServiceImpl::disable() {
+ try {
+ disableImpl();
+ } catch (const std::exception& e) {
+ logger_->log_error("{}", e.what());
+ throw;
+ }
+}
+
+std::string ControllerServiceImpl::getName() const {
+ return metadata_.name;
+}
+
+utils::Identifier ControllerServiceImpl::getUUID() const {
+ return metadata_.uuid;
+}
+
+utils::SmallString<36> ControllerServiceImpl::getUUIDStr() const {
+ return getUUID().to_string();
+}
+
+} // namespace org::apache::nifi::minifi::api::core
diff --git a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
index 958274d7d..d260b9c0c 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
@@ -38,4 +38,17 @@ bool ProcessContext::hasNonEmptyProperty(std::string_view
name) const {
return MinifiProcessContextHasNonEmptyProperty(impl_,
utils::toStringView(name));
}
+std::expected<MinifiControllerService*, std::error_code>
ProcessContext::getControllerService(const std::string_view
controller_service_name,
+ const std::string_view controller_service_class) const {
+ MinifiControllerService* controller_service = nullptr;
+ if (const MinifiStatus status =
MinifiProcessContextGetControllerService(impl_,
+ utils::toStringView(controller_service_name),
+ utils::toStringView(controller_service_class),
+ &controller_service);
+ status != MINIFI_STATUS_SUCCESS) {
+ return std::unexpected{utils::make_error_code(status)};
+ }
+ return controller_service;
+}
+
} // namespace org::apache::nifi::minifi::api::core
diff --git a/extensions/llamacpp/processors/ExtensionInitializer.cpp
b/extensions/llamacpp/processors/ExtensionInitializer.cpp
index f11b53956..85d610a7d 100644
--- a/extensions/llamacpp/processors/ExtensionInitializer.cpp
+++ b/extensions/llamacpp/processors/ExtensionInitializer.cpp
@@ -34,7 +34,7 @@ CEXTENSIONAPI void
MinifiInitExtension(MinifiExtensionContext* extension_context
.user_data = nullptr
};
auto* extension = MinifiRegisterExtension(extension_context,
&extension_definition);
-
minifi::api::core::useProcessorClassDescription<minifi::extensions::llamacpp::processors::RunLlamaCppInference>([&]
(const MinifiProcessorClassDefinition& description) {
- MinifiRegisterProcessor(extension, &description);
+
minifi::api::core::useProcessorClassDefinition<minifi::extensions::llamacpp::processors::RunLlamaCppInference>([&]
(const MinifiProcessorClassDefinition& definition) {
+ MinifiRegisterProcessor(extension, &definition);
});
}
diff --git a/libminifi/include/utils/CControllerService.h
b/libminifi/include/utils/CControllerService.h
index 9dea54dbe..6e000548d 100644
--- a/libminifi/include/utils/CControllerService.h
+++ b/libminifi/include/utils/CControllerService.h
@@ -26,8 +26,12 @@
#include "minifi-cpp/core/Property.h"
#include "minifi-cpp/core/controller/ControllerServiceApi.h"
+namespace org::apache::nifi::minifi {
+struct ClassDescription;
+}
+
namespace org::apache::nifi::minifi::utils {
-class CControllerService;
+
struct CControllerServiceClassDescription {
@@ -108,7 +112,6 @@ class CControllerService final : public
core::controller::ControllerServiceApi,
};
void useCControllerServiceClassDescription(const
MinifiControllerServiceClassDefinition& class_description,
- const BundleIdentifier& bundle_id,
- const std::function<void(ClassDescription,
CControllerServiceClassDescription)>& fn);
+ const std::function<void(const ClassDescription&,
CControllerServiceClassDescription)>& fn);
} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/src/minifi-c.cpp b/libminifi/src/minifi-c.cpp
index 90ccb969b..54a3dba8f 100644
--- a/libminifi/src/minifi-c.cpp
+++ b/libminifi/src/minifi-c.cpp
@@ -255,7 +255,7 @@ void useCProcessorClassDescription(const
MinifiProcessorClassDefinition& class_d
}
void useCControllerServiceClassDescription(const
MinifiControllerServiceClassDefinition& class_description,
- const std::function<void(ClassDescription,
CControllerServiceClassDescription)>& fn) {
+ const std::function<void(const ClassDescription&,
CControllerServiceClassDescription)>& 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) {
diff --git
a/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
b/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
index 2b2b981ae..546a22905 100644
--- a/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
+++ b/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
@@ -50,7 +50,7 @@ CEXTENSIONAPI void
MinifiInitExtension(MinifiExtensionContext* extension_context
.user_data = nullptr
};
auto extension = MinifiRegisterExtension(extension_context,
&extension_definition);
- minifi::api::core::useProcessorClassDescription<DummyCProcessor>([&] (const
MinifiProcessorClassDefinition& description) {
- MinifiRegisterProcessor(extension, &description);
+ minifi::api::core::useProcessorClassDefinition<DummyCProcessor>([&] (const
MinifiProcessorClassDefinition& definition) {
+ MinifiRegisterProcessor(extension, &definition);
});
}
diff --git a/libminifi/test/libtest/unit/TestBase.cpp
b/libminifi/test/libtest/unit/TestBase.cpp
index fad99f2c4..23a967ae9 100644
--- a/libminifi/test/libtest/unit/TestBase.cpp
+++ b/libminifi/test/libtest/unit/TestBase.cpp
@@ -24,30 +24,30 @@
#include <sstream>
#include <utility>
-#include "core/Processor.h"
+#include "Connection.h"
+#include "LogUtils.h"
+#include "ProvenanceTestHelper.h"
+#include "ResourceClaim.h"
+#include "TestUtils.h"
#include "core/ProcessContextImpl.h"
-#include "minifi-cpp/core/PropertyDefinition.h"
+#include "core/ProcessSessionFactory.h"
+#include "core/Processor.h"
+#include "core/controller/StandardControllerServiceNode.h"
+#include "core/controller/StandardControllerServiceProvider.h"
#include "core/logging/LoggerConfiguration.h"
#include "core/state/nodes/FlowInformation.h"
-#include "core/controller/StandardControllerServiceProvider.h"
-#include "ProvenanceTestHelper.h"
+#include "fmt/format.h"
+#include "io/StreamPipe.h"
+#include "minifi-cpp/core/ProcessContext.h"
+#include "minifi-cpp/core/PropertyDefinition.h"
+#include "spdlog/sinks/dist_sink.h"
+#include "spdlog/sinks/ostream_sink.h"
+#include "spdlog/sinks/stdout_sinks.h"
#include "utils/ClassUtils.h"
-#include "TestUtils.h"
+#include "utils/GeneralUtils.h"
#include "utils/Id.h"
#include "utils/StringUtils.h"
#include "utils/span.h"
-#include "LogUtils.h"
-#include "utils/GeneralUtils.h"
-#include "Connection.h"
-#include "minifi-cpp/core/ProcessContext.h"
-#include "core/ProcessSessionFactory.h"
-#include "ResourceClaim.h"
-#include "io/StreamPipe.h"
-
-#include "fmt/format.h"
-#include "spdlog/sinks/stdout_sinks.h"
-#include "spdlog/sinks/ostream_sink.h"
-#include "spdlog/sinks/dist_sink.h"
std::shared_ptr<LogTestController> LogTestController::getInstance(const
std::shared_ptr<logging::LoggerProperties>& logger_properties) {
static std::map<std::shared_ptr<logging::LoggerProperties>,
std::shared_ptr<LogTestController>> map;
@@ -360,6 +360,27 @@ minifi::Connection*
TestPlan::addConnection(minifi::core::Processor* source_proc
return retVal;
}
+std::shared_ptr<minifi::core::controller::ControllerServiceNode>
TestPlan::addController(const std::string& name, const
std::shared_ptr<core::controller::ControllerService>& cs) {
+ if (finalized) {
+ return nullptr;
+ }
+ std::lock_guard<std::recursive_mutex> guard(mutex);
+
+ std::shared_ptr<minifi::core::controller::ControllerServiceNode>
controller_service_node =
std::make_shared<core::controller::StandardControllerServiceNode>(cs, name,
configuration_);
+
+ controller_service_nodes_.push_back(controller_service_node);
+
+ minifi::utils::Identifier uuid =
minifi::utils::IdGenerator::getIdGenerator()->generate();
+
+ controller_service_node->initialize();
+ controller_service_node->setUUID(uuid);
+
+ root_process_group_->addControllerService(name, controller_service_node,
uuid.to_string());
+ controller_services_provider_->putControllerServiceNode(name,
controller_service_node, root_process_group_.get(), uuid.to_string());
+
+ return controller_service_node;
+}
+
std::shared_ptr<minifi::core::controller::ControllerServiceNode>
TestPlan::addController(const std::string &controller_name, const std::string
&name) {
if (finalized) {
return nullptr;
@@ -486,7 +507,7 @@ std::shared_ptr<minifi::core::ProcessContext>
TestPlan::getProcessContextForProc
}
void TestPlan::scheduleProcessor(minifi::core::Processor* processor, const
std::shared_ptr<minifi::core::ProcessContext>& context) {
- if (std::find(configured_processors_.begin(), configured_processors_.end(),
processor) == configured_processors_.end()) {
+ if (std::ranges::find(configured_processors_, processor) ==
configured_processors_.end()) {
// Ordering on factories and list of configured processors do not matter
const auto factory =
std::make_shared<minifi::core::ProcessSessionFactoryImpl>(context);
factories_.push_back(factory);
@@ -507,7 +528,7 @@ void TestPlan::scheduleProcessors() {
}
}
-bool TestPlan::runProcessor(minifi::core::Processor* processor, const
PreTriggerVerifier& verify) {
+bool TestPlan::runProcessor(const minifi::core::Processor* processor, const
PreTriggerVerifier& verify) {
const auto processor_location =
gsl::narrow<size_t>(std::distance(processor_queue_.begin(),
getProcessorItByUuid(processor->getUUIDStr())));
return runProcessor(processor_location, verify);
}
@@ -578,7 +599,7 @@ bool
TestPlan::runCurrentProcessorUntilFlowfileIsProduced(std::chrono::milliseco
const auto isFlowFileProduced = [&] {
runCurrentProcessor();
const std::vector<minifi::Connection*> connections =
getProcessorOutboundConnections(processor_queue_.at(location));
- return std::any_of(connections.cbegin(), connections.cend(), [] (const
minifi::Connection* conn) { return !conn->isEmpty(); });
+ return std::ranges::any_of(connections, [] (const minifi::Connection*
conn) { return !conn->isEmpty(); });
};
return verifyEventHappenedInPollTime(wait_duration, isFlowFileProduced);
}
@@ -602,8 +623,7 @@ std::shared_ptr<minifi::core::FlowFile>
TestPlan::getFlowFileProducedByCurrentPr
std::vector<minifi::Connection*> connections =
getProcessorOutboundConnections(processor);
for (auto connection : connections) {
std::set<std::shared_ptr<minifi::core::FlowFile>> expiredFlowRecords;
- std::shared_ptr<minifi::core::FlowFile> flowfile =
connection->poll(expiredFlowRecords);
- if (flowfile) {
+ if (std::shared_ptr<minifi::core::FlowFile> flowfile =
connection->poll(expiredFlowRecords)) {
return flowfile;
}
if (expiredFlowRecords.empty()) {
diff --git a/libminifi/test/libtest/unit/TestBase.h
b/libminifi/test/libtest/unit/TestBase.h
index f3b85eb63..9e5557194 100644
--- a/libminifi/test/libtest/unit/TestBase.h
+++ b/libminifi/test/libtest/unit/TestBase.h
@@ -275,6 +275,7 @@ class TestPlan {
minifi::Connection* addConnection(minifi::core::Processor* source_proc,
const minifi::core::Relationship& source_relationship, minifi::core::Processor*
destination_proc);
+ std::shared_ptr<minifi::core::controller::ControllerServiceNode>
addController(const std::string& name, const
std::shared_ptr<minifi::core::controller::ControllerService>& cs);
std::shared_ptr<minifi::core::controller::ControllerServiceNode>
addController(const std::string &controller_name, const std::string &name);
bool setProperty(minifi::core::Processor* processor, const
core::PropertyReference& property, std::string_view value);
@@ -298,7 +299,7 @@ class TestPlan {
void scheduleProcessor(minifi::core::Processor* processor);
void scheduleProcessors();
- bool runProcessor(minifi::core::Processor* processor, const
PreTriggerVerifier& verify = nullptr);
+ bool runProcessor(const minifi::core::Processor* processor, const
PreTriggerVerifier& verify = nullptr);
bool runProcessor(size_t target_location, const PreTriggerVerifier& verify =
nullptr);
bool runNextProcessor(const PreTriggerVerifier& verify = nullptr);
bool runCurrentProcessor();
