You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2026/07/07 12:14:09 UTC
(nifi-minifi-cpp) branch main updated: MINIFICPP-2850 Finalize stable API before 1.0 (#2205)
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 ee49532b2 MINIFICPP-2850 Finalize stable API before 1.0 (#2205)
ee49532b2 is described below
commit ee49532b20daa0f26eabe0267a52163dac32c3e6
Author: Martin Zink <ma...@apache.org>
AuthorDate: Tue Jul 7 14:14:03 2026 +0200
MINIFICPP-2850 Finalize stable API before 1.0 (#2205)
This commit finalizes the “stable” MiNiFi C/C++ extension API surface in preparation for a 1.0 release by renaming and normalizing exported C symbols/headers, updating the C++ extension framework wrappers, and migrating extensions/tests to the new API (including new metrics reporting and trigger-when-empty control paths).
* removed MINIFI_NULL
* removed struct typedefs
* snake_case type and function names
* removed stdbool.h, in C23, bool, true, and false are keywords
* PropertyDefinition: type -> allowed_type
* getTriggerWhenEmpty member -> setTriggerWhenEmpty free function, to be called from onSchedule
* onTrigger -> trigger, onSchedule -> schedule, onUnSchedule -> unschedule
* calculate_metrics -> call report_metrics on process_context instead of callback
* processor_class_definition: removed class_ prefixes from members
* minifi-c.h -> minifi-api.h
* removed get/setTriggerWhenEmpty from ProcessorApi
* moved MinifiLoggerSetMaxLogSize, MinifiLoggerLevel to AdvancedLogger
* moved AdvancedLogger.h to minifi-api/include/minifi-cpp/core/logging/AdvancedLogger.h
* minifi_processor_class_definition -> minifi_processor_definition
* minifi_controller_service_class_definition -> minifi_controller_service_definition
* removed <= c23 workarounds
* removed unused macro magic
---
.clang-format | 9 +-
.../ubuntu_22_04_clang_arm_manifest.json | 16 +-
.github/workflows/create-release-artifacts.yml | 4 +-
Extensions.md | 34 +--
core-framework/include/core/ProcessorImpl.h | 10 -
core-framework/src/core/ProcessorImpl.cpp | 1 -
.../include/api/core/ControllerServiceContext.h | 8 +-
.../include/api/core/ControllerServiceImpl.h | 4 +-
.../cpp-extension-lib/include/api/core/FlowFile.h | 6 +-
.../include/api/core/ProcessContext.h | 19 +-
.../include/api/core/ProcessSession.h | 8 +-
.../include/api/core/ProcessorImpl.h | 20 +-
.../cpp-extension-lib/include/api/core/Resource.h | 87 +++---
.../include/api/core/logging/Logger.h | 8 +-
.../include/api/utils/minifi-c-utils.h | 31 +-
.../libtest/CProcessorTestUtils.h | 6 +-
.../cpp-extension-lib/mocklib/include/MockLogger.h | 2 -
.../mocklib/include/MockProcessContext.h | 13 +-
.../mocklib/include/MockProcessSession.h | 24 +-
.../mocklib/src/MockProcessContext.cpp | 6 +-
.../mocklib/src/MockProcessSession.cpp | 8 +-
.../mocklib/src/mock-minifi-api.cpp | 149 ++++++++++
.../mocklib/src/mock-minifi-c.cpp | 142 ----------
.../src/core/ControllerServiceContext.cpp | 4 +-
.../src/core/ControllerServiceImpl.cpp | 2 +-
.../cpp-extension-lib/src/core/ProcessContext.cpp | 79 ++++--
.../cpp-extension-lib/src/core/ProcessSession.cpp | 46 +--
.../cpp-extension-lib/src/core/ProcessorImpl.cpp | 5 +-
.../cpp-extension-lib/src/core/logging/Logger.cpp | 28 +-
.../cpp-extension-lib/src/utils/minifi-c-utils.cpp | 4 +-
.../include/utils/ExtensionInitUtils.h | 13 +-
extensions/ExtensionInitializer.cpp | 10 +-
extensions/aws/AwsLoader.cpp | 21 +-
extensions/aws/utils/AWSSdkLogger.cpp | 6 +-
extensions/gcp/ExtensionInitializer.cpp | 8 +-
.../GCPCredentialsControllerService.cpp | 2 +-
.../GCPCredentialsControllerService.h | 2 +-
extensions/gcp/processors/DeleteGCSObject.cpp | 2 +-
extensions/gcp/processors/DeleteGCSObject.h | 2 +-
extensions/gcp/processors/FetchGCSObject.cpp | 4 +-
extensions/gcp/processors/FetchGCSObject.h | 4 +-
extensions/gcp/processors/GCSProcessor.cpp | 2 +-
extensions/gcp/processors/GCSProcessor.h | 4 +-
extensions/gcp/processors/ListGCSBucket.cpp | 10 +-
extensions/gcp/processors/ListGCSBucket.h | 4 +-
extensions/gcp/processors/PutGCSObject.cpp | 6 +-
extensions/gcp/processors/PutGCSObject.h | 4 +-
extensions/kafka/ConsumeKafka.cpp | 10 +-
extensions/kafka/ConsumeKafka.h | 8 +-
extensions/kafka/ExtensionInitializer.cpp | 8 +-
extensions/kafka/KafkaProcessorBase.h | 4 +-
extensions/kafka/PublishKafka.cpp | 4 +-
extensions/kafka/PublishKafka.h | 4 +-
extensions/kafka/tests/PublishKafkaTests.cpp | 2 +-
extensions/libarchive/BinFiles.cpp | 2 +-
extensions/llamacpp/ExtensionInitializer.cpp | 8 +-
.../llamacpp/processors/RunLlamaCppInference.cpp | 7 +-
.../llamacpp/processors/RunLlamaCppInference.h | 5 +-
extensions/python/PythonCreator.h | 7 +-
.../python/pythonlibloader/PythonLibLoader.cpp | 16 +-
extensions/python/pythonloader/PyProcLoader.cpp | 14 +-
extensions/sftp/SFTPLoader.cpp | 10 +-
.../AnimalControllerServices.cpp | 2 +-
.../stable-api-testing/AnimalControllerServices.h | 4 +-
.../stable-api-testing/ExtensionInitializer.cpp | 8 +-
extensions/stable-api-testing/ZooProcessor.cpp | 4 +-
extensions/stable-api-testing/ZooProcessor.h | 4 +-
.../processors/DefragmentText.cpp | 2 +-
.../standard-processors/processors/InvokeHTTP.cpp | 1 +
.../standard-processors/processors/InvokeHTTP.h | 1 -
.../standard-processors/processors/LogAttribute.h | 8 +-
.../tests/unit/ProcessorTests.cpp | 3 +-
libminifi/include/RemoteProcessGroupPort.h | 88 +++---
libminifi/include/core/ProcessContextImpl.h | 1 +
libminifi/include/core/Processor.h | 2 +
libminifi/include/core/extension/Extension.h | 4 +-
libminifi/include/core/logging/LoggerBase.h | 22 +-
.../reporting/SiteToSiteProvenanceReportingTask.h | 3 +-
libminifi/include/utils/CControllerService.h | 16 +-
libminifi/include/utils/CProcessor.h | 36 +--
libminifi/src/RemoteProcessGroupPort.cpp | 6 +
libminifi/src/c2/HeartbeatLogger.cpp | 4 +-
libminifi/src/core/ProcessContextImpl.cpp | 3 +
libminifi/src/core/ProcessSession.cpp | 2 +-
libminifi/src/core/Processor.cpp | 8 +-
libminifi/src/core/extension/ApiVersion.cpp | 3 +-
libminifi/src/core/extension/Extension.cpp | 22 +-
.../src/core/flow/StructuredConfiguration.cpp | 3 +-
.../SiteToSiteProvenanceReportingTask.cpp | 3 +-
.../src/core/state/nodes/AgentInformation.cpp | 2 +-
libminifi/src/{minifi-c.cpp => minifi-api.cpp} | 314 ++++++++++-----------
libminifi/src/utils/CProcessor.cpp | 7 +-
.../extension-verification-test/CApiExtension.cpp | 12 +-
.../CppApiExtension.cpp | 4 +-
.../CreateNotCalled.cpp | 6 +-
.../ExtensionVerificationTests.cpp | 8 +-
.../libtest/unit/ContentRepositoryDependentTests.h | 2 +-
libminifi/test/libtest/unit/MockClasses.h | 8 +-
minifi-api/CMakeLists.txt | 4 +-
.../include/minifi-cpp/core/logging/Logger.h | 2 -
.../common/include/minifi-cpp/io/StreamCallback.h | 14 +-
minifi-api/include/minifi-api.h | 306 ++++++++++++++++++++
minifi-api/include/minifi-c/minifi-c.h | 314 ---------------------
.../minifi-cpp/core/ControllerServiceType.h | 7 +-
.../include/minifi-cpp/core/ProcessContext.h | 1 +
minifi-api/include/minifi-cpp/core/ProcessorApi.h | 1 -
.../minifi-cpp/core/logging/AdvancedLogger.h | 36 +--
minifi-api/minifi-api.def | 31 ++
minifi-api/minifi-c-api.def | 33 ---
109 files changed, 1173 insertions(+), 1198 deletions(-)
diff --git a/.clang-format b/.clang-format
index 013a12279..bae2470ae 100644
--- a/.clang-format
+++ b/.clang-format
@@ -5,10 +5,11 @@ AlignConsecutiveAssignments: None
AlignOperands: DontAlign
AllowAllParametersOfDeclarationOnNextLine: true
AllowAllConstructorInitializersOnNextLine: false
-AllowShortBlocksOnASingleLine: Always
-AllowShortCaseLabelsOnASingleLine: true
-AllowShortFunctionsOnASingleLine: Inline
-AllowShortIfStatementsOnASingleLine: Always
+AllowShortBlocksOnASingleLine: Empty
+AllowShortCaseLabelsOnASingleLine: false
+AllowShortEnumsOnASingleLine: false
+AllowShortFunctionsOnASingleLine: None
+AllowShortIfStatementsOnASingleLine: WithoutElse
AllowShortLambdasOnASingleLine: All
AllowShortLoopsOnASingleLine: false
AlwaysBreakAfterReturnType: None
diff --git a/.github/references/ubuntu_22_04_clang_arm_manifest.json b/.github/references/ubuntu_22_04_clang_arm_manifest.json
index 139f413bf..bcb495d3e 100644
--- a/.github/references/ubuntu_22_04_clang_arm_manifest.json
+++ b/.github/references/ubuntu_22_04_clang_arm_manifest.json
@@ -11089,13 +11089,13 @@
"type": "org.apache.nifi.minifi.api_testing.DogController",
"providedApiImplementations": [
{
- "artifact": "minifi-system",
- "group": "org.apache.nifi.minifi",
+ "artifact": "minifi-stable-api-testing",
+ "group": "org.apache.nifi.minifi.api_testing",
"type": "org.apache.nifi.minifi.api_testing.CanFlyControllerApi"
},
{
- "artifact": "minifi-system",
- "group": "org.apache.nifi.minifi",
+ "artifact": "minifi-stable-api-testing",
+ "group": "org.apache.nifi.minifi.api_testing",
"type": "org.apache.nifi.minifi.api_testing.NumberOfLegsControllerApi"
}
]
@@ -11107,13 +11107,13 @@
"type": "org.apache.nifi.minifi.api_testing.DuckController",
"providedApiImplementations": [
{
- "artifact": "minifi-system",
- "group": "org.apache.nifi.minifi",
+ "artifact": "minifi-stable-api-testing",
+ "group": "org.apache.nifi.minifi.api_testing",
"type": "org.apache.nifi.minifi.api_testing.CanFlyControllerApi"
},
{
- "artifact": "minifi-system",
- "group": "org.apache.nifi.minifi",
+ "artifact": "minifi-stable-api-testing",
+ "group": "org.apache.nifi.minifi.api_testing",
"type": "org.apache.nifi.minifi.api_testing.NumberOfLegsControllerApi"
}
]
diff --git a/.github/workflows/create-release-artifacts.yml b/.github/workflows/create-release-artifacts.yml
index 87d52da9d..53c91e34d 100644
--- a/.github/workflows/create-release-artifacts.yml
+++ b/.github/workflows/create-release-artifacts.yml
@@ -95,8 +95,8 @@ jobs:
mkdir minifi-native-sdk
cp behave_framework/dist/*.whl minifi-native-sdk/
- cp minifi-api/minifi-c-api.def minifi-native-sdk/
- cp minifi-api/include/minifi-c/minifi-c.h minifi-native-sdk/
+ cp minifi-api/minifi-api.def minifi-native-sdk/
+ cp minifi-api/include/minifi-api.h minifi-native-sdk/
zip -r minifi-native-sdk.zip minifi-native-sdk/
diff --git a/Extensions.md b/Extensions.md
index 0213d68f1..95ce88eee 100644
--- a/Extensions.md
+++ b/Extensions.md
@@ -20,36 +20,36 @@ To enable all extensions for your platform, you may use -DENABLE_ALL=TRUE OR sel
Extensions are dynamic libraries loaded at runtime by the agent.
## C extensions
-You can build shared libraries using the API defined in `minifi-c.h`
-For the shared library to be considered a valid extension, it must export a global symbol with the name `MinifiApiVersion`
-with its value equal to the uint32_t constant `MINIFI_API_VERSION` from `minifi-c.h`.
+You can build shared libraries using the API defined in `minifi-api.h`
+For the shared library to be considered a valid extension, it must export a global symbol with the name `minifi_api_version`
+with its value equal to the uint32_t constant `MINIFI_API_VERSION` from `minifi-api.h`.
### Resource Lifetime
-Unless otherwise specified, the following lifetime rules apply to all functions called by the agent (e.g., `MinifiInitExtension`, `MinifiProcessorCallbacks::onTrigger`, or other callbacks):
+Unless otherwise specified, the following lifetime rules apply to all functions called by the agent (e.g., `minifi_init_extension`, `minifi_processor_callbacks::trigger`, or other callbacks):
* Arguments: The lifetime of any resource provided as a function argument is limited to the duration of that function call.
-* Created Resources: The lifetime of resources created within these functions (e.g., a handle returned by `MinifiProcessSessionGet` inside `MinifiProcessorCallbacks::onTrigger`)
+* Created Resources: The lifetime of resources created within these functions (e.g., a handle returned by `minifi_process_session_get` inside `minifi_processor_callbacks::trigger`)
is limited to the scope of the innermost callback.
-(the return value of `MinifiRegisterExtension` is only valid during the execution of `MinifiInitExtension`).
+(the return value of `minifi_register_extension` is only valid during the execution of `minifi_init_extension`).
-Because of these scoping rules, all processor and controller service registrations must occur within the `MinifiInitExtension` call.
+Because of these scoping rules, all processor and controller service registrations must occur within the `minifi_init_extension` call.
One possible example of this is:
```C++
-extern "C" const uint32_t MinifiApiVersion = MINIFI_API_VERSION;
+extern "C" const uint32_t minifi_api_version = MINIFI_API_VERSION;
-extern "C" void MinifiInitExtension(MinifiExtensionContext* extension_context) {
- MinifiExtensionDefinition extension_definition{
+extern "C" void minifi_init_extension(minifi_extension_context* extension_context) {
+ minifi_extension_definition extension_definition{
.name = minifi::api::utils::toStringView(MAKESTRING(EXTENSION_NAME)),
.version = minifi::api::utils::toStringView(MAKESTRING(EXTENSION_VERSION)),
.deinit = nullptr,
.user_data = nullptr
};
- auto* extension = MinifiRegisterExtension(extension_context, &extension_definition);
- minifi::api::core::useProcessorClassDefinition<minifi::extensions::llamacpp::processors::RunLlamaCppInference>([&] (const MinifiProcessorClassDefinition& definition) {
- MinifiRegisterProcessor(extension, &definition);
+ auto* extension = minifi_register_extension(extension_context, &extension_definition);
+ minifi::api::core::useProcessorClassDefinition<minifi::extensions::llamacpp::processors::RunLlamaCppInference>([&] (const minifi_processor_definition& definition) {
+ minifi_register_processor(extension, &definition);
});
}
```
@@ -73,18 +73,18 @@ REGISTER_RESOURCE(RESTSender, DescriptionOnly);
```
Some extensions (e.g. `Python`) require initialization before use.
-You need to define an `MinifiInitCppExtension` function of type `MinifiExtension*(MinifiExtensionContext*)` to be called.
+You need to define an `minifi_init_cpp_extension` function with signature `void minifi_init_cpp_extension(minifi_extension_context*)` to be called.
```C++
-extern "C" void MinifiInitCppExtension(MinifiExtensionContext* extension_context) {
+extern "C" void minifi_init_cpp_extension(minifi_extension_context* extension_context) {
static PythonLibLoader python_lib_loader([&] (std::string_view key) -> std::optional<std::string> {
std::optional<std::string> result;
- MinifiConfigGet(extension_context, minifi::utils::toStringView(key), [] (void* user_data, MinifiStringView value) {
+ minifi_config_get(extension_context, minifi::utils::toStringView(key), [] (void* user_data, minifi_string_view value) {
*static_cast<std::optional<std::string>*>(user_data) = std::string{value.data, value.length};
}, &result);
return result;
});
- MinifiExtensionDefinition extension_definition{
+ minifi_extension_definition extension_definition{
.name = minifi::utils::toStringView(MAKESTRING(MODULE_NAME)),
.version = minifi::utils::toStringView(minifi::AgentBuild::VERSION),
.deinit = nullptr,
diff --git a/core-framework/include/core/ProcessorImpl.h b/core-framework/include/core/ProcessorImpl.h
index aa4c34006..b022d79fc 100644
--- a/core-framework/include/core/ProcessorImpl.h
+++ b/core-framework/include/core/ProcessorImpl.h
@@ -87,14 +87,6 @@ class ProcessorImpl : public virtual ProcessorApi {
std::string getProcessorType() const override = 0;
- void setTriggerWhenEmpty(bool trigger_when_empty) {
- trigger_when_empty_ = trigger_when_empty;
- }
-
- bool getTriggerWhenEmpty() const override {
- return trigger_when_empty_;
- }
-
void initialize(ProcessorDescriptor& self) final;
void setSupportedRelationships(std::span<const RelationshipDefinition> relationships);
@@ -140,8 +132,6 @@ class ProcessorImpl : public virtual ProcessorApi {
ProcessorMetadata metadata_;
- std::atomic<bool> trigger_when_empty_;
-
std::shared_ptr<ProcessorMetricsExtension> metrics_extension_;
std::shared_ptr<logging::Logger> logger_;
diff --git a/core-framework/src/core/ProcessorImpl.cpp b/core-framework/src/core/ProcessorImpl.cpp
index 31ec58e8b..7a7a6982e 100644
--- a/core-framework/src/core/ProcessorImpl.cpp
+++ b/core-framework/src/core/ProcessorImpl.cpp
@@ -44,7 +44,6 @@ namespace org::apache::nifi::minifi::core {
ProcessorImpl::ProcessorImpl(ProcessorMetadata metadata)
: metadata_(std::move(metadata)),
- trigger_when_empty_(false),
logger_(metadata_.logger) {
logger_->log_debug("Processor {} created with uuid {}", getName(), getUUIDStr());
}
diff --git a/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceContext.h b/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceContext.h
index 561bef0ac..3fbad0be4 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceContext.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceContext.h
@@ -17,17 +17,17 @@
#pragma once
-#include <string>
#include <expected>
+#include <string>
-#include "minifi-c.h"
+#include "minifi-api.h"
#include "minifi-cpp/core/PropertyDefinition.h"
namespace org::apache::nifi::minifi::api::core {
class ControllerServiceContext {
public:
- explicit ControllerServiceContext(MinifiControllerServiceContext* impl) : impl_(impl) {}
+ explicit ControllerServiceContext(minifi_controller_service_context* impl) : impl_(impl) {}
[[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 {
@@ -35,7 +35,7 @@ class ControllerServiceContext {
}
private:
- MinifiControllerServiceContext* impl_;
+ minifi_controller_service_context* 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
index b4b747384..6a163c8d5 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceImpl.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ControllerServiceImpl.h
@@ -43,7 +43,7 @@ class ControllerServiceImpl {
virtual ~ControllerServiceImpl();
- MinifiStatus enable(ControllerServiceContext&);
+ minifi_status enable(ControllerServiceContext&);
void disable();
[[nodiscard]] std::string getName() const;
@@ -51,7 +51,7 @@ class ControllerServiceImpl {
[[nodiscard]] minifi::utils::SmallString<36> getUUIDStr() const;
protected:
- virtual MinifiStatus enableImpl(api::core::ControllerServiceContext&) = 0;
+ virtual minifi_status enableImpl(api::core::ControllerServiceContext&) = 0;
virtual void disableImpl() {}
minifi::core::ControllerServiceMetadata metadata_;
diff --git a/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h b/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h
index da9e6e8e1..c2a5fb424 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h
@@ -21,12 +21,12 @@
#include <memory>
#include <stdexcept>
-#include "minifi-c.h"
+#include "minifi-api.h"
namespace org::apache::nifi::minifi::api::core {
struct EnsureMovedFromDeleter {
- void operator()(MinifiFlowFile* ff) {
+ void operator()(minifi_flow_file* ff) {
if (ff) {
if (std::uncaught_exceptions()) {
// there is already an exception in progress, do not terminate the process (although there are scenarios we could throw here)
@@ -37,6 +37,6 @@ struct EnsureMovedFromDeleter {
}
};
-using FlowFile = std::unique_ptr<MinifiFlowFile, EnsureMovedFromDeleter>;
+using FlowFile = std::unique_ptr<minifi_flow_file, EnsureMovedFromDeleter>;
} // namespace org::apache::nifi::minifi::api::core
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 26a55a44a..e3f821cfd 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
@@ -20,10 +20,11 @@
#include <expected>
#include <string>
+#include "PublishedMetrics.h"
#include "api/core/FlowFile.h"
#include "api/utils/Proxy.h"
#include "api/utils/Ssl.h"
-#include "minifi-c.h"
+#include "minifi-api.h"
#include "minifi-cpp/core/PropertyDefinition.h"
namespace org::apache::nifi::minifi::api::core {
@@ -40,32 +41,36 @@ class ProcessContext {
[[nodiscard]] virtual std::expected<std::string, std::error_code> getProperty(const minifi::core::PropertyReference& prop,
const FlowFile* ff) const = 0;
- [[nodiscard]] virtual std::expected<MinifiControllerService*, std::error_code> getControllerService(const minifi::core::PropertyReference& prop) const = 0;
- [[nodiscard]] virtual bool hasNonEmptyProperty(std::string_view name) const = 0;
+ [[nodiscard]] virtual std::expected<minifi_controller_service*, std::error_code> getControllerService(const minifi::core::PropertyReference& prop) const = 0;
[[nodiscard]] virtual std::map<std::string, std::string> getDynamicProperties(const FlowFile* flow_file) const = 0;
[[nodiscard]] virtual std::expected<std::optional<utils::net::SslData>, std::error_code> getSslData(const minifi::core::PropertyReference& prop) const = 0;
[[nodiscard]] virtual std::expected<std::optional<utils::ProxyData>, std::error_code> getProxyData(const minifi::core::PropertyReference& prop) const = 0;
+
+ [[nodiscard]] virtual std::expected<void, std::error_code> reportMetrics(const PublishedMetrics& metrics) const = 0;
+ [[nodiscard]] virtual std::expected<void, std::error_code> setTriggerWhenEmpty(bool) = 0;
};
class CffiProcessContext : public ProcessContext {
public:
- explicit CffiProcessContext(MinifiProcessContext* impl) : impl_(impl) {}
+ explicit CffiProcessContext(minifi_process_context* impl) : impl_(impl) {}
[[nodiscard]] std::expected<std::string, std::error_code> getProperty(const minifi::core::PropertyReference& property_reference,
const FlowFile* flow_file) const override;
- [[nodiscard]] std::expected<MinifiControllerService*, std::error_code> getControllerService(const minifi::core::PropertyReference& prop) const override;
+ [[nodiscard]] std::expected<minifi_controller_service*, std::error_code> getControllerService(const minifi::core::PropertyReference& prop) const override;
[[nodiscard]] std::map<std::string, std::string> getDynamicProperties(const FlowFile* flow_file) const override;
- [[nodiscard]] bool hasNonEmptyProperty(std::string_view name) const override;
[[nodiscard]] std::expected<std::optional<utils::net::SslData>, std::error_code> getSslData(const minifi::core::PropertyReference& prop) const override;
[[nodiscard]] std::expected<std::optional<utils::ProxyData>, std::error_code> getProxyData(const minifi::core::PropertyReference& prop) const override;
+ [[nodiscard]] std::expected<void, std::error_code> reportMetrics(const PublishedMetrics& metrics) const override;
+ [[nodiscard]] std::expected<void, std::error_code> setTriggerWhenEmpty(bool) override;
+
private:
[[nodiscard]] std::expected<std::string, std::error_code> getProperty(std::string_view name, const FlowFile* flow_file) const;
private:
- MinifiProcessContext* impl_;
+ minifi_process_context* impl_;
};
} // namespace org::apache::nifi::minifi::api::core
diff --git a/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h b/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h
index 1ef50770d..7212046db 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h
@@ -20,10 +20,10 @@
#include <map>
#include <span>
-#include "minifi-c.h"
+#include "FlowFile.h"
+#include "minifi-api.h"
#include "minifi-cpp/core/Relationship.h"
#include "minifi-cpp/io/StreamCallback.h"
-#include "FlowFile.h"
namespace org::apache::nifi::minifi::api::core {
@@ -61,7 +61,7 @@ class ProcessSession {
class CffiProcessSession : public ProcessSession {
public:
- explicit CffiProcessSession(MinifiProcessSession* impl): impl_(impl) {}
+ explicit CffiProcessSession(minifi_process_session* impl): impl_(impl) {}
FlowFile create(const FlowFile* parent = nullptr) override;
FlowFile get() override;
@@ -79,7 +79,7 @@ class CffiProcessSession : public ProcessSession {
[[nodiscard]] uint64_t getFlowFileSize(const FlowFile& ff) const override;
private:
- MinifiProcessSession* impl_;
+ minifi_process_session* impl_;
};
} // namespace org::apache::nifi::minifi::api::core
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 f5869715e..51a57c3e3 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessorImpl.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessorImpl.h
@@ -51,17 +51,9 @@ class ProcessorImpl {
virtual ~ProcessorImpl();
- void setTriggerWhenEmpty(const bool trigger_when_empty) {
- trigger_when_empty_ = trigger_when_empty;
- }
+ minifi_status onTrigger(ProcessContext&, ProcessSession&);
- virtual bool getTriggerWhenEmpty() const {
- return trigger_when_empty_;
- }
-
- MinifiStatus onTrigger(ProcessContext&, ProcessSession&);
-
- MinifiStatus onSchedule(ProcessContext&);
+ minifi_status onSchedule(ProcessContext&);
virtual void onUnSchedule() {}
@@ -73,17 +65,13 @@ class ProcessorImpl {
minifi::utils::Identifier getUUID() const;
minifi::utils::SmallString<36> getUUIDStr() const;
- virtual PublishedMetrics calculateMetrics() const {return {};}
-
protected:
- virtual MinifiStatus onTriggerImpl(ProcessContext&, ProcessSession&) {return MINIFI_STATUS_SUCCESS;}
+ virtual minifi_status onTriggerImpl(ProcessContext&, ProcessSession&) {return MINIFI_STATUS_SUCCESS;}
- virtual MinifiStatus onScheduleImpl(ProcessContext&) {return MINIFI_STATUS_SUCCESS;}
+ virtual minifi_status onScheduleImpl(ProcessContext&) {return MINIFI_STATUS_SUCCESS;}
minifi::core::ProcessorMetadata metadata_;
- std::atomic<bool> trigger_when_empty_;
-
std::shared_ptr<minifi::core::logging::Logger> logger_;
};
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 b121f0197..071a34e56 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/Resource.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/Resource.h
@@ -33,7 +33,7 @@
#include "api/utils/minifi-c-utils.h"
#include "core/ClassName.h"
#include "logging/Logger.h"
-#include "minifi-c.h"
+#include "minifi-api.h"
#include "minifi-cpp/core/ControllerServiceMetadata.h"
#include "minifi-cpp/core/ProcessorMetadata.h"
@@ -41,35 +41,35 @@ namespace org::apache::nifi::minifi::api::core {
template<typename Class, typename Fn>
void useProcessorClassDefinition(Fn&& fn) {
- std::vector<std::vector<MinifiStringView>> string_vector_cache;
+ std::vector<std::vector<minifi_string_view>> string_vector_cache;
const auto full_name = minifi::core::className<Class>();
- std::vector<MinifiPropertyDefinition> class_properties = utils::toProperties(Class::Properties, string_vector_cache);
- std::vector<MinifiDynamicPropertyDefinition> dynamic_properties;
+ std::vector<minifi_property_definition> class_properties = utils::toProperties(Class::Properties, string_vector_cache);
+ std::vector<minifi_dynamic_property_definition> dynamic_properties;
for (auto& prop : Class::DynamicProperties) {
- dynamic_properties.push_back(MinifiDynamicPropertyDefinition {
+ dynamic_properties.push_back(minifi_dynamic_property_definition {
.name = utils::minifiStringView(prop.name),
.value = utils::minifiStringView(prop.value),
.description = utils::minifiStringView(prop.description),
.supports_expression_language = prop.supports_expression_language
});
}
- std::vector<MinifiRelationshipDefinition> relationships;
+ std::vector<minifi_relationship_definition> relationships;
for (auto& rel : Class::Relationships) {
- relationships.push_back(MinifiRelationshipDefinition{
+ relationships.push_back(minifi_relationship_definition{
.name = utils::minifiStringView(rel.name),
.description = utils::minifiStringView(rel.description)
});
}
- std::vector<std::vector<MinifiStringView>> attribute_relationships_cache;
- std::vector<MinifiOutputAttributeDefinition> output_attributes;
+ std::vector<std::vector<minifi_string_view>> attribute_relationships_cache;
+ std::vector<minifi_output_attribute_definition> output_attributes;
for (auto& attr : Class::OutputAttributes) {
- std::vector<MinifiStringView> rel_cache;
+ std::vector<minifi_string_view> rel_cache;
for (auto& rel : attr.relationships) {
rel_cache.push_back(utils::minifiStringView(rel.name));
}
- output_attributes.push_back(MinifiOutputAttributeDefinition {
+ output_attributes.push_back(minifi_output_attribute_definition {
.name = utils::minifiStringView(attr.name),
.relationships_count = gsl::narrow<uint32_t>(attr.relationships.size()),
.relationships_ptr = rel_cache.data(),
@@ -78,15 +78,15 @@ void useProcessorClassDefinition(Fn&& fn) {
attribute_relationships_cache.push_back(std::move(rel_cache));
}
- MinifiProcessorClassDefinition definition{
+ minifi_processor_definition definition{
.full_name = utils::minifiStringView(full_name),
.description = utils::minifiStringView(Class::Description),
- .class_properties_count = gsl::narrow<uint32_t>(class_properties.size()),
- .class_properties_ptr = class_properties.data(),
+ .properties_count = gsl::narrow<uint32_t>(class_properties.size()),
+ .properties_ptr = class_properties.data(),
.dynamic_properties_count = gsl::narrow<uint32_t>(dynamic_properties.size()),
.dynamic_properties_ptr = dynamic_properties.data(),
- .class_relationships_count = gsl::narrow<uint32_t>(relationships.size()),
- .class_relationships_ptr = relationships.data(),
+ .relationships_count = gsl::narrow<uint32_t>(relationships.size()),
+ .relationships_ptr = relationships.data(),
.output_attributes_count = gsl::narrow<uint32_t>(output_attributes.size()),
.output_attributes_ptr = output_attributes.data(),
.supports_dynamic_properties = Class::SupportsDynamicProperties,
@@ -94,8 +94,8 @@ void useProcessorClassDefinition(Fn&& fn) {
.input_requirement = utils::toInputRequirement(Class::InputRequirement),
.is_single_threaded = Class::IsSingleThreaded,
- .callbacks = MinifiProcessorCallbacks{
- .create = [] (MinifiProcessorMetadata metadata) -> MINIFI_OWNED void* {
+ .callbacks = minifi_processor_callbacks{
+ .create = [] (minifi_processor_metadata metadata) -> MINIFI_OWNED void* {
try {
return new Class{minifi::core::ProcessorMetadata{
.uuid = minifi::utils::Identifier::parse(std::string{metadata.uuid.data, metadata.uuid.length}).value(),
@@ -106,10 +106,7 @@ void useProcessorClassDefinition(Fn&& fn) {
.destroy = [] (MINIFI_OWNED void* self) -> void {
delete static_cast<Class*>(self);
},
- .getTriggerWhenEmpty = [] (void* self) -> MinifiBool {
- return static_cast<Class*>(self)->getTriggerWhenEmpty();
- },
- .onTrigger = [] (void* self, MinifiProcessContext* context, MinifiProcessSession* session) -> MinifiStatus {
+ .trigger = [] (void* self, minifi_process_context* context, minifi_process_session* session) -> minifi_status {
CffiProcessContext context_wrapper(context);
CffiProcessSession session_wrapper(session);
try {
@@ -118,7 +115,7 @@ void useProcessorClassDefinition(Fn&& fn) {
return MINIFI_STATUS_UNKNOWN_ERROR;
}
},
- .onSchedule = [] (void* self, MinifiProcessContext* context) -> MinifiStatus {
+ .schedule = [] (void* self, minifi_process_context* context) -> minifi_status {
CffiProcessContext context_wrapper(context);
try {
return static_cast<Class*>(self)->onSchedule(context_wrapper);
@@ -126,20 +123,10 @@ void useProcessorClassDefinition(Fn&& fn) {
return MINIFI_STATUS_UNKNOWN_ERROR;
}
},
- .onUnSchedule = [] (void* self) -> void {
+ .unschedule = [] (void* self) -> void {
try {
static_cast<Class*>(self)->onUnSchedule();
} catch (...) {}
- },
- .calculateMetrics = [] (void* self) -> MINIFI_OWNED MinifiPublishedMetrics* {
- auto metrics = static_cast<Class*>(self)->calculateMetrics();
- std::vector<MinifiStringView> names;
- std::vector<double> values;
- for (auto& [name, val] : metrics) {
- names.push_back(utils::minifiStringView(name));
- values.push_back(val);
- }
- return MinifiPublishedMetricsCreate(gsl::narrow<uint32_t>(metrics.size()), names.data(), values.data());
}
}
};
@@ -149,12 +136,12 @@ void useProcessorClassDefinition(Fn&& fn) {
template<typename Class, typename Fn>
void useControllerServiceClassDefinition(Fn&& fn) {
- std::vector<std::vector<MinifiStringView>> string_vector_cache;
+ std::vector<std::vector<minifi_string_view>> string_vector_cache;
const auto full_name = minifi::core::className<Class>();
- std::vector<MinifiPropertyDefinition> class_properties = utils::toProperties(Class::Properties, string_vector_cache);
- std::vector<MinifiStringView> provided_interfaces;
+ std::vector<minifi_property_definition> class_properties = utils::toProperties(Class::Properties, string_vector_cache);
+ std::vector<minifi_string_view> provided_interfaces;
if constexpr (requires { Class::ProvidedInterfaces; }) {
provided_interfaces.reserve(Class::ProvidedInterfaces.size());
for (const auto& iface : Class::ProvidedInterfaces) {
@@ -162,16 +149,16 @@ void useControllerServiceClassDefinition(Fn&& fn) {
}
}
- MinifiControllerServiceClassDefinition definition{.full_name = utils::minifiStringView(full_name),
+ minifi_controller_service_definition definition{.full_name = utils::minifiStringView(full_name),
.description = utils::minifiStringView(Class::Description),
- .class_properties_count = gsl::narrow<uint32_t>(class_properties.size()),
- .class_properties_ptr = class_properties.data(),
+ .properties_count = gsl::narrow<uint32_t>(class_properties.size()),
+ .properties_ptr = class_properties.data(),
.provided_interfaces_count = gsl::narrow<uint32_t>(provided_interfaces.size()),
.provided_interfaces_ptr = provided_interfaces.data(),
- .callbacks = MinifiControllerServiceCallbacks{
- .create = [](MinifiControllerServiceMetadata metadata) -> MINIFI_OWNED void* {
+ .callbacks = minifi_controller_service_callbacks{
+ .create = [](minifi_controller_service_metadata 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(),
@@ -180,7 +167,7 @@ void useControllerServiceClassDefinition(Fn&& fn) {
} catch (...) { return nullptr; }
},
.destroy = [](MINIFI_OWNED void* self) -> void { delete static_cast<Class*>(self); },
- .enable = [](void* self, MinifiControllerServiceContext* context) -> MinifiStatus {
+ .enable = [](void* self, minifi_controller_service_context* context) -> minifi_status {
ControllerServiceContext context_wrapper(context);
try {
return static_cast<Class*>(self)->enable(context_wrapper);
@@ -191,7 +178,7 @@ void useControllerServiceClassDefinition(Fn&& fn) {
static_cast<Class*>(self)->disable();
} catch (...) {}
},
- .get_interface = [](void* self, MinifiStringView interface_name) -> void* {
+ .get_interface = [](void* self, minifi_string_view interface_name) -> void* {
try {
if constexpr (requires { Class::ProvidedInterfaces; }) {
const std::string_view name_view{interface_name.data, interface_name.length};
@@ -210,16 +197,16 @@ void useControllerServiceClassDefinition(Fn&& fn) {
}
template <typename... Processors>
-void registerProcessors(MinifiExtension* extension) {
- (core::useProcessorClassDefinition<Processors>([&](const MinifiProcessorClassDefinition& definition) {
- MinifiRegisterProcessor(extension, &definition);
+void registerProcessors(minifi_extension* extension) {
+ (core::useProcessorClassDefinition<Processors>([&](const minifi_processor_definition& definition) {
+ minifi_register_processor(extension, &definition);
}), ...);
}
template <typename... ControllerServices>
-void registerControllerServices(MinifiExtension* extension) {
- (core::useControllerServiceClassDefinition<ControllerServices>([&](const MinifiControllerServiceClassDefinition& definition) {
- MinifiRegisterControllerService(extension, &definition);
+void registerControllerServices(minifi_extension* extension) {
+ (core::useControllerServiceClassDefinition<ControllerServices>([&](const minifi_controller_service_definition& definition) {
+ minifi_register_controller_service(extension, &definition);
}), ...);
}
diff --git a/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h b/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
index 5a4401344..979824fb4 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
@@ -20,22 +20,20 @@
#include <iostream>
#include <string>
-#include "minifi-c.h"
+#include "minifi-api.h"
#include "minifi-cpp/core/logging/Logger.h"
namespace org::apache::nifi::minifi::api::core::logging {
class CffiLogger : public minifi::core::logging::Logger {
public:
- explicit CffiLogger(MinifiLogger* impl): impl_(impl) {}
+ explicit CffiLogger(minifi_logger* impl): impl_(impl) {}
- void set_max_log_size(int size) override;
void log_string(minifi::core::logging::LOG_LEVEL level, std::string str) override;
bool should_log(minifi::core::logging::LOG_LEVEL level) override;
- [[nodiscard]] minifi::core::logging::LOG_LEVEL level() const override;
private:
- MinifiLogger* impl_;
+ minifi_logger* impl_;
};
} // namespace org::apache::nifi::minifi::api::core::logging
diff --git a/extension-framework/cpp-extension-lib/include/api/utils/minifi-c-utils.h b/extension-framework/cpp-extension-lib/include/api/utils/minifi-c-utils.h
index 227ad4c80..cd8b06137 100644
--- a/extension-framework/cpp-extension-lib/include/api/utils/minifi-c-utils.h
+++ b/extension-framework/cpp-extension-lib/include/api/utils/minifi-c-utils.h
@@ -16,13 +16,14 @@
*/
#pragma once
-#include "minifi-c.h"
#include <string_view>
+
+#include "core/ClassName.h"
+#include "minifi-api.h"
#include "minifi-cpp/core/Annotation.h"
+#include "minifi-cpp/core/PropertyDefinition.h"
#include "minifi-cpp/utils/gsl.h"
-#include "core/ClassName.h"
#include "utils/StringUtils.h"
-#include "minifi-cpp/core/PropertyDefinition.h"
#ifdef WIN32
#define CEXTENSIONAPI extern "C" __declspec(dllexport)
@@ -32,15 +33,15 @@
namespace org::apache::nifi::minifi::api::utils {
-inline MinifiStringView minifiStringView(const std::string_view str) {
- return MinifiStringView{.data = str.data(), .length = str.length()};
+inline minifi_string_view minifiStringView(const std::string_view str) {
+ return minifi_string_view{.data = str.data(), .length = str.length()};
}
-inline std::string toString(const MinifiStringView sv) {
+inline std::string toString(const minifi_string_view sv) {
return {sv.data, sv.length};
}
-inline std::string_view toStringView(const MinifiStringView sv) {
+inline std::string_view toStringView(const minifi_string_view sv) {
return {sv.data, sv.length};
}
@@ -50,9 +51,9 @@ std::string classNameWithDots() {
return minifi::utils::string::replaceAll(class_name, "::", ".");
}
-MinifiValidator toStandardPropertyValidator(const minifi::core::PropertyValidator* validator);
+minifi_validator toStandardPropertyValidator(const minifi::core::PropertyValidator* validator);
-inline MinifiInputRequirement toInputRequirement(minifi::core::annotation::Input req) {
+inline minifi_input_requirement toInputRequirement(minifi::core::annotation::Input req) {
switch (req) {
case minifi::core::annotation::Input::INPUT_REQUIRED: return MINIFI_INPUT_REQUIRED;
case minifi::core::annotation::Input::INPUT_ALLOWED: return MINIFI_INPUT_ALLOWED;
@@ -61,10 +62,10 @@ inline MinifiInputRequirement toInputRequirement(minifi::core::annotation::Input
gsl_FailFast();
}
-inline std::vector<MinifiPropertyDefinition> toProperties(std::span<const minifi::core::PropertyReference> props, std::vector<std::vector<MinifiStringView>>& cache) {
- std::vector<MinifiPropertyDefinition> properties;
+inline std::vector<minifi_property_definition> toProperties(std::span<const minifi::core::PropertyReference> props, std::vector<std::vector<minifi_string_view>>& cache) {
+ std::vector<minifi_property_definition> properties;
for (auto& prop : props) {
- std::vector<MinifiStringView> sv_cache;
+ std::vector<minifi_string_view> sv_cache;
const size_t allowed_values_begin = sv_cache.size();
for (auto& allowed_value : prop.allowed_values) {
sv_cache.emplace_back(minifiStringView(allowed_value));
@@ -84,7 +85,7 @@ inline std::vector<MinifiPropertyDefinition> toProperties(std::span<const minifi
sv_cache.emplace_back(minifiStringView(*prop.default_value));
return sv_cache.size() - 1;
}();
- properties.push_back(MinifiPropertyDefinition{
+ properties.push_back(minifi_property_definition{
.name = minifiStringView(prop.name),
.display_name = minifiStringView(prop.display_name),
.description = minifiStringView(prop.description),
@@ -96,7 +97,7 @@ inline std::vector<MinifiPropertyDefinition> toProperties(std::span<const minifi
.allowed_values_ptr = sv_cache.data() + allowed_values_begin,
.validator = toStandardPropertyValidator(prop.validator),
- .type = allowed_types_begin ? sv_cache.data() + allowed_types_begin.value() : nullptr,
+ .allowed_type = allowed_types_begin ? sv_cache.data() + allowed_types_begin.value() : nullptr,
.supports_expression_language = prop.supports_expression_language
});
cache.emplace_back(std::move(sv_cache));
@@ -104,6 +105,6 @@ inline std::vector<MinifiPropertyDefinition> toProperties(std::span<const minifi
return properties;
}
-std::error_code make_error_code(MinifiStatus status);
+std::error_code make_error_code(minifi_status status);
} // 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 2defdff5c..9f9df63cf 100644
--- a/extension-framework/cpp-extension-lib/libtest/CProcessorTestUtils.h
+++ b/extension-framework/cpp-extension-lib/libtest/CProcessorTestUtils.h
@@ -32,7 +32,7 @@ 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::useProcessorClassDefinition<T>([&](const MinifiProcessorClassDefinition& definition) {
+ minifi::api::core::useProcessorClassDefinition<T>([&](const minifi_processor_definition& 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)...));
});
@@ -44,8 +44,8 @@ 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) {
+ minifi::api::core::useControllerServiceClassDefinition<T>([&](const minifi_controller_service_definition& definition) {
+ minifi::utils::useCControllerServiceClassDescription("test_bundle", 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)...));
diff --git a/extension-framework/cpp-extension-lib/mocklib/include/MockLogger.h b/extension-framework/cpp-extension-lib/mocklib/include/MockLogger.h
index 2c54cc663..0e19d93e5 100644
--- a/extension-framework/cpp-extension-lib/mocklib/include/MockLogger.h
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockLogger.h
@@ -25,10 +25,8 @@
namespace org::apache::nifi::minifi::mock {
class MockLogger : public core::logging::Logger {
public:
- void set_max_log_size(int) override {}
void log_string(const core::logging::LOG_LEVEL level, std::string s) override { logs_[level].emplace_back(std::move(s)); }
[[nodiscard]] bool should_log(const core::logging::LOG_LEVEL level) override { return level >= log_level_; }
- [[nodiscard]] core::logging::LOG_LEVEL level() const override { return log_level_; }
core::logging::LOG_LEVEL log_level_ = core::logging::LOG_LEVEL::trace;
std::map<core::logging::LOG_LEVEL, std::vector<std::string>> logs_;
diff --git a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
index cddcc9761..a239eed67 100644
--- a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
@@ -29,14 +29,23 @@ class MockProcessContext : public api::core::ProcessContext {
[[nodiscard]] std::expected<std::string, std::error_code> getProperty(const core::PropertyReference& property_reference,
const api::core::FlowFile* flow_file) const override;
- [[nodiscard]] std::expected<MinifiControllerService*, std::error_code> getControllerService(const minifi::core::PropertyReference& prop) const override;
+ [[nodiscard]] std::expected<minifi_controller_service*, std::error_code> getControllerService(const minifi::core::PropertyReference& prop) const override;
[[nodiscard]] std::map<std::string, std::string> getDynamicProperties(const api::core::FlowFile* flow_file) const override;
- [[nodiscard]] bool hasNonEmptyProperty(std::string_view name) const override;
[[nodiscard]] std::expected<std::optional<api::utils::net::SslData>, std::error_code> getSslData(const minifi::core::PropertyReference& prop) const override;
[[nodiscard]] std::expected<std::optional<api::utils::ProxyData>, std::error_code> getProxyData(const minifi::core::PropertyReference& prop) const override;
+ [[nodiscard]] std::expected<void, std::error_code> reportMetrics(const api::core::PublishedMetrics& metrics) const override {
+ *last_published_metrics_ = metrics;
+ return {};
+ }
+
+ [[nodiscard]] std::expected<void, std::error_code> setTriggerWhenEmpty(bool) override {
+ return {};
+ }
+
std::map<std::string, std::string, std::less<>> properties_;
+ std::unique_ptr<api::core::PublishedMetrics> last_published_metrics_ = std::make_unique<api::core::PublishedMetrics>();
private:
[[nodiscard]] std::expected<std::string, std::error_code> getProperty(std::string_view name,
diff --git a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h
index 51aeeb020..ad3aeb109 100644
--- a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h
@@ -24,20 +24,20 @@
#include "range/v3/range/conversion.hpp"
#include "range/v3/view/transform.hpp"
-struct MinifiFlowFile {
- explicit MinifiFlowFile(std::string content_str) {
+struct minifi_flow_file {
+ explicit minifi_flow_file(std::string content_str) {
this->content = content_str
| ranges::views::transform([](char c) { return static_cast<std::byte>(c); })
| ranges::to<std::vector<std::byte>>();
}
- ~MinifiFlowFile() = default;
+ ~minifi_flow_file() = default;
- MinifiFlowFile() = default;
- MinifiFlowFile(const MinifiFlowFile&) = default;
- MinifiFlowFile(MinifiFlowFile&&) = default;
- MinifiFlowFile& operator=(const MinifiFlowFile&) = default;
- MinifiFlowFile& operator=(MinifiFlowFile&&) = default;
+ minifi_flow_file() = default;
+ minifi_flow_file(const minifi_flow_file&) = default;
+ minifi_flow_file(minifi_flow_file&&) = default;
+ minifi_flow_file& operator=(const minifi_flow_file&) = default;
+ minifi_flow_file& operator=(minifi_flow_file&&) = default;
std::map<std::string, std::string> attributes;
std::vector<std::byte> content;
@@ -69,13 +69,13 @@ class MockProcessSession : public api::core::ProcessSession {
[[nodiscard]] std::string getFlowFileId(const api::core::FlowFile& ff) const override;
[[nodiscard]] uint64_t getFlowFileSize(const api::core::FlowFile& ff) const override;
- void addInputFlowFile(std::unique_ptr<MinifiFlowFile> flow_file);
+ void addInputFlowFile(std::unique_ptr<minifi_flow_file> flow_file);
private:
- std::vector<std::unique_ptr<MinifiFlowFile>> input_flow_files_;
+ std::vector<std::unique_ptr<minifi_flow_file>> input_flow_files_;
using RelationshipName = std::string;
- std::map<RelationshipName, std::vector<std::unique_ptr<MinifiFlowFile>>> transferred_flow_files_;
- std::vector<std::unique_ptr<MinifiFlowFile>> removed_flow_files_;
+ std::map<RelationshipName, std::vector<std::unique_ptr<minifi_flow_file>>> transferred_flow_files_;
+ std::vector<std::unique_ptr<minifi_flow_file>> removed_flow_files_;
};
} // namespace org::apache::nifi::minifi::mock
diff --git a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp
index 2b571afcf..4469a6b64 100644
--- a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp
+++ b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp
@@ -32,7 +32,7 @@ std::expected<std::string, std::error_code> MockProcessContext::getProperty(cons
return std::unexpected{make_error_code(core::PropertyErrorCode::PropertyNotSet)};
}
-std::expected<MinifiControllerService*, std::error_code> MockProcessContext::getControllerService(const minifi::core::PropertyReference&) const {
+std::expected<minifi_controller_service*, std::error_code> MockProcessContext::getControllerService(const minifi::core::PropertyReference&) const {
return nullptr;
}
@@ -40,10 +40,6 @@ std::map<std::string, std::string> MockProcessContext::getDynamicProperties(cons
return {};
}
-bool MockProcessContext::hasNonEmptyProperty(const std::string_view name) const {
- return properties_.contains(name);
-}
-
std::expected<std::optional<api::utils::net::SslData>, std::error_code> MockProcessContext::getSslData(const minifi::core::PropertyReference&) const {
return std::nullopt;
}
diff --git a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp
index c3edc0d66..6af5178e5 100644
--- a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp
+++ b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp
@@ -22,7 +22,7 @@
namespace org::apache::nifi::minifi::mock {
api::core::FlowFile MockProcessSession::create(const api::core::FlowFile*) {
- return api::core::FlowFile{new MinifiFlowFile};
+ return api::core::FlowFile{new minifi_flow_file};
}
api::core::FlowFile MockProcessSession::get() {
if (input_flow_files_.empty()) { return nullptr; }
@@ -38,11 +38,11 @@ void MockProcessSession::penalize(api::core::FlowFile& ff) {
}
void MockProcessSession::transfer(api::core::FlowFile ff, const minifi::core::Relationship& relationship) {
- transferred_flow_files_[relationship.getName()].push_back(std::unique_ptr<MinifiFlowFile>(ff.release()));
+ transferred_flow_files_[relationship.getName()].push_back(std::unique_ptr<minifi_flow_file>(ff.release()));
}
void MockProcessSession::remove(api::core::FlowFile ff) {
- removed_flow_files_.push_back(std::unique_ptr<MinifiFlowFile>(ff.release()));
+ removed_flow_files_.push_back(std::unique_ptr<minifi_flow_file>(ff.release()));
}
void MockProcessSession::write(api::core::FlowFile& ff, const io::OutputStreamCallback& callback) {
@@ -81,7 +81,7 @@ uint64_t MockProcessSession::getFlowFileSize(const api::core::FlowFile& ff) cons
return ff->content.size();
}
-void MockProcessSession::addInputFlowFile(std::unique_ptr<MinifiFlowFile> flow_file) {
+void MockProcessSession::addInputFlowFile(std::unique_ptr<minifi_flow_file> flow_file) {
input_flow_files_.push_back(std::move(flow_file));
}
} // namespace org::apache::nifi::minifi::mock
diff --git a/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp
new file mode 100644
index 000000000..5647bfa11
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp
@@ -0,0 +1,149 @@
+/**
+ * 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 <stdexcept>
+
+#include "minifi-api.h"
+
+extern "C" {
+minifi_extension* MINIFI_REGISTER_EXTENSION_FN(minifi_extension_context*, const minifi_extension_definition*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_register_processor(minifi_extension*, const minifi_processor_definition*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_register_controller_service(minifi_extension*, const minifi_controller_service_definition*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_context_get_property(minifi_process_context*, minifi_string_view, minifi_flow_file*,
+ void (*)(void* user_ctx, minifi_string_view property_value), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_context_get_controller_service_from_property(minifi_process_context*, minifi_string_view, minifi_string_view,
+ minifi_controller_service**) {
+ throw std::runtime_error("Not implemented");
+}
+
+void minifi_process_context_get_dynamic_properties(minifi_process_context*, minifi_flow_file*,
+ void (*)(void* user_ctx, minifi_string_view dynamic_property_name, minifi_string_view dynamic_property_value), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+void minifi_logger_log_string(minifi_logger*, minifi_log_level, minifi_string_view) {
+ throw std::runtime_error("Not implemented");
+}
+
+bool minifi_logger_should_log(minifi_logger*, minifi_log_level) {
+ throw std::runtime_error("Not implemented");
+}
+
+MINIFI_OWNED minifi_flow_file* minifi_process_session_get(minifi_process_session*) {
+ throw std::runtime_error("Not implemented");
+}
+
+MINIFI_OWNED minifi_flow_file* minifi_process_session_create(minifi_process_session*, minifi_flow_file*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_session_penalize(minifi_process_session*, minifi_flow_file*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_session_transfer(minifi_process_session*, MINIFI_OWNED minifi_flow_file*, minifi_string_view) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_session_remove(minifi_process_session*, MINIFI_OWNED minifi_flow_file*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_session_read(minifi_process_session*, minifi_flow_file*, int64_t (*)(void* user_ctx, minifi_input_stream*), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_session_write(minifi_process_session*, minifi_flow_file*, int64_t (*)(void* user_ctx, minifi_output_stream*), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+void minifi_config_get(minifi_extension_context*, minifi_string_view, void (*)(void* user_ctx, minifi_string_view config_value), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+size_t minifi_input_stream_size(minifi_input_stream*) {
+ throw std::runtime_error("Not implemented");
+}
+
+int64_t minifi_input_stream_read(minifi_input_stream*, char*, size_t) {
+ throw std::runtime_error("Not implemented");
+}
+
+int64_t minifi_output_stream_write(minifi_output_stream*, const char*, size_t) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_session_set_flow_file_attribute(minifi_process_session*, minifi_flow_file*, minifi_string_view,
+ const minifi_string_view*) {
+ throw std::runtime_error("Not implemented");
+}
+
+bool minifi_process_session_get_flow_file_attribute(minifi_process_session*, minifi_flow_file*, minifi_string_view,
+ void (*)(void* user_ctx, minifi_string_view attribute_value), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+void minifi_process_session_get_flow_file_attributes(minifi_process_session*, minifi_flow_file*,
+ void (*)(void* user_ctx, minifi_string_view attribute_name, minifi_string_view attribute_value), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+uint64_t minifi_process_session_get_flow_file_size(minifi_process_session*, minifi_flow_file*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_session_get_flow_file_id(minifi_process_session*, minifi_flow_file*,
+ void (*)(void* user_ctx, minifi_string_view flow_file_id), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+enum minifi_status minifi_controller_service_context_get_property(struct minifi_controller_service_context*, minifi_string_view,
+ void (*)(void* user_ctx, minifi_string_view property_value), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+enum minifi_status minifi_process_context_get_ssl_data_from_property(struct minifi_process_context*, struct minifi_string_view,
+ void (*)(void*, const struct minifi_ssl_data*), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+enum minifi_status minifi_process_context_get_proxy_data_from_property(struct minifi_process_context*, struct minifi_string_view,
+ void (*)(void*, const struct minifi_proxy_data*), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_context_set_trigger_when_empty(minifi_process_context*, bool) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_process_context_report_metrics(minifi_process_context*, size_t, const minifi_string_view*, const double*) {
+ throw std::runtime_error("Not implemented");
+}
+
+} // extern "C"
diff --git a/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-c.cpp b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-c.cpp
deleted file mode 100644
index 873b6527a..000000000
--- a/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-c.cpp
+++ /dev/null
@@ -1,142 +0,0 @@
-/**
- * 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 <stdexcept>
-
-#include "minifi-c.h"
-
-extern "C" {
-MinifiExtension* MINIFI_REGISTER_EXTENSION_FN(MinifiExtensionContext*, const MinifiExtensionDefinition*) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiRegisterProcessor(MinifiExtension*, const MinifiProcessorClassDefinition*) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiRegisterControllerService(MinifiExtension*, const MinifiControllerServiceClassDefinition*) {
- throw std::runtime_error("Not implemented");
-}
-
-MINIFI_OWNED MinifiPublishedMetrics* MinifiPublishedMetricsCreate(size_t, const MinifiStringView*, const double*) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiProcessContextGetProperty(MinifiProcessContext*, MinifiStringView, MinifiFlowFile*,
- void (*)(void* user_ctx, MinifiStringView property_value), void*) {
- throw std::runtime_error("Not implemented");
-}
-MinifiBool MinifiProcessContextHasNonEmptyProperty(MinifiProcessContext*, MinifiStringView) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiProcessContextGetControllerServiceFromProperty(MinifiProcessContext*, MinifiStringView, MinifiStringView, MinifiControllerService**) {
- throw std::runtime_error("Not implemented");
-}
-void MinifiProcessContextGetDynamicProperties(MinifiProcessContext*, MinifiFlowFile*,
- void (*)(void* user_ctx, MinifiStringView dynamic_property_name, MinifiStringView dynamic_property_value), void*) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiProcessContextGetSslDataFromProperty(MinifiProcessContext*, MinifiStringView,
- void (*)(void* user_ctx, const MinifiSslData* ssl_data), void*) {
- throw std::runtime_error("Not implemented");
-}
-
-void MinifiLoggerSetMaxLogSize(MinifiLogger*, int32_t) {
- throw std::runtime_error("Not implemented");
-}
-void MinifiLoggerLogString(MinifiLogger*, MinifiLogLevel, MinifiStringView) {
- throw std::runtime_error("Not implemented");
-}
-MinifiBool MinifiLoggerShouldLog(MinifiLogger*, MinifiLogLevel) {
- throw std::runtime_error("Not implemented");
-}
-MinifiLogLevel MinifiLoggerLevel(MinifiLogger*) {
- throw std::runtime_error("Not implemented");
-}
-
-MINIFI_OWNED MinifiFlowFile* MinifiProcessSessionGet(MinifiProcessSession*) {
- throw std::runtime_error("Not implemented");
-}
-MINIFI_OWNED MinifiFlowFile* MinifiProcessSessionCreate(MinifiProcessSession*, MinifiFlowFile*) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiProcessSessionPenalize(MinifiProcessSession*, MinifiFlowFile*) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiProcessSessionTransfer(MinifiProcessSession*, MINIFI_OWNED MinifiFlowFile*, MinifiStringView) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiProcessSessionRemove(MinifiProcessSession*, MINIFI_OWNED MinifiFlowFile*) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiProcessSessionRead(MinifiProcessSession*, MinifiFlowFile*, int64_t (*)(void* user_ctx, MinifiInputStream*), void*) {
- throw std::runtime_error("Not implemented");
-}
-MinifiStatus MinifiProcessSessionWrite(MinifiProcessSession*, MinifiFlowFile*, int64_t (*)(void* user_ctx, MinifiOutputStream*), void*) {
- throw std::runtime_error("Not implemented");
-}
-
-void MinifiConfigGet(MinifiExtensionContext*, MinifiStringView, void (*)(void* user_ctx, MinifiStringView config_value), void*) {
- throw std::runtime_error("Not implemented");
-}
-
-size_t MinifiInputStreamSize(MinifiInputStream*) {
- throw std::runtime_error("Not implemented");
-}
-
-int64_t MinifiInputStreamRead(MinifiInputStream*, char*, size_t) {
- throw std::runtime_error("Not implemented");
-}
-int64_t MinifiOutputStreamWrite(MinifiOutputStream*, const char*, size_t) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiProcessSessionSetFlowFileAttribute(MinifiProcessSession*, MinifiFlowFile*, MinifiStringView, const MinifiStringView*) {
- throw std::runtime_error("Not implemented");
-}
-MinifiBool MinifiProcessSessionGetFlowFileAttribute(MinifiProcessSession*, MinifiFlowFile*, MinifiStringView,
- void (*)(void* user_ctx, MinifiStringView attribute_value), void*) {
- throw std::runtime_error("Not implemented");
-}
-void MinifiProcessSessionGetFlowFileAttributes(MinifiProcessSession*, MinifiFlowFile*,
- void (*)(void* user_ctx, MinifiStringView attribute_name, MinifiStringView attribute_value), void*) {
- throw std::runtime_error("Not implemented");
-}
-uint64_t MinifiProcessSessionGetFlowFileSize(MinifiProcessSession*, MinifiFlowFile*) {
- throw std::runtime_error("Not implemented");
-}
-MinifiStatus MinifiProcessSessionGetFlowFileId(MinifiProcessSession*, MinifiFlowFile*, void (*)(void* user_ctx, MinifiStringView flow_file_id),
- void*) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiControllerServiceContextGetProperty(MinifiControllerServiceContext*, MinifiStringView,
- void (*)(void* user_ctx, MinifiStringView property_value), void*) {
- throw std::runtime_error("Not implemented");
-}
-
-MinifiStatus MinifiProcessContextGetProxyDataFromProperty(MinifiProcessContext*, MinifiStringView,
- void (*)(void* user_ctx, const MinifiProxyData* proxy_data), void*) {
- throw std::runtime_error("Not implemented");
-}
-} // extern "C"
diff --git a/extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp b/extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp
index be5104c3b..e0e8b946b 100644
--- a/extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp
@@ -22,8 +22,8 @@ namespace org::apache::nifi::minifi::api::core {
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::minifiStringView(name),
- [] (void* data, const MinifiStringView result) {
+ const minifi_status status = minifi_controller_service_context_get_property(impl_, utils::minifiStringView(name),
+ [] (void* data, const minifi_string_view result) {
(*static_cast<std::optional<std::string>*>(data)) = std::string(result.data, result.length);
}, &value);
diff --git a/extension-framework/cpp-extension-lib/src/core/ControllerServiceImpl.cpp b/extension-framework/cpp-extension-lib/src/core/ControllerServiceImpl.cpp
index 920dabf97..45ecc63ed 100644
--- a/extension-framework/cpp-extension-lib/src/core/ControllerServiceImpl.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ControllerServiceImpl.cpp
@@ -32,7 +32,7 @@ ControllerServiceImpl::~ControllerServiceImpl() {
logger_->log_debug("Destroying controller service {} with uuid {}", getName(), getUUIDStr());
}
-MinifiStatus ControllerServiceImpl::enable(ControllerServiceContext& ctx) {
+minifi_status ControllerServiceImpl::enable(ControllerServiceContext& ctx) {
try {
return enableImpl(ctx);
} catch (const std::exception& e) {
diff --git a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
index e2f3a65c2..361dde063 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
@@ -29,25 +29,21 @@ std::expected<std::string, std::error_code> CffiProcessContext::getProperty(cons
std::expected<std::string, std::error_code> CffiProcessContext::getProperty(std::string_view name, const FlowFile* flow_file) const {
std::optional<std::string> value;
- const MinifiStatus status = MinifiProcessContextGetProperty(
+ const minifi_status status = minifi_process_context_get_property(
impl_,
utils::minifiStringView(name),
- flow_file ? flow_file->get() : MINIFI_NULL,
- [](void* data, const MinifiStringView result) { (*static_cast<std::optional<std::string>*>(data)) = std::string(result.data, result.length); },
+ flow_file ? flow_file->get() : nullptr,
+ [](void* data, const minifi_string_view result) { (*static_cast<std::optional<std::string>*>(data)) = std::string(result.data, result.length); },
&value);
if (!value) { return std::unexpected{utils::make_error_code(status)}; }
return value.value();
}
-bool CffiProcessContext::hasNonEmptyProperty(std::string_view name) const {
- return MinifiProcessContextHasNonEmptyProperty(impl_, utils::minifiStringView(name));
-}
-
-std::expected<MinifiControllerService*, std::error_code> CffiProcessContext::getControllerService(const minifi::core::PropertyReference& prop) const {
- MinifiControllerService* controller_service = nullptr;
+std::expected<minifi_controller_service*, std::error_code> CffiProcessContext::getControllerService(const minifi::core::PropertyReference& prop) const {
+ minifi_controller_service* controller_service = nullptr;
gsl_Assert(prop.allowed_types.size() == 1);
- const MinifiStatus status = MinifiProcessContextGetControllerServiceFromProperty(impl_,
+ const minifi_status status = minifi_process_context_get_controller_service_from_property(impl_,
utils::minifiStringView(prop.name),
utils::minifiStringView(prop.allowed_types[0]),
&controller_service);
@@ -62,10 +58,10 @@ std::expected<MinifiControllerService*, std::error_code> CffiProcessContext::get
std::map<std::string, std::string> CffiProcessContext::getDynamicProperties(const FlowFile* flow_file) const {
std::map<std::string, std::string> result;
- MinifiProcessContextGetDynamicProperties(
+ minifi_process_context_get_dynamic_properties(
impl_,
- flow_file ? flow_file->get() : MINIFI_NULL,
- [](void* user_ctx, const MinifiStringView key, const MinifiStringView value) {
+ flow_file ? flow_file->get() : nullptr,
+ [](void* user_ctx, const minifi_string_view key, const minifi_string_view value) {
static_cast<std::map<std::string, std::string>*>(user_ctx)->emplace(utils::toString(key), utils::toString(value));
},
&result);
@@ -73,19 +69,23 @@ std::map<std::string, std::string> CffiProcessContext::getDynamicProperties(cons
}
std::expected<std::optional<utils::net::SslData>, std::error_code> CffiProcessContext::getSslData(const minifi::core::PropertyReference& prop) const {
- const auto controller_name = getProperty(prop, nullptr);
- if (!controller_name) { return std::nullopt; }
-
auto ssl_data = utils::net::SslData{};
- if (const auto status = MinifiProcessContextGetSslDataFromProperty(impl_, utils::minifiStringView(prop.name), [](void* data, const MinifiSslData* minifi_ssl_data) {
- auto* my_ssl_data = static_cast<utils::net::SslData*>(data);
- my_ssl_data->ca_loc = utils::toString(minifi_ssl_data->ca_certificate_file);
- my_ssl_data->cert_loc = utils::toString(minifi_ssl_data->certificate_file);
- my_ssl_data->key_loc = utils::toString(minifi_ssl_data->private_key_file);
- my_ssl_data->key_pw = utils::toString(minifi_ssl_data->passphrase);
- }, &ssl_data);
+ if (const auto status = minifi_process_context_get_ssl_data_from_property(
+ impl_,
+ utils::minifiStringView(prop.name),
+ [](void* data, const minifi_ssl_data* minifi_ssl_data) {
+ auto* my_ssl_data = static_cast<utils::net::SslData*>(data);
+ my_ssl_data->ca_loc = utils::toString(minifi_ssl_data->ca_certificate_file);
+ my_ssl_data->cert_loc = utils::toString(minifi_ssl_data->certificate_file);
+ my_ssl_data->key_loc = utils::toString(minifi_ssl_data->private_key_file);
+ my_ssl_data->key_pw = utils::toString(minifi_ssl_data->passphrase);
+ },
+ &ssl_data);
status != MINIFI_STATUS_SUCCESS) {
+ if (status == MINIFI_STATUS_PROPERTY_NOT_SET) {
+ return std::nullopt;
+ }
return std::unexpected{utils::make_error_code(status)};
}
@@ -94,10 +94,10 @@ std::expected<std::optional<utils::net::SslData>, std::error_code> CffiProcessCo
std::expected<std::optional<utils::ProxyData>, std::error_code> CffiProcessContext::getProxyData(const minifi::core::PropertyReference& prop) const {
auto proxy_data = utils::ProxyData{};
- const auto status = MinifiProcessContextGetProxyDataFromProperty(
+ const auto status = minifi_process_context_get_proxy_data_from_property(
impl_,
utils::minifiStringView(prop.name),
- [](void* data, const MinifiProxyData* minifi_proxy_data) {
+ [](void* data, const minifi_proxy_data* minifi_proxy_data) {
auto* proxy = static_cast<utils::ProxyData*>(data);
proxy->host = utils::toString(minifi_proxy_data->hostname);
proxy->port = minifi_proxy_data->port;
@@ -119,4 +119,33 @@ std::expected<std::optional<utils::ProxyData>, std::error_code> CffiProcessConte
return std::unexpected{utils::make_error_code(status)};
}
+std::expected<void, std::error_code> CffiProcessContext::reportMetrics(const PublishedMetrics& metrics) const {
+ const auto metrics_size = metrics.size();
+ std::vector<minifi_string_view> keys;
+ std::vector<double> values;
+
+ keys.reserve(metrics_size);
+ values.reserve(metrics_size);
+
+ for (const auto& [name, value] : metrics) {
+ keys.push_back(utils::minifiStringView(name));
+ values.push_back(value);
+ }
+
+ const auto status = minifi_process_context_report_metrics(impl_, metrics_size, keys.data(), values.data());
+ if (status != MINIFI_STATUS_SUCCESS) {
+ return std::unexpected{utils::make_error_code(status)};
+ }
+ return {};
+}
+
+std::expected<void, std::error_code> CffiProcessContext::setTriggerWhenEmpty(bool trigger_when_empty) {
+ const auto status = minifi_process_context_set_trigger_when_empty(impl_, trigger_when_empty);
+ if (status != MINIFI_STATUS_SUCCESS) {
+ return std::unexpected{utils::make_error_code(status)};
+ }
+ return {};
+}
+
+
} // namespace org::apache::nifi::minifi::api::core
diff --git a/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp b/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp
index 5bf12b1d6..41b8be081 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp
@@ -29,10 +29,10 @@ namespace {
class MinifiOutputStreamWrapper : public io::OutputStreamImpl {
public:
- explicit MinifiOutputStreamWrapper(MinifiOutputStream* impl): impl_(impl) {}
+ explicit MinifiOutputStreamWrapper(minifi_output_stream* impl): impl_(impl) {}
size_t write(const uint8_t *value, size_t len) override {
- return MinifiOutputStreamWrite(impl_, reinterpret_cast<const char*>(value), len);
+ return minifi_output_stream_write(impl_, reinterpret_cast<const char*>(value), len);
}
void close() override {gsl_FailFast();}
@@ -42,19 +42,19 @@ class MinifiOutputStreamWrapper : public io::OutputStreamImpl {
[[nodiscard]] std::span<const std::byte> getBuffer() const override {gsl_FailFast();}
private:
- MinifiOutputStream* impl_;
+ minifi_output_stream* impl_;
};
class MinifiInputStreamWrapper : public io::InputStreamImpl {
public:
- explicit MinifiInputStreamWrapper(MinifiInputStream* impl): impl_(impl) {}
+ explicit MinifiInputStreamWrapper(minifi_input_stream* impl): impl_(impl) {}
size_t read(std::span<std::byte> out_buffer) override {
- return MinifiInputStreamRead(impl_, reinterpret_cast<char*>(out_buffer.data()), out_buffer.size());
+ return minifi_input_stream_read(impl_, reinterpret_cast<char*>(out_buffer.data()), out_buffer.size());
}
[[nodiscard]] size_t size() const override {
- return MinifiInputStreamSize(impl_);
+ return minifi_input_stream_size(impl_);
}
void close() override {gsl_FailFast();}
@@ -64,43 +64,43 @@ class MinifiInputStreamWrapper : public io::InputStreamImpl {
[[nodiscard]] std::span<const std::byte> getBuffer() const override {gsl_FailFast();}
private:
- MinifiInputStream* impl_;
+ minifi_input_stream* impl_;
};
} // namespace
FlowFile CffiProcessSession::get() {
- return FlowFile{MinifiProcessSessionGet(impl_)};
+ return FlowFile{minifi_process_session_get(impl_)};
}
FlowFile CffiProcessSession::create(const FlowFile* parent) {
- return FlowFile{MinifiProcessSessionCreate(impl_, parent ? parent->get() : MINIFI_NULL)};
+ return FlowFile{minifi_process_session_create(impl_, parent ? parent->get() : nullptr)};
}
void CffiProcessSession::penalize(FlowFile& ff) {
- if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionPenalize(impl_, ff.get())) {
+ if (MINIFI_STATUS_SUCCESS != minifi_process_session_penalize(impl_, ff.get())) {
throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to penalize flowfile");
}
}
void CffiProcessSession::transfer(FlowFile ff, const minifi::core::Relationship& relationship) {
const auto rel_name = relationship.getName();
- if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionTransfer(impl_, ff.release(), utils::minifiStringView(rel_name))) {
+ if (MINIFI_STATUS_SUCCESS != minifi_process_session_transfer(impl_, ff.release(), utils::minifiStringView(rel_name))) {
throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to transfer flowfile");
}
}
void CffiProcessSession::remove(FlowFile ff) {
- if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionRemove(impl_, ff.release())) {
+ if (MINIFI_STATUS_SUCCESS != minifi_process_session_remove(impl_, ff.release())) {
throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to remove flowfile");
}
}
void CffiProcessSession::write(FlowFile& flow_file, const io::OutputStreamCallback& callback) {
- const auto status = MinifiProcessSessionWrite(
+ const auto status = minifi_process_session_write(
impl_,
flow_file.get(),
- [](void* data, MinifiOutputStream* output) -> int64_t {
+ [](void* data, minifi_output_stream* output) -> int64_t {
const auto result =
(*static_cast<const io::OutputStreamCallback*>(data))(std::make_shared<MinifiOutputStreamWrapper>(output));
return result.toI64();
@@ -110,10 +110,10 @@ void CffiProcessSession::write(FlowFile& flow_file, const io::OutputStreamCallba
}
void CffiProcessSession::read(FlowFile& flow_file, const io::InputStreamCallback& callback) {
- const auto status = MinifiProcessSessionRead(
+ const auto status = minifi_process_session_read(
impl_,
flow_file.get(),
- [](void* data, MinifiInputStream* input) -> int64_t {
+ [](void* data, minifi_input_stream* input) -> int64_t {
const auto result =
(*static_cast<const io::InputStreamCallback*>(data))(std::make_shared<MinifiInputStreamWrapper>(input));
return result.toI64();
@@ -123,21 +123,21 @@ void CffiProcessSession::read(FlowFile& flow_file, const io::InputStreamCallback
}
void CffiProcessSession::setAttribute(FlowFile& ff, const std::string_view key, std::string value) { // NOLINT(performance-unnecessary-value-param)
- const MinifiStringView value_ref = utils::minifiStringView(value);
- if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionSetFlowFileAttribute(impl_, ff.get(), utils::minifiStringView(key), &value_ref)) {
+ const minifi_string_view value_ref = utils::minifiStringView(value);
+ if (MINIFI_STATUS_SUCCESS != minifi_process_session_set_flow_file_attribute(impl_, ff.get(), utils::minifiStringView(key), &value_ref)) {
throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to set attribute");
}
}
void CffiProcessSession::removeAttribute(FlowFile& ff, const std::string_view key) {
- if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionSetFlowFileAttribute(impl_, ff.get(), utils::minifiStringView(key), nullptr)) {
+ if (MINIFI_STATUS_SUCCESS != minifi_process_session_set_flow_file_attribute(impl_, ff.get(), utils::minifiStringView(key), nullptr)) {
throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to remove attribute");
}
}
std::optional<std::string> CffiProcessSession::getAttribute(FlowFile& ff, std::string_view key) {
std::optional<std::string> result;
- MinifiProcessSessionGetFlowFileAttribute(impl_, ff.get(), utils::minifiStringView(key), [] (void* user_ctx, MinifiStringView value) {
+ minifi_process_session_get_flow_file_attribute(impl_, ff.get(), utils::minifiStringView(key), [] (void* user_ctx, minifi_string_view value) {
*static_cast<std::optional<std::string>*>(user_ctx) = std::string{value.data, value.length};
}, &result);
return result;
@@ -145,7 +145,7 @@ std::optional<std::string> CffiProcessSession::getAttribute(FlowFile& ff, std::s
std::map<std::string, std::string> CffiProcessSession::getAttributes(const FlowFile& ff) const {
std::map<std::string, std::string> result;
- MinifiProcessSessionGetFlowFileAttributes(impl_, ff.get(), [] (void* user_ctx, const MinifiStringView key, const MinifiStringView value) {
+ minifi_process_session_get_flow_file_attributes(impl_, ff.get(), [] (void* user_ctx, const minifi_string_view key, const minifi_string_view value) {
static_cast<std::map<std::string, std::string>*>(user_ctx)->insert({std::string{key.data, key.length}, std::string{value.data, value.length}});
}, &result);
return result;
@@ -153,14 +153,14 @@ std::map<std::string, std::string> CffiProcessSession::getAttributes(const FlowF
std::string CffiProcessSession::getFlowFileId(const FlowFile& ff) const {
std::string result;
- MinifiProcessSessionGetFlowFileId(impl_, ff.get(), [](void* user_ctx, const MinifiStringView value) {
+ minifi_process_session_get_flow_file_id(impl_, ff.get(), [](void* user_ctx, const minifi_string_view value) {
*static_cast<std::string*>(user_ctx) = std::string{value.data, value.length};
}, &result);
return result;
}
uint64_t CffiProcessSession::getFlowFileSize(const FlowFile& ff) const {
- return MinifiProcessSessionGetFlowFileSize(impl_, ff.get());
+ return minifi_process_session_get_flow_file_size(impl_, ff.get());
}
void ProcessSession::writeBuffer(FlowFile& flow_file, std::span<const char> buffer) {
diff --git a/extension-framework/cpp-extension-lib/src/core/ProcessorImpl.cpp b/extension-framework/cpp-extension-lib/src/core/ProcessorImpl.cpp
index 077d95929..f63aefedf 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessorImpl.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ProcessorImpl.cpp
@@ -31,7 +31,6 @@ namespace org::apache::nifi::minifi::api::core {
ProcessorImpl::ProcessorImpl(minifi::core::ProcessorMetadata metadata)
: metadata_(std::move(metadata)),
- trigger_when_empty_(false),
logger_(metadata_.logger) {
logger_->log_debug("Processor {} created with uuid {}", getName(), getUUIDStr());
}
@@ -52,7 +51,7 @@ utils::SmallString<36> ProcessorImpl::getUUIDStr() const {
return getUUID().to_string();
}
-MinifiStatus ProcessorImpl::onSchedule(ProcessContext& ctx) {
+minifi_status ProcessorImpl::onSchedule(ProcessContext& ctx) {
try {
return onScheduleImpl(ctx);
} catch (const std::exception& e) {
@@ -61,7 +60,7 @@ MinifiStatus ProcessorImpl::onSchedule(ProcessContext& ctx) {
}
}
-MinifiStatus ProcessorImpl::onTrigger(ProcessContext& ctx, ProcessSession& session) {
+minifi_status ProcessorImpl::onTrigger(ProcessContext& ctx, ProcessSession& session) {
try {
return onTriggerImpl(ctx, session);
} catch (const std::exception& e) {
diff --git a/extension-framework/cpp-extension-lib/src/core/logging/Logger.cpp b/extension-framework/cpp-extension-lib/src/core/logging/Logger.cpp
index 4f5f2f422..17db41928 100644
--- a/extension-framework/cpp-extension-lib/src/core/logging/Logger.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/logging/Logger.cpp
@@ -21,7 +21,7 @@ namespace org::apache::nifi::minifi::api::core::logging {
namespace {
-MinifiLogLevel toCLogLevel(minifi::core::logging::LOG_LEVEL lvl) {
+minifi_log_level toCLogLevel(minifi::core::logging::LOG_LEVEL lvl) {
switch (lvl) {
case minifi::core::logging::trace: return MINIFI_LOG_LEVEL_TRACE;
case minifi::core::logging::debug: return MINIFI_LOG_LEVEL_DEBUG;
@@ -33,37 +33,15 @@ MinifiLogLevel toCLogLevel(minifi::core::logging::LOG_LEVEL lvl) {
}
gsl_FailFast();
}
-
-minifi::core::logging::LOG_LEVEL toLogLevel(MinifiLogLevel level) {
- switch (level) {
- case MINIFI_LOG_LEVEL_TRACE: return minifi::core::logging::trace;
- case MINIFI_LOG_LEVEL_DEBUG: return minifi::core::logging::debug;
- case MINIFI_LOG_LEVEL_INFO: return minifi::core::logging::info;
- case MINIFI_LOG_LEVEL_WARNING: return minifi::core::logging::warn;
- case MINIFI_LOG_LEVEL_ERROR: return minifi::core::logging::err;
- case MINIFI_LOG_LEVEL_CRITICAL: return minifi::core::logging::critical;
- case MINIFI_LOG_LEVEL_OFF: return minifi::core::logging::off;
- }
- gsl_FailFast();
-}
-
} // namespace
-void CffiLogger::set_max_log_size(const int size) {
- MinifiLoggerSetMaxLogSize(impl_, size);
-}
void CffiLogger::log_string(const minifi::core::logging::LOG_LEVEL level, const std::string str) {
- MinifiLoggerLogString(impl_, toCLogLevel(level), MinifiStringView{.data = str.data(), .length = str.length()});
+ minifi_logger_log_string(impl_, toCLogLevel(level), minifi_string_view{.data = str.data(), .length = str.length()});
}
bool CffiLogger::should_log(const minifi::core::logging::LOG_LEVEL level) {
- return MinifiLoggerShouldLog(impl_, toCLogLevel(level));
-}
-
-[[nodiscard]] minifi::core::logging::LOG_LEVEL CffiLogger::level() const {
- return toLogLevel(MinifiLoggerLevel(impl_));
+ return minifi_logger_should_log(impl_, toCLogLevel(level));
}
-
} // namespace org::apache::nifi::minifi::api::core::logging
diff --git a/extension-framework/cpp-extension-lib/src/utils/minifi-c-utils.cpp b/extension-framework/cpp-extension-lib/src/utils/minifi-c-utils.cpp
index a2690d170..e11fd8972 100644
--- a/extension-framework/cpp-extension-lib/src/utils/minifi-c-utils.cpp
+++ b/extension-framework/cpp-extension-lib/src/utils/minifi-c-utils.cpp
@@ -20,7 +20,7 @@
namespace org::apache::nifi::minifi::api::utils {
-std::error_code make_error_code(MinifiStatus status) {
+std::error_code make_error_code(minifi_status status) {
switch (status) {
case MINIFI_STATUS_NOT_SUPPORTED_PROPERTY: return make_error_code(minifi::core::PropertyErrorCode::NotSupportedProperty);
case MINIFI_STATUS_DYNAMIC_PROPERTIES_NOT_SUPPORTED: return make_error_code(minifi::core::PropertyErrorCode::DynamicPropertiesNotSupported);
@@ -30,7 +30,7 @@ std::error_code make_error_code(MinifiStatus status) {
}
}
-MinifiValidator toStandardPropertyValidator(const minifi::core::PropertyValidator* validator) {
+minifi_validator toStandardPropertyValidator(const minifi::core::PropertyValidator* validator) {
if (validator->getEquivalentNifiStandardValidatorName() == minifi::core::StandardPropertyValidators::ALWAYS_VALID_VALIDATOR.getEquivalentNifiStandardValidatorName()) {
return MINIFI_VALIDATOR_ALWAYS_VALID;
}
diff --git a/extension-framework/include/utils/ExtensionInitUtils.h b/extension-framework/include/utils/ExtensionInitUtils.h
index 9ac895aab..685bc82f4 100644
--- a/extension-framework/include/utils/ExtensionInitUtils.h
+++ b/extension-framework/include/utils/ExtensionInitUtils.h
@@ -16,21 +16,22 @@
*/
#pragma once
+#include <functional>
+#include <optional>
#include <string>
#include <string_view>
-#include <optional>
-#include <functional>
-#include "minifi-c/minifi-c.h"
+
+#include "minifi-api.h"
namespace org::apache::nifi::minifi::utils {
-inline MinifiStringView toStringView(std::string_view str) {
- return MinifiStringView{.data = str.data(), .length = str.length()};
+inline minifi_string_view toStringView(std::string_view str) {
+ return minifi_string_view{.data = str.data(), .length = str.length()};
}
using ConfigReader = std::function<std::optional<std::string>(std::string_view key)>;
-static inline void MinifiRegisterCppExtension(MinifiExtensionContext* extension_context, const MinifiExtensionDefinition* extension_definition) {
+static inline void MinifiRegisterCppExtension(minifi_extension_context* extension_context, const minifi_extension_definition* extension_definition) {
MINIFI_REGISTER_EXTENSION_FN(extension_context, extension_definition);
}
diff --git a/extensions/ExtensionInitializer.cpp b/extensions/ExtensionInitializer.cpp
index 3cc5c6cb7..716a53dc7 100644
--- a/extensions/ExtensionInitializer.cpp
+++ b/extensions/ExtensionInitializer.cpp
@@ -15,15 +15,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "minifi-c/minifi-c.h"
-#include "utils/ExtensionInitUtils.h"
-#include "minifi-cpp/agent/agent_version.h"
#include "core/Resource.h"
+#include "minifi-api.h"
+#include "minifi-cpp/agent/agent_version.h"
+#include "utils/ExtensionInitUtils.h"
namespace minifi = org::apache::nifi::minifi;
-extern "C" void MinifiInitCppExtension(MinifiExtensionContext* extension_context) {
- MinifiExtensionDefinition extension_definition{
+extern "C" void minifi_init_cpp_extension(minifi_extension_context* extension_context) {
+ minifi_extension_definition extension_definition{
.name = minifi::utils::toStringView(MAKESTRING(MODULE_NAME)),
.version = minifi::utils::toStringView(minifi::AgentBuild::VERSION),
.deinit = nullptr,
diff --git a/extensions/aws/AwsLoader.cpp b/extensions/aws/AwsLoader.cpp
index 8ff738058..49ef2a5c3 100644
--- a/extensions/aws/AwsLoader.cpp
+++ b/extensions/aws/AwsLoader.cpp
@@ -17,11 +17,12 @@
#include <cstring>
#include <memory>
-#include "utils/AWSSdkLogger.h"
+
#include "aws/core/Aws.h"
-#include "minifi-c/minifi-c.h"
-#include "utils/ExtensionInitUtils.h"
+#include "minifi-api.h"
#include "minifi-cpp/agent/agent_version.h"
+#include "utils/AWSSdkLogger.h"
+#include "utils/ExtensionInitUtils.h"
#define MKSOC(x) #x
#define MAKESTRING(x) MKSOC(x) // NOLINT(cppcoreguidelines-macro-usage)
@@ -31,14 +32,14 @@ namespace minifi = org::apache::nifi::minifi;
namespace org::apache::nifi::minifi::aws::init {
namespace {
template<size_t N>
-MinifiStringView toStringView(const char (&s)[N]) { // NOLINT(cppcoreguidelines-avoid-c-arrays): using template argument deduction for string literal length requires reference to array
- return MinifiStringView{.data = s, .length = N};
+minifi_string_view toStringView(const char (&s)[N]) { // NOLINT(cppcoreguidelines-avoid-c-arrays): using template argument deduction for string literal length requires reference to array
+ return minifi_string_view{.data = s, .length = N};
}
-MinifiStringView toStringView(const char* s) {
+minifi_string_view toStringView(const char* s) {
if (!s) {
- return MinifiStringView{.data = nullptr, .length = 0};
+ return minifi_string_view{.data = nullptr, .length = 0};
}
- return MinifiStringView{.data = s, .length = std::strlen(s)};
+ return minifi_string_view{.data = s, .length = std::strlen(s)};
}
} // namespace
@@ -50,13 +51,13 @@ void deinit(gsl::owner<void*> sdk_opts_ptr) {
}
} // namespace org::apache::nifi::minifi::aws::init
-extern "C" void MinifiInitCppExtension(MinifiExtensionContext* extension_context) {
+extern "C" void minifi_init_cpp_extension(minifi_extension_context* extension_context) {
using minifi::aws::init::toStringView;
auto sdk_options = std::make_unique<Aws::SDKOptions>();
Aws::InitAPI(*sdk_options);
Aws::Utils::Logging::InitializeAWSLogging(std::make_shared<minifi::aws::utils::AWSSdkLogger>());
- MinifiExtensionDefinition ext_definition{.name = toStringView(MAKESTRING(MODULE_NAME)),
+ minifi_extension_definition ext_definition{.name = toStringView(MAKESTRING(MODULE_NAME)),
.version = toStringView(minifi::AgentBuild::VERSION),
.deinit = &minifi::aws::init::deinit,
.user_data = sdk_options.get()};
diff --git a/extensions/aws/utils/AWSSdkLogger.cpp b/extensions/aws/utils/AWSSdkLogger.cpp
index fd8a5d9c3..af7a02fed 100644
--- a/extensions/aws/utils/AWSSdkLogger.cpp
+++ b/extensions/aws/utils/AWSSdkLogger.cpp
@@ -17,10 +17,11 @@
*/
#include "AWSSdkLogger.h"
-#include <cstdio>
#include <cstdarg>
+#include <cstdio>
#include "aws/core/utils/logging/LogLevel.h"
+#include "minifi-cpp/core/logging/AdvancedLogger.h"
#include "minifi-cpp/utils/gsl.h"
namespace org::apache::nifi::minifi::aws::utils {
@@ -58,7 +59,8 @@ core::logging::LOG_LEVEL mapFromAwsLevels(Aws::Utils::Logging::LogLevel level) {
} // namespace
Aws::Utils::Logging::LogLevel AWSSdkLogger::GetLogLevel() const {
- return mapToAwsLevels(logger_->level());
+ const auto level = core::logging::AdvancedLogger::getLevel(logger_.get()) | minifi::utils::orTerminate("logger_ should be an AdvancedLogger");
+ return mapToAwsLevels(level);
}
void AWSSdkLogger::Log(Aws::Utils::Logging::LogLevel log_level, const char* tag, const char* format_str, ...) { // NOLINT(cert-dcl50-cpp)
diff --git a/extensions/gcp/ExtensionInitializer.cpp b/extensions/gcp/ExtensionInitializer.cpp
index 89b00e9ff..267c5fed6 100644
--- a/extensions/gcp/ExtensionInitializer.cpp
+++ b/extensions/gcp/ExtensionInitializer.cpp
@@ -28,16 +28,16 @@
namespace minifi = org::apache::nifi::minifi;
-CEXTENSIONAPI const uint32_t MinifiApiVersion = MINIFI_API_VERSION;
+CEXTENSIONAPI const uint32_t minifi_api_version = MINIFI_API_VERSION;
-CEXTENSIONAPI void MinifiInitExtension(MinifiExtensionContext* extension_context) {
- const MinifiExtensionDefinition extension_definition{
+CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* extension_context) {
+ const minifi_extension_definition extension_definition{
.name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
.version = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_VERSION)),
.deinit = nullptr,
.user_data = nullptr
};
- auto* extension = MinifiRegisterExtension(extension_context, &extension_definition);
+ auto* extension = minifi_register_extension(extension_context, &extension_definition);
minifi::api::core::registerProcessors<minifi::extensions::gcp::DeleteGCSObject,
minifi::extensions::gcp::FetchGCSObject,
minifi::extensions::gcp::ListGCSBucket,
diff --git a/extensions/gcp/controllerservices/GCPCredentialsControllerService.cpp b/extensions/gcp/controllerservices/GCPCredentialsControllerService.cpp
index d84bbeb4b..6408a5cee 100644
--- a/extensions/gcp/controllerservices/GCPCredentialsControllerService.cpp
+++ b/extensions/gcp/controllerservices/GCPCredentialsControllerService.cpp
@@ -56,7 +56,7 @@ std::shared_ptr<google::cloud::Credentials> GCPCredentialsControllerService::cre
return google::cloud::MakeServiceAccountCredentials(*json_contents);
}
-MinifiStatus GCPCredentialsControllerService::enableImpl(api::core::ControllerServiceContext& ctx) {
+minifi_status GCPCredentialsControllerService::enableImpl(api::core::ControllerServiceContext& ctx) {
std::optional<CredentialsLocation> credentials_location;
if (const auto value = ctx.getProperty(CredentialsLoc.name)) {
credentials_location = magic_enum::enum_cast<CredentialsLocation>(*value);
diff --git a/extensions/gcp/controllerservices/GCPCredentialsControllerService.h b/extensions/gcp/controllerservices/GCPCredentialsControllerService.h
index b53971ae2..d7f56b28e 100644
--- a/extensions/gcp/controllerservices/GCPCredentialsControllerService.h
+++ b/extensions/gcp/controllerservices/GCPCredentialsControllerService.h
@@ -92,7 +92,7 @@ class GCPCredentialsControllerService : public api::core::ControllerServiceImpl
using ControllerServiceImpl::ControllerServiceImpl;
- MinifiStatus enableImpl(api::core::ControllerServiceContext& ctx) override;
+ minifi_status enableImpl(api::core::ControllerServiceContext& ctx) override;
[[nodiscard]] const auto& getCredentials() const { return credentials_; }
diff --git a/extensions/gcp/processors/DeleteGCSObject.cpp b/extensions/gcp/processors/DeleteGCSObject.cpp
index 35206f17e..82510f683 100644
--- a/extensions/gcp/processors/DeleteGCSObject.cpp
+++ b/extensions/gcp/processors/DeleteGCSObject.cpp
@@ -27,7 +27,7 @@ namespace gcs = ::google::cloud::storage;
namespace org::apache::nifi::minifi::extensions::gcp {
-MinifiStatus DeleteGCSObject::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
+minifi_status DeleteGCSObject::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
gsl_Expects(gcp_credentials_);
auto flow_file = session.get();
diff --git a/extensions/gcp/processors/DeleteGCSObject.h b/extensions/gcp/processors/DeleteGCSObject.h
index 48364cd45..2666fbe7b 100644
--- a/extensions/gcp/processors/DeleteGCSObject.h
+++ b/extensions/gcp/processors/DeleteGCSObject.h
@@ -76,7 +76,7 @@ class DeleteGCSObject : public GCSProcessor {
EXTENSIONAPI static constexpr bool IsSingleThreaded = false;
protected:
- MinifiStatus onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
+ minifi_status onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
};
} // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/FetchGCSObject.cpp b/extensions/gcp/processors/FetchGCSObject.cpp
index 93f500806..679e05ee0 100644
--- a/extensions/gcp/processors/FetchGCSObject.cpp
+++ b/extensions/gcp/processors/FetchGCSObject.cpp
@@ -80,7 +80,7 @@ class FetchFromGCSCallback {
};
} // namespace
-MinifiStatus FetchGCSObject::onScheduleImpl(api::core::ProcessContext& context) {
+minifi_status FetchGCSObject::onScheduleImpl(api::core::ProcessContext& context) {
const auto status = GCSProcessor::onScheduleImpl(context);
if (MINIFI_STATUS_SUCCESS != status) { return status; }
if (auto encryption_key = context.getProperty(EncryptionKey, nullptr)) {
@@ -94,7 +94,7 @@ MinifiStatus FetchGCSObject::onScheduleImpl(api::core::ProcessContext& context)
return MINIFI_STATUS_SUCCESS;
}
-MinifiStatus FetchGCSObject::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
+minifi_status FetchGCSObject::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
gsl_Expects(gcp_credentials_);
auto flow_file = session.get();
diff --git a/extensions/gcp/processors/FetchGCSObject.h b/extensions/gcp/processors/FetchGCSObject.h
index 51d72dab9..8bd83356d 100644
--- a/extensions/gcp/processors/FetchGCSObject.h
+++ b/extensions/gcp/processors/FetchGCSObject.h
@@ -77,8 +77,8 @@ class FetchGCSObject : public GCSProcessor {
EXTENSIONAPI static constexpr bool IsSingleThreaded = false;
protected:
- MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
- MinifiStatus onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
+ minifi_status onScheduleImpl(api::core::ProcessContext& context) override;
+ minifi_status onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
private:
google::cloud::storage::EncryptionKey encryption_key_;
diff --git a/extensions/gcp/processors/GCSProcessor.cpp b/extensions/gcp/processors/GCSProcessor.cpp
index bf9342af8..1199bba88 100644
--- a/extensions/gcp/processors/GCSProcessor.cpp
+++ b/extensions/gcp/processors/GCSProcessor.cpp
@@ -31,7 +31,7 @@ std::shared_ptr<google::cloud::Credentials> GCSProcessor::getCredentials(const a
return nullptr;
}
-MinifiStatus GCSProcessor::onScheduleImpl(api::core::ProcessContext& context) {
+minifi_status GCSProcessor::onScheduleImpl(api::core::ProcessContext& context) {
if (const auto number_of_retries = api::utils::parseOptionalU64Property(context, NumberOfRetries)) {
retry_policy_ = std::make_shared<google::cloud::storage::LimitedErrorCountRetryPolicy>(gsl::narrow<int>(*number_of_retries));
}
diff --git a/extensions/gcp/processors/GCSProcessor.h b/extensions/gcp/processors/GCSProcessor.h
index a05dde421..99d1ee4cd 100644
--- a/extensions/gcp/processors/GCSProcessor.h
+++ b/extensions/gcp/processors/GCSProcessor.h
@@ -53,7 +53,7 @@ class GCSProcessor : public api::core::ProcessorImpl {
.build();
EXTENSIONAPI static constexpr auto ProxyConfigurationService = core::PropertyDefinitionBuilder<>::createProperty("Proxy Configuration Service")
.withDescription("Specifies the Proxy Configuration Controller Service to proxy network requests.")
- .withAllowedType<MINIFI_PROXY_CONFIGURATION_SERVICE_PROPERTY_TYPE>()
+ .withAllowedType<MINIFI_PROXY_CONFIGURATION_SERVICE_INTERFACE_PROPERTY_TYPE>()
.build();
EXTENSIONAPI static constexpr auto Properties = std::to_array<core::PropertyReference>({
GCPCredentials,
@@ -63,7 +63,7 @@ class GCSProcessor : public api::core::ProcessorImpl {
});
protected:
- MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
+ minifi_status onScheduleImpl(api::core::ProcessContext& context) override;
virtual google::cloud::storage::Client getClient() const;
static std::shared_ptr<google::cloud::Credentials> getCredentials(const api::core::ProcessContext& context);
diff --git a/extensions/gcp/processors/ListGCSBucket.cpp b/extensions/gcp/processors/ListGCSBucket.cpp
index ac4cadd40..0c16e6d69 100644
--- a/extensions/gcp/processors/ListGCSBucket.cpp
+++ b/extensions/gcp/processors/ListGCSBucket.cpp
@@ -28,16 +28,16 @@ namespace gcs = ::google::cloud::storage;
namespace org::apache::nifi::minifi::extensions::gcp {
-MinifiStatus ListGCSBucket::onScheduleImpl(api::core::ProcessContext& context) {
+minifi_status ListGCSBucket::onScheduleImpl(api::core::ProcessContext& context) {
const auto status = GCSProcessor::onScheduleImpl(context);
- if (status != MinifiStatus::MINIFI_STATUS_SUCCESS) {
+ if (status != minifi_status::MINIFI_STATUS_SUCCESS) {
return status;
}
bucket_ = api::utils::parseProperty(context, Bucket);
- return MinifiStatus::MINIFI_STATUS_SUCCESS;
+ return minifi_status::MINIFI_STATUS_SUCCESS;
}
-MinifiStatus ListGCSBucket::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
+minifi_status ListGCSBucket::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
gsl_Expects(gcp_credentials_);
gcs::Client client = getClient();
@@ -54,7 +54,7 @@ MinifiStatus ListGCSBucket::onTriggerImpl(api::core::ProcessContext& context, ap
logger_->log_error("Invalid object in bucket {}", bucket_);
}
}
- return MinifiStatus::MINIFI_STATUS_SUCCESS;
+ return minifi_status::MINIFI_STATUS_SUCCESS;
}
} // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/ListGCSBucket.h b/extensions/gcp/processors/ListGCSBucket.h
index e654a314d..1e7e01c25 100644
--- a/extensions/gcp/processors/ListGCSBucket.h
+++ b/extensions/gcp/processors/ListGCSBucket.h
@@ -118,8 +118,8 @@ class ListGCSBucket : public GCSProcessor {
EXTENSIONAPI static constexpr bool IsSingleThreaded = true;
protected:
- MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
- MinifiStatus onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
+ minifi_status onScheduleImpl(api::core::ProcessContext& context) override;
+ minifi_status onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
private:
std::string bucket_;
diff --git a/extensions/gcp/processors/PutGCSObject.cpp b/extensions/gcp/processors/PutGCSObject.cpp
index 97178eb9a..c6c651b89 100644
--- a/extensions/gcp/processors/PutGCSObject.cpp
+++ b/extensions/gcp/processors/PutGCSObject.cpp
@@ -102,9 +102,9 @@ class UploadToGCSCallback {
} // namespace
-MinifiStatus PutGCSObject::onScheduleImpl(api::core::ProcessContext& context) {
+minifi_status PutGCSObject::onScheduleImpl(api::core::ProcessContext& context) {
const auto status = GCSProcessor::onScheduleImpl(context);
- if (status != MinifiStatus::MINIFI_STATUS_SUCCESS) {
+ if (status != minifi_status::MINIFI_STATUS_SUCCESS) {
return status;
}
@@ -119,7 +119,7 @@ MinifiStatus PutGCSObject::onScheduleImpl(api::core::ProcessContext& context) {
return MINIFI_STATUS_SUCCESS;
}
-MinifiStatus PutGCSObject::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
+minifi_status PutGCSObject::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
gsl_Expects(gcp_credentials_);
auto flow_file = session.get();
diff --git a/extensions/gcp/processors/PutGCSObject.h b/extensions/gcp/processors/PutGCSObject.h
index 73d8c8218..2da270c74 100644
--- a/extensions/gcp/processors/PutGCSObject.h
+++ b/extensions/gcp/processors/PutGCSObject.h
@@ -190,8 +190,8 @@ class PutGCSObject : public GCSProcessor {
EXTENSIONAPI static constexpr bool IsSingleThreaded = false;
protected:
- MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
- MinifiStatus onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
+ minifi_status onScheduleImpl(api::core::ProcessContext& context) override;
+ minifi_status onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
private:
google::cloud::storage::EncryptionKey encryption_key_;
diff --git a/extensions/kafka/ConsumeKafka.cpp b/extensions/kafka/ConsumeKafka.cpp
index c01c20402..25f63568f 100644
--- a/extensions/kafka/ConsumeKafka.cpp
+++ b/extensions/kafka/ConsumeKafka.cpp
@@ -39,7 +39,7 @@ struct std::hash<org::apache::nifi::minifi::processors::ConsumeKafka::KafkaMessa
namespace org::apache::nifi::minifi::processors {
-MinifiStatus ConsumeKafka::onScheduleImpl(api::core::ProcessContext& context) {
+minifi_status ConsumeKafka::onScheduleImpl(api::core::ProcessContext& context) {
using utils::KafkaEncoding;
namespace utils = api::utils;
// Required properties
@@ -66,7 +66,7 @@ MinifiStatus ConsumeKafka::onScheduleImpl(api::core::ProcessContext& context) {
configureNewConnection(context);
if (commit_policy_ == consume_kafka::CommitPolicyEnum::CommitFromIncomingFlowFiles) {
- setTriggerWhenEmpty(true);
+ std::ignore = context.setTriggerWhenEmpty(true);
}
return MINIFI_STATUS_SUCCESS;
}
@@ -339,7 +339,7 @@ void ConsumeKafka::commitOffsetsFromIncomingFlowFiles(api::core::ProcessSession&
for (auto& [ff, relationship]: flow_files) { session.transfer(std::move(ff), relationship); }
}
-MinifiStatus ConsumeKafka::processMessages(api::core::ProcessSession& session, const std::unordered_map<KafkaMessageLocation, MessageBundle>& message_bundles) const {
+minifi_status ConsumeKafka::processMessages(api::core::ProcessSession& session, const std::unordered_map<KafkaMessageLocation, MessageBundle>& message_bundles) const {
for (const auto& msg_bundle: message_bundles | std::views::values) {
for (const auto& message: msg_bundle.getMessages()) {
std::string message_content = extractMessage(*message);
@@ -353,7 +353,7 @@ MinifiStatus ConsumeKafka::processMessages(api::core::ProcessSession& session, c
return MINIFI_STATUS_SUCCESS;
}
-MinifiStatus ConsumeKafka::processMessageBundles(api::core::ProcessSession& session,
+minifi_status ConsumeKafka::processMessageBundles(api::core::ProcessSession& session,
const std::unordered_map<KafkaMessageLocation, MessageBundle>& message_bundles, const std::string_view message_demarcator) const {
for (const auto& msg_bundle: message_bundles | std::views::values) {
auto flow_file = session.create();
@@ -367,7 +367,7 @@ MinifiStatus ConsumeKafka::processMessageBundles(api::core::ProcessSession& sess
return MINIFI_STATUS_SUCCESS;
}
-MinifiStatus ConsumeKafka::onTriggerImpl(api::core::ProcessContext&, api::core::ProcessSession& session) {
+minifi_status ConsumeKafka::onTriggerImpl(api::core::ProcessContext&, api::core::ProcessSession& session) {
if (commit_policy_ == consume_kafka::CommitPolicyEnum::CommitFromIncomingFlowFiles) { commitOffsetsFromIncomingFlowFiles(session); }
const auto message_bundles = pollKafkaMessages();
if (message_bundles.empty()) {
diff --git a/extensions/kafka/ConsumeKafka.h b/extensions/kafka/ConsumeKafka.h
index ded235f09..8c592186c 100644
--- a/extensions/kafka/ConsumeKafka.h
+++ b/extensions/kafka/ConsumeKafka.h
@@ -284,8 +284,8 @@ class ConsumeKafka final : public KafkaProcessorBase {
ConsumeKafka& operator=(ConsumeKafka&&) = delete;
~ConsumeKafka() override = default;
- MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
- MinifiStatus onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
+ minifi_status onScheduleImpl(api::core::ProcessContext& context) override;
+ minifi_status onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
private:
struct KafkaMessageLocation {
@@ -321,8 +321,8 @@ class ConsumeKafka final : public KafkaProcessorBase {
std::vector<std::pair<std::string, std::string>> getFlowFilesAttributesFromMessageHeaders(const rd_kafka_message_t& message) const;
void addAttributesToSingleMessageFlowFile(api::core::ProcessSession& session, api::core::FlowFile& flow_file, const rd_kafka_message_t& message) const;
void addAttributesToMessageBundleFlowFile(api::core::ProcessSession& session, api::core::FlowFile& flow_file, const MessageBundle& message_bundle) const;
- MinifiStatus processMessages(api::core::ProcessSession& session, const std::unordered_map<KafkaMessageLocation, MessageBundle>& message_bundles) const;
- MinifiStatus processMessageBundles(api::core::ProcessSession& session, const std::unordered_map<KafkaMessageLocation, MessageBundle>& message_bundles,
+ minifi_status processMessages(api::core::ProcessSession& session, const std::unordered_map<KafkaMessageLocation, MessageBundle>& message_bundles) const;
+ minifi_status processMessageBundles(api::core::ProcessSession& session, const std::unordered_map<KafkaMessageLocation, MessageBundle>& message_bundles,
std::string_view message_demarcator) const;
void commitOffsetsFromMessages(const std::unordered_map<KafkaMessageLocation, MessageBundle>& message_bundles) const;
diff --git a/extensions/kafka/ExtensionInitializer.cpp b/extensions/kafka/ExtensionInitializer.cpp
index 792ab083b..b85521acb 100644
--- a/extensions/kafka/ExtensionInitializer.cpp
+++ b/extensions/kafka/ExtensionInitializer.cpp
@@ -25,13 +25,13 @@
namespace minifi = org::apache::nifi::minifi;
-CEXTENSIONAPI const uint32_t MinifiApiVersion = MINIFI_API_VERSION;
+CEXTENSIONAPI const uint32_t minifi_api_version = MINIFI_API_VERSION;
-CEXTENSIONAPI void MinifiInitExtension(MinifiExtensionContext* extension_context) {
- const MinifiExtensionDefinition extension_definition{.name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
+CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* extension_context) {
+ const minifi_extension_definition extension_definition{.name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
.version = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_VERSION)),
.deinit = nullptr,
.user_data = nullptr};
- auto* extension = MinifiRegisterExtension(extension_context, &extension_definition);
+ auto* extension = minifi_register_extension(extension_context, &extension_definition);
minifi::api::core::registerProcessors<minifi::processors::PublishKafka, minifi::processors::ConsumeKafka>(extension);
}
diff --git a/extensions/kafka/KafkaProcessorBase.h b/extensions/kafka/KafkaProcessorBase.h
index 6cdc2feec..a72e628f3 100644
--- a/extensions/kafka/KafkaProcessorBase.h
+++ b/extensions/kafka/KafkaProcessorBase.h
@@ -26,7 +26,7 @@
#include "core/PropertyDefinitionBuilder.h"
#include "rdkafka_utils.h"
#include "utils/Enum.h"
-#include "minifi-c.h"
+#include "minifi-api.h"
namespace org::apache::nifi::minifi::processors {
@@ -41,7 +41,7 @@ class KafkaProcessorBase : public api::core::ProcessorImpl {
EXTENSIONAPI static constexpr auto SSLContextService =
core::PropertyDefinitionBuilder<>::createProperty("SSL Context Service")
.withDescription("SSL Context Service Name")
- .withAllowedType<MINIFI_SSL_CONTEXT_SERVICE_PROPERTY_TYPE>()
+ .withAllowedType<MINIFI_SSL_CONTEXT_SERVICE_INTERFACE_PROPERTY_TYPE>()
.build();
EXTENSIONAPI static constexpr auto SecurityProtocol =
core::PropertyDefinitionBuilder<magic_enum::enum_count<kafka::SecurityProtocolOption>()>::createProperty(
diff --git a/extensions/kafka/PublishKafka.cpp b/extensions/kafka/PublishKafka.cpp
index 05f3de3e2..b934675ca 100644
--- a/extensions/kafka/PublishKafka.cpp
+++ b/extensions/kafka/PublishKafka.cpp
@@ -328,7 +328,7 @@ void messageDeliveryCallback(rd_kafka_t* rk, const rd_kafka_message_t* rkmessage
}
} // namespace
-MinifiStatus PublishKafka::onScheduleImpl(api::core::ProcessContext& context) {
+minifi_status PublishKafka::onScheduleImpl(api::core::ProcessContext& context) {
interrupted_ = false;
// Try to get a KafkaConnection
@@ -607,7 +607,7 @@ std::optional<api::utils::net::SslData> PublishKafka::getSslData(api::core::Proc
return ssl_data;
}
-MinifiStatus PublishKafka::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
+minifi_status PublishKafka::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
// Check whether we have been interrupted
if (interrupted_) {
logger_->log_info("The processor has been interrupted, not running onTrigger");
diff --git a/extensions/kafka/PublishKafka.h b/extensions/kafka/PublishKafka.h
index 682e4218e..2096fe42b 100644
--- a/extensions/kafka/PublishKafka.h
+++ b/extensions/kafka/PublishKafka.h
@@ -250,8 +250,8 @@ class PublishKafka final : public KafkaProcessorBase {
PublishKafka& operator=(PublishKafka&&) = delete;
~PublishKafka() override = default;
- MinifiStatus onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
- MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
+ minifi_status onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
+ minifi_status onScheduleImpl(api::core::ProcessContext& context) override;
void onUnSchedule() override;
class Messages;
diff --git a/extensions/kafka/tests/PublishKafkaTests.cpp b/extensions/kafka/tests/PublishKafkaTests.cpp
index 367b776e0..f0de8ec9f 100644
--- a/extensions/kafka/tests/PublishKafkaTests.cpp
+++ b/extensions/kafka/tests/PublishKafkaTests.cpp
@@ -49,7 +49,7 @@ TEST_CASE("Trigger without valid broker", "[PublishKafka]") {
REQUIRE_NOTHROW(publish_kafka.onScheduleImpl(context));
auto session = mock::MockProcessSession{};
- session.addInputFlowFile(std::make_unique<MinifiFlowFile>("test_input_flow_file"));
+ session.addInputFlowFile(std::make_unique<minifi_flow_file>("test_input_flow_file"));
REQUIRE_NOTHROW(publish_kafka.onTriggerImpl(context, session));
CHECK_FALSE(logger->logs_.empty());
diff --git a/extensions/libarchive/BinFiles.cpp b/extensions/libarchive/BinFiles.cpp
index b4e0c8df4..584eff433 100644
--- a/extensions/libarchive/BinFiles.cpp
+++ b/extensions/libarchive/BinFiles.cpp
@@ -72,7 +72,7 @@ void BinFiles::onSchedule(core::ProcessContext& context, core::ProcessSessionFac
if (auto max_bin_age = utils::parseOptionalDurationProperty(context, MaxBinAge)) {
// We need to trigger the processor even when there are no incoming flow files so that it can flush the bins.
- setTriggerWhenEmpty(true);
+ context.setTriggerWhenEmpty(true);
this->binManager_.setBinAge(*max_bin_age);
logger_->log_debug("BinFiles: MaxBinAge [{}]", *max_bin_age);
}
diff --git a/extensions/llamacpp/ExtensionInitializer.cpp b/extensions/llamacpp/ExtensionInitializer.cpp
index 09b142ff2..e2ee8b9f6 100644
--- a/extensions/llamacpp/ExtensionInitializer.cpp
+++ b/extensions/llamacpp/ExtensionInitializer.cpp
@@ -24,15 +24,15 @@
namespace minifi = org::apache::nifi::minifi;
-CEXTENSIONAPI const uint32_t MinifiApiVersion = MINIFI_API_VERSION;
+CEXTENSIONAPI const uint32_t minifi_api_version = MINIFI_API_VERSION;
-CEXTENSIONAPI void MinifiInitExtension(MinifiExtensionContext* extension_context) {
- MinifiExtensionDefinition extension_definition{
+CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* extension_context) {
+ minifi_extension_definition extension_definition{
.name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
.version = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_VERSION)),
.deinit = nullptr,
.user_data = nullptr
};
- auto* extension = MinifiRegisterExtension(extension_context, &extension_definition);
+ auto* extension = minifi_register_extension(extension_context, &extension_definition);
minifi::api::core::registerProcessors<minifi::extensions::llamacpp::processors::RunLlamaCppInference>(extension);
}
diff --git a/extensions/llamacpp/processors/RunLlamaCppInference.cpp b/extensions/llamacpp/processors/RunLlamaCppInference.cpp
index 888280cb1..d937f3012 100644
--- a/extensions/llamacpp/processors/RunLlamaCppInference.cpp
+++ b/extensions/llamacpp/processors/RunLlamaCppInference.cpp
@@ -28,7 +28,7 @@
namespace org::apache::nifi::minifi::extensions::llamacpp::processors {
-MinifiStatus RunLlamaCppInference::onScheduleImpl(api::core::ProcessContext& context) {
+minifi_status RunLlamaCppInference::onScheduleImpl(api::core::ProcessContext& context) {
model_path_.clear();
model_path_ = api::utils::parseProperty(context, ModelPath);
multimodal_model_path_ = api::utils::parseOptionalProperty(context, MultiModalModelPath);
@@ -69,7 +69,10 @@ void RunLlamaCppInference::increaseTokensOut(uint64_t token_count) {
metrics_.tokens_out += token_count;
}
-MinifiStatus RunLlamaCppInference::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
+minifi_status RunLlamaCppInference::onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) {
+ auto report_metrics = gsl::finally([&](){
+ std::ignore = context.reportMetrics(metrics_.calculateMetrics());
+ });
auto flow_file = session.get();
if (!flow_file) {
return MINIFI_STATUS_PROCESSOR_YIELD;
diff --git a/extensions/llamacpp/processors/RunLlamaCppInference.h b/extensions/llamacpp/processors/RunLlamaCppInference.h
index 5634df4de..16cf7a9c6 100644
--- a/extensions/llamacpp/processors/RunLlamaCppInference.h
+++ b/extensions/llamacpp/processors/RunLlamaCppInference.h
@@ -169,10 +169,9 @@ class RunLlamaCppInference : public api::core::ProcessorImpl {
EXTENSIONAPI static constexpr core::annotation::Input InputRequirement = core::annotation::Input::INPUT_REQUIRED;
EXTENSIONAPI static constexpr bool IsSingleThreaded = true;
- MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
- MinifiStatus onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
+ minifi_status onScheduleImpl(api::core::ProcessContext& context) override;
+ minifi_status onTriggerImpl(api::core::ProcessContext& context, api::core::ProcessSession& session) override;
void onUnSchedule() override;
- minifi::api::core::PublishedMetrics calculateMetrics() const override {return metrics_.calculateMetrics();}
private:
void increaseTokensIn(uint64_t token_count);
diff --git a/extensions/python/PythonCreator.h b/extensions/python/PythonCreator.h
index 6b64d9a52..a043d01b6 100644
--- a/extensions/python/PythonCreator.h
+++ b/extensions/python/PythonCreator.h
@@ -32,15 +32,16 @@
#include "core/Resource.h"
#include "core/logging/LoggerFactory.h"
#include "minifi-cpp/agent/agent_version.h"
+#include "minifi-cpp/core/ProcessorDescriptor.h"
+#include "minifi-cpp/core/logging/AdvancedLogger.h"
#include "range/v3/algorithm.hpp"
#include "range/v3/view/filter.hpp"
#include "utils/Environment.h"
+#include "utils/ExtensionInitUtils.h"
#include "utils/Locations.h"
#include "utils/StringUtils.h"
#include "utils/file/FilePattern.h"
#include "utils/file/FileUtils.h"
-#include "minifi-cpp/core/ProcessorDescriptor.h"
-#include "utils/ExtensionInitUtils.h"
namespace org::apache::nifi::minifi::extensions::python {
@@ -53,7 +54,7 @@ class DummyProcessorDescriptor : public core::ProcessorDescriptor {
void setSupportedProperties(std::span<const core::Property> /*properties*/) override {}
};
-class DummyLogger : public core::logging::Logger {
+class DummyLogger : public core::logging::AdvancedLogger {
public:
void set_max_log_size(int /*size*/) override {}
void log_string(core::logging::LOG_LEVEL /*level*/, std::string /*str*/) override {}
diff --git a/extensions/python/pythonlibloader/PythonLibLoader.cpp b/extensions/python/pythonlibloader/PythonLibLoader.cpp
index e080da22a..77fe8e60c 100644
--- a/extensions/python/pythonlibloader/PythonLibLoader.cpp
+++ b/extensions/python/pythonlibloader/PythonLibLoader.cpp
@@ -16,16 +16,18 @@
* limitations under the License.
*/
#include <dlfcn.h>
+
+#include <array>
#include <cstdio>
#include <iostream>
#include <string>
-#include <array>
-#include "utils/StringUtils.h"
+
+#include "core/Resource.h"
#include "core/logging/LoggerFactory.h"
+#include "minifi-api.h"
#include "minifi-cpp/agent/agent_version.h"
-#include "minifi-c/minifi-c.h"
#include "utils/ExtensionInitUtils.h"
-#include "core/Resource.h"
+#include "utils/StringUtils.h"
#if defined(WIN32)
static_assert(false, "The Python library loader should only be used on Linux or macOS.");
@@ -98,15 +100,15 @@ class PythonLibLoader {
std::shared_ptr<minifi::core::logging::Logger> logger_ = minifi::core::logging::LoggerFactory<PythonLibLoader>::getLogger();
};
-extern "C" void MinifiInitCppExtension(MinifiExtensionContext* extension_context) {
+extern "C" void minifi_init_cpp_extension(minifi_extension_context* extension_context) {
static PythonLibLoader python_lib_loader([&] (std::string_view key) -> std::optional<std::string> {
std::optional<std::string> result;
- MinifiConfigGet(extension_context, minifi::utils::toStringView(key), [] (void* user_data, MinifiStringView value) {
+ minifi_config_get(extension_context, minifi::utils::toStringView(key), [] (void* user_data, minifi_string_view value) {
*static_cast<std::optional<std::string>*>(user_data) = std::string{value.data, value.length};
}, &result);
return result;
});
- MinifiExtensionDefinition extension_definition{
+ minifi_extension_definition extension_definition{
.name = minifi::utils::toStringView(MAKESTRING(MODULE_NAME)),
.version = minifi::utils::toStringView(minifi::AgentBuild::VERSION),
.deinit = nullptr,
diff --git a/extensions/python/pythonloader/PyProcLoader.cpp b/extensions/python/pythonloader/PyProcLoader.cpp
index 9d81503fc..09440c1e7 100644
--- a/extensions/python/pythonloader/PyProcLoader.cpp
+++ b/extensions/python/pythonloader/PyProcLoader.cpp
@@ -15,12 +15,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "minifi-c/minifi-c.h"
-#include "utils/ExtensionInitUtils.h"
-#include "PythonCreator.h"
-#include "minifi-cpp/agent/agent_version.h"
#include "PythonBindings.h"
+#include "PythonCreator.h"
#include "core/Resource.h"
+#include "minifi-api.h"
+#include "minifi-cpp/agent/agent_version.h"
+#include "utils/ExtensionInitUtils.h"
namespace minifi = org::apache::nifi::minifi;
@@ -33,15 +33,15 @@ static minifi::extensions::python::PythonCreator& getPythonCreator() {
// the symbols of the python library
extern "C" const int LOAD_MODULE_AS_GLOBAL = 1;
-extern "C" void MinifiInitCppExtension(MinifiExtensionContext* extension_context) {
+extern "C" void minifi_init_cpp_extension(minifi_extension_context* extension_context) {
getPythonCreator().configure([&] (std::string_view key) -> std::optional<std::string> {
std::optional<std::string> result;
- MinifiConfigGet(extension_context, minifi::utils::toStringView(key), [] (void* user_data, MinifiStringView value) {
+ minifi_config_get(extension_context, minifi::utils::toStringView(key), [] (void* user_data, minifi_string_view value) {
*static_cast<std::optional<std::string>*>(user_data) = std::string{value.data, value.length};
}, &result);
return result;
});
- MinifiExtensionDefinition extension_definition{
+ minifi_extension_definition extension_definition{
.name = minifi::utils::toStringView(MAKESTRING(MODULE_NAME)),
.version = minifi::utils::toStringView(minifi::AgentBuild::VERSION),
.deinit = nullptr,
diff --git a/extensions/sftp/SFTPLoader.cpp b/extensions/sftp/SFTPLoader.cpp
index 620a31f41..88d0ec200 100644
--- a/extensions/sftp/SFTPLoader.cpp
+++ b/extensions/sftp/SFTPLoader.cpp
@@ -16,16 +16,16 @@
* limitations under the License.
*/
+#include "client/SFTPClient.h"
+#include "core/Resource.h"
+#include "minifi-api.h"
#include "minifi-cpp/agent/agent_version.h"
#include "minifi-cpp/properties/Configure.h"
-#include "client/SFTPClient.h"
-#include "minifi-c/minifi-c.h"
#include "utils/ExtensionInitUtils.h"
-#include "core/Resource.h"
namespace minifi = org::apache::nifi::minifi;
-extern "C" void MinifiInitCppExtension(MinifiExtensionContext* extension_context) {
+extern "C" void minifi_init_cpp_extension(minifi_extension_context* extension_context) {
if (libssh2_init(0) != 0) {
return;
}
@@ -33,7 +33,7 @@ extern "C" void MinifiInitCppExtension(MinifiExtensionContext* extension_context
libssh2_exit();
return;
}
- MinifiExtensionDefinition extension_definition{
+ minifi_extension_definition extension_definition{
.name = minifi::utils::toStringView(MAKESTRING(MODULE_NAME)),
.version = minifi::utils::toStringView(minifi::AgentBuild::VERSION),
.deinit = [] (void* /*user_data*/) {
diff --git a/extensions/stable-api-testing/AnimalControllerServices.cpp b/extensions/stable-api-testing/AnimalControllerServices.cpp
index f05ab6eea..be4ddd8a0 100644
--- a/extensions/stable-api-testing/AnimalControllerServices.cpp
+++ b/extensions/stable-api-testing/AnimalControllerServices.cpp
@@ -20,7 +20,7 @@
#include "api/utils/ProcessorConfigUtils.h"
namespace org::apache::nifi::minifi::api_testing {
-MinifiStatus DogController::enableImpl(api::core::ControllerServiceContext& ctx) {
+minifi_status DogController::enableImpl(api::core::ControllerServiceContext& ctx) {
this->has_jetpack_ = ctx.getProperty(HasJetpack) | minifi::utils::andThen(parsing::parseBool) |
minifi::utils::orThrow(fmt::format("Expected parsable bool from \"{}\"", HasJetpack.name));
diff --git a/extensions/stable-api-testing/AnimalControllerServices.h b/extensions/stable-api-testing/AnimalControllerServices.h
index 6a6e1e78e..d5b9c14bb 100644
--- a/extensions/stable-api-testing/AnimalControllerServices.h
+++ b/extensions/stable-api-testing/AnimalControllerServices.h
@@ -43,7 +43,7 @@ class DogController : public api::core::ControllerServiceImpl, public CanFlyCont
core::createProvidedInterface<NumberOfLegsControllerApi, DogController>()});
using ControllerServiceImpl::ControllerServiceImpl;
- MinifiStatus enableImpl(api::core::ControllerServiceContext& ctx) override;
+ minifi_status enableImpl(api::core::ControllerServiceContext& ctx) override;
uint8_t numberOfLegs() const override { return 4; }
bool canFly() const override { return has_jetpack_; }
@@ -61,7 +61,7 @@ class DuckController : public api::core::ControllerServiceImpl, public CanFlyCon
std::to_array<core::ProvidedInterface>({core::createProvidedInterface<CanFlyControllerApi, DuckController>(),
core::createProvidedInterface<NumberOfLegsControllerApi, DuckController>()});
using ControllerServiceImpl::ControllerServiceImpl;
- MinifiStatus enableImpl(api::core::ControllerServiceContext&) override { return MINIFI_STATUS_SUCCESS; }
+ minifi_status enableImpl(api::core::ControllerServiceContext&) override { return MINIFI_STATUS_SUCCESS; }
uint8_t numberOfLegs() const override { return 2; }
bool canFly() const override { return true; }
diff --git a/extensions/stable-api-testing/ExtensionInitializer.cpp b/extensions/stable-api-testing/ExtensionInitializer.cpp
index dae253407..c47a04d5c 100644
--- a/extensions/stable-api-testing/ExtensionInitializer.cpp
+++ b/extensions/stable-api-testing/ExtensionInitializer.cpp
@@ -25,14 +25,14 @@
namespace minifi = org::apache::nifi::minifi;
-CEXTENSIONAPI const uint32_t MinifiApiVersion = MINIFI_API_VERSION;
+CEXTENSIONAPI const uint32_t minifi_api_version = MINIFI_API_VERSION;
-CEXTENSIONAPI void MinifiInitExtension(MinifiExtensionContext* extension_context) {
- const MinifiExtensionDefinition extension_definition{.name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
+CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* extension_context) {
+ const minifi_extension_definition extension_definition{.name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
.version = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_VERSION)),
.deinit = nullptr,
.user_data = nullptr};
- auto* extension = MinifiRegisterExtension(extension_context, &extension_definition);
+ auto* extension = minifi_register_extension(extension_context, &extension_definition);
minifi::api::core::registerProcessors<minifi::api_testing::ZooProcessor>(extension);
minifi::api::core::registerControllerServices<minifi::api_testing::DogController, minifi::api_testing::DuckController>(extension);
}
diff --git a/extensions/stable-api-testing/ZooProcessor.cpp b/extensions/stable-api-testing/ZooProcessor.cpp
index 79314c75b..ed6fb7a9e 100644
--- a/extensions/stable-api-testing/ZooProcessor.cpp
+++ b/extensions/stable-api-testing/ZooProcessor.cpp
@@ -22,7 +22,7 @@
namespace org::apache::nifi::minifi::api_testing {
-MinifiStatus ZooProcessor::onTriggerImpl(api::core::ProcessContext& process_context, api::core::ProcessSession& process_session) {
+minifi_status ZooProcessor::onTriggerImpl(api::core::ProcessContext& process_context, api::core::ProcessSession& process_session) {
if (const auto can_fly_opaque = process_context.getControllerService(CanFlyService)) {
if (*can_fly_opaque) {
const auto can_fly_controller_name = process_context.getProperty(CanFlyService, nullptr) | utils::orThrow("Should be here");
@@ -40,7 +40,7 @@ MinifiStatus ZooProcessor::onTriggerImpl(api::core::ProcessContext& process_cont
return ProcessorImpl::onTriggerImpl(process_context, process_session);
}
-MinifiStatus ZooProcessor::onScheduleImpl(api::core::ProcessContext& process_context) {
+minifi_status ZooProcessor::onScheduleImpl(api::core::ProcessContext& process_context) {
return ProcessorImpl::onScheduleImpl(process_context);
}
} // namespace org::apache::nifi::minifi::api_testing
diff --git a/extensions/stable-api-testing/ZooProcessor.h b/extensions/stable-api-testing/ZooProcessor.h
index 7c028b7d8..5213f0a67 100644
--- a/extensions/stable-api-testing/ZooProcessor.h
+++ b/extensions/stable-api-testing/ZooProcessor.h
@@ -58,8 +58,8 @@ class ZooProcessor : public api::core::ProcessorImpl {
using ProcessorImpl::ProcessorImpl;
protected:
- MinifiStatus onTriggerImpl(api::core::ProcessContext&, api::core::ProcessSession&) override;
- MinifiStatus onScheduleImpl(api::core::ProcessContext&) override;
+ minifi_status onTriggerImpl(api::core::ProcessContext&, api::core::ProcessSession&) override;
+ minifi_status onScheduleImpl(api::core::ProcessContext&) override;
};
} // namespace org::apache::nifi::minifi::api_testing
diff --git a/extensions/standard-processors/processors/DefragmentText.cpp b/extensions/standard-processors/processors/DefragmentText.cpp
index dae7511de..a66eea44f 100644
--- a/extensions/standard-processors/processors/DefragmentText.cpp
+++ b/extensions/standard-processors/processors/DefragmentText.cpp
@@ -41,7 +41,7 @@ void DefragmentText::initialize() {
void DefragmentText::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) {
if (auto max_buffer_age = context.getProperty(MaxBufferAge) | utils::andThen(parsing::parseDuration<std::chrono::milliseconds>)) {
max_age_ = *max_buffer_age;
- setTriggerWhenEmpty(true);
+ context.setTriggerWhenEmpty(true);
logger_->log_trace("The Buffer maximum age is configured to be {}", max_age_);
}
if (auto max_buffer_size = context.getProperty(MaxBufferSize) | utils::andThen(parsing::parseDataSize)) {
diff --git a/extensions/standard-processors/processors/InvokeHTTP.cpp b/extensions/standard-processors/processors/InvokeHTTP.cpp
index 4716aa994..60aa97750 100644
--- a/extensions/standard-processors/processors/InvokeHTTP.cpp
+++ b/extensions/standard-processors/processors/InvokeHTTP.cpp
@@ -190,6 +190,7 @@ gsl::not_null<std::unique_ptr<http::HTTPClient>> InvokeHTTP::createHTTPClientFro
void InvokeHTTP::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) {
+ context.setTriggerWhenEmpty(true);
setupMembersFromProperties(context);
auto create_client = [this](const std::string& url) -> gsl::not_null<std::unique_ptr<minifi::http::HTTPClient>> {
diff --git a/extensions/standard-processors/processors/InvokeHTTP.h b/extensions/standard-processors/processors/InvokeHTTP.h
index 702804853..9d1ac2cbf 100644
--- a/extensions/standard-processors/processors/InvokeHTTP.h
+++ b/extensions/standard-processors/processors/InvokeHTTP.h
@@ -129,7 +129,6 @@ class InvokeHTTP : public core::ProcessorImpl {
explicit InvokeHTTP(core::ProcessorMetadata metadata)
: ProcessorImpl(metadata) {
- setTriggerWhenEmpty(true);
}
EXTENSIONAPI static constexpr const char* Description = "An HTTP client processor which can interact with a configurable HTTP Endpoint. "
diff --git a/extensions/standard-processors/processors/LogAttribute.h b/extensions/standard-processors/processors/LogAttribute.h
index 169de29ad..2c4e0f957 100644
--- a/extensions/standard-processors/processors/LogAttribute.h
+++ b/extensions/standard-processors/processors/LogAttribute.h
@@ -25,14 +25,14 @@
#include <vector>
#include "core/Core.h"
+#include "core/ProcessSession.h"
#include "core/ProcessorImpl.h"
+#include "core/PropertyDefinitionBuilder.h"
#include "minifi-cpp/core/ProcessContext.h"
-#include "core/ProcessSession.h"
#include "minifi-cpp/core/PropertyDefinition.h"
-#include "core/PropertyDefinitionBuilder.h"
#include "minifi-cpp/core/PropertyValidator.h"
#include "minifi-cpp/core/RelationshipDefinition.h"
-#include "core/logging/LoggerFactory.h"
+#include "minifi-cpp/core/logging/AdvancedLogger.h"
#include "minifi-cpp/utils/Export.h"
namespace org::apache::nifi::minifi::processors {
@@ -41,7 +41,7 @@ class LogAttribute : public core::ProcessorImpl {
public:
explicit LogAttribute(core::ProcessorMetadata metadata)
: ProcessorImpl(metadata) {
- logger_->set_max_log_size(-1);
+ gsl_Assert(core::logging::AdvancedLogger::setMaxLogSize(logger_.get(), -1));
}
~LogAttribute() override = default;
diff --git a/extensions/standard-processors/tests/unit/ProcessorTests.cpp b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
index aa0e048ca..8f52fc80e 100644
--- a/extensions/standard-processors/tests/unit/ProcessorTests.cpp
+++ b/extensions/standard-processors/tests/unit/ProcessorTests.cpp
@@ -545,7 +545,8 @@ void testRPGBypass(const std::string &host, const std::string &port, bool has_er
std::shared_ptr<core::Repository> test_repo = std::make_shared<TestRepository>();
std::shared_ptr<TestRepository> repo = std::dynamic_pointer_cast<TestRepository>(test_repo);
- auto rpg = minifi::test::utils::make_custom_processor<minifi::RemoteProcessGroupPort>("rpg", "http://localhost:8989/nifi", configuration, utils::IdGenerator::getIdGenerator()->generate());
+ auto rpg = minifi::test::utils::make_custom_processor<minifi::RemoteProcessGroupPort>("rpg", "http://localhost:8989/nifi", configuration, utils::IdGenerator::getIdGenerator()->generate(),
+ minifi::sitetosite::TransferDirection::SEND);
rpg->initialize();
REQUIRE(rpg->setProperty(minifi::RemoteProcessGroupPort::hostName.name, host));
REQUIRE(rpg->setProperty(minifi::RemoteProcessGroupPort::port.name, port));
diff --git a/libminifi/include/RemoteProcessGroupPort.h b/libminifi/include/RemoteProcessGroupPort.h
index a1572b939..b1da2d90d 100644
--- a/libminifi/include/RemoteProcessGroupPort.h
+++ b/libminifi/include/RemoteProcessGroupPort.h
@@ -19,25 +19,25 @@
*/
#pragma once
+#include <memory>
+#include <mutex>
+#include <stack>
#include <string>
#include <utility>
#include <vector>
-#include <mutex>
-#include <memory>
-#include <stack>
-#include "http/BaseHTTPClient.h"
#include "concurrentqueue.h"
-#include "core/ProcessorImpl.h"
+#include "core/ClassLoader.h"
#include "core/ProcessSession.h"
-#include "minifi-cpp/core/PropertyDefinition.h"
+#include "core/ProcessorImpl.h"
#include "core/PropertyDefinitionBuilder.h"
-#include "minifi-cpp/core/RelationshipDefinition.h"
-#include "sitetosite/SiteToSiteClient.h"
-#include "minifi-cpp/controllers/SSLContextServiceInterface.h"
#include "core/logging/LoggerFactory.h"
+#include "http/BaseHTTPClient.h"
+#include "minifi-cpp/controllers/SSLContextServiceInterface.h"
+#include "minifi-cpp/core/PropertyDefinition.h"
+#include "minifi-cpp/core/RelationshipDefinition.h"
#include "minifi-cpp/utils/Export.h"
-#include "core/ClassLoader.h"
+#include "sitetosite/SiteToSiteClient.h"
namespace org::apache::nifi::minifi {
@@ -49,10 +49,13 @@ struct RPG {
class RemoteProcessGroupPort : public core::ProcessorImpl {
public:
- RemoteProcessGroupPort(std::string_view name, std::string url, std::shared_ptr<Configure> configure, const utils::Identifier &uuid, std::shared_ptr<core::logging::Logger> logger = nullptr)
- : core::ProcessorImpl({.uuid = uuid, .name = std::string{name}, .logger = logger ? logger : core::logging::LoggerFactory<RemoteProcessGroupPort>::getLogger(uuid)}),
+ RemoteProcessGroupPort(std::string_view name, std::string url, std::shared_ptr<Configure> configure, const utils::Identifier& uuid,
+ sitetosite::TransferDirection direction, std::shared_ptr<core::logging::Logger> logger = nullptr)
+ : core::ProcessorImpl({.uuid = uuid,
+ .name = std::string{name},
+ .logger = logger ? logger : core::logging::LoggerFactory<RemoteProcessGroupPort>::getLogger(uuid)}),
configure_(std::move(configure)),
- direction_(sitetosite::TransferDirection::SEND),
+ direction_(direction),
transmitting_(false),
protocol_uuid_(uuid),
client_type_(sitetosite::ClientType::RAW),
@@ -62,32 +65,24 @@ class RemoteProcessGroupPort : public core::ProcessorImpl {
}
virtual ~RemoteProcessGroupPort() = default;
- MINIFIAPI static constexpr auto hostName = core::PropertyDefinitionBuilder<>::createProperty("Host Name")
- .withDescription("Remote Host Name.")
- .build();
- MINIFIAPI static constexpr auto SSLContext = core::PropertyDefinitionBuilder<>::createProperty("SSL Context Service")
- .withDescription("The SSL Context Service used to provide client certificate information for TLS/SSL (https) connections.")
- .build();
- MINIFIAPI static constexpr auto port = core::PropertyDefinitionBuilder<>::createProperty("Port")
- .withDescription("Remote Port")
- .build();
- MINIFIAPI static constexpr auto portUUID = core::PropertyDefinitionBuilder<>::createProperty("Port UUID")
- .withDescription("Specifies remote NiFi Port UUID.")
- .build();
- MINIFIAPI static constexpr auto idleTimeout = core::PropertyDefinitionBuilder<>::createProperty("Idle Timeout")
- .withDescription("Max idle time for remote service")
- .isRequired(true)
- .withValidator(core::StandardPropertyValidators::TIME_PERIOD_VALIDATOR)
- .withDefaultValue("15 s")
- .build();
-
- MINIFIAPI static constexpr auto Properties = std::to_array<core::PropertyReference>({
- hostName,
- SSLContext,
- port,
- portUUID,
- idleTimeout
- });
+ MINIFIAPI static constexpr auto hostName =
+ core::PropertyDefinitionBuilder<>::createProperty("Host Name").withDescription("Remote Host Name.").build();
+ MINIFIAPI static constexpr auto SSLContext =
+ core::PropertyDefinitionBuilder<>::createProperty("SSL Context Service")
+ .withDescription("The SSL Context Service used to provide client certificate information for TLS/SSL (https) connections.")
+ .build();
+ MINIFIAPI static constexpr auto port = core::PropertyDefinitionBuilder<>::createProperty("Port").withDescription("Remote Port").build();
+ MINIFIAPI static constexpr auto portUUID =
+ core::PropertyDefinitionBuilder<>::createProperty("Port UUID").withDescription("Specifies remote NiFi Port UUID.").build();
+ MINIFIAPI static constexpr auto idleTimeout =
+ core::PropertyDefinitionBuilder<>::createProperty("Idle Timeout")
+ .withDescription("Max idle time for remote service")
+ .isRequired(true)
+ .withValidator(core::StandardPropertyValidators::TIME_PERIOD_VALIDATOR)
+ .withDefaultValue("15 s")
+ .build();
+
+ MINIFIAPI static constexpr auto Properties = std::to_array<core::PropertyReference>({hostName, SSLContext, port, portUUID, idleTimeout});
MINIFIAPI static constexpr auto DefaultRelationship = core::RelationshipDefinition{"undefined", ""};
MINIFIAPI static constexpr auto Relationships = std::array{DefaultRelationship};
@@ -103,15 +98,6 @@ class RemoteProcessGroupPort : public core::ProcessorImpl {
void onTrigger(core::ProcessContext& context, core::ProcessSession& session) override;
void initialize() override;
- void setDirection(sitetosite::TransferDirection direction) {
- direction_ = direction;
- if (direction_ == sitetosite::TransferDirection::RECEIVE) {
- setTriggerWhenEmpty(true);
- } else {
- setTriggerWhenEmpty(false);
- }
- }
-
void setTimeout(std::chrono::milliseconds timeout) {
timeout_ = timeout;
}
@@ -124,7 +110,7 @@ class RemoteProcessGroupPort : public core::ProcessorImpl {
transmitting_ = val;
}
- void setInterface(const std::string &ifc) {
+ void setInterface(const std::string& ifc) {
local_network_interface_ = ifc;
}
@@ -134,7 +120,7 @@ class RemoteProcessGroupPort : public core::ProcessorImpl {
return nifi_instances_;
}
- void setHTTPProxy(const http::HTTPProxy &proxy) {
+ void setHTTPProxy(const http::HTTPProxy& proxy) {
proxy_ = proxy;
}
@@ -188,7 +174,7 @@ class RemoteProcessGroupPort : public core::ProcessorImpl {
moodycamel::ConcurrentQueue<std::unique_ptr<sitetosite::SiteToSiteClient>> available_protocols_;
std::shared_ptr<Configure> configure_;
- sitetosite::TransferDirection direction_;
+ const sitetosite::TransferDirection direction_;
std::atomic<bool> transmitting_;
std::optional<std::chrono::milliseconds> timeout_;
std::string local_network_interface_;
diff --git a/libminifi/include/core/ProcessContextImpl.h b/libminifi/include/core/ProcessContextImpl.h
index 9466e69df..7bbf3ccf5 100644
--- a/libminifi/include/core/ProcessContextImpl.h
+++ b/libminifi/include/core/ProcessContextImpl.h
@@ -193,6 +193,7 @@ class ProcessContextImpl : public core::VariableRegistryImpl, public virtual Pro
}
gsl::not_null<Configure*> getConfiguration() const override { return gsl::make_not_null(configure_.get()); }
+ void setTriggerWhenEmpty(bool trigger_when_empty) override;
private:
std::shared_ptr<logging::Logger> logger_;
diff --git a/libminifi/include/core/Processor.h b/libminifi/include/core/Processor.h
index 5cb2ed9ce..4eff8d8b2 100644
--- a/libminifi/include/core/Processor.h
+++ b/libminifi/include/core/Processor.h
@@ -83,6 +83,7 @@ class Processor : public ConnectableImpl, public ConfigurableComponentImpl, publ
void setMaxConcurrentTasks(uint8_t tasks) override;
bool isSingleThreaded() const;
std::string getProcessorType() const;
+ void setTriggerWhenEmpty(bool trigger_when_empty);
bool getTriggerWhenEmpty() const;
uint8_t getActiveTasks() const;
void incrementActiveTasks();
@@ -147,6 +148,7 @@ class Processor : public ConnectableImpl, public ConfigurableComponentImpl, publ
std::atomic<std::chrono::steady_clock::duration> yield_period_;
std::atomic<uint8_t> active_tasks_;
+ std::atomic<bool> trigger_when_empty_;
std::string cron_period_;
diff --git a/libminifi/include/core/extension/Extension.h b/libminifi/include/core/extension/Extension.h
index 060c889f0..11e91db62 100644
--- a/libminifi/include/core/extension/Extension.h
+++ b/libminifi/include/core/extension/Extension.h
@@ -24,10 +24,10 @@
#include <optional>
#include <string>
-#include "minifi-c/minifi-c.h"
+#include "ApiVersion.h"
+#include "minifi-api.h"
#include "minifi-cpp/core/logging/Logger.h"
#include "minifi-cpp/properties/Configure.h"
-#include "ApiVersion.h"
namespace org::apache::nifi::minifi::core::extension {
diff --git a/libminifi/include/core/logging/LoggerBase.h b/libminifi/include/core/logging/LoggerBase.h
index 175867f49..a2cad65fd 100644
--- a/libminifi/include/core/logging/LoggerBase.h
+++ b/libminifi/include/core/logging/LoggerBase.h
@@ -17,26 +17,18 @@
*/
#pragma once
-#include <string>
-#include <mutex>
#include <memory>
+#include <mutex>
#include <optional>
-#include <sstream>
-#include <utility>
-#include <iostream>
-#include <vector>
-#include <algorithm>
+#include <string>
+#include "fmt/std.h"
+#include "minifi-cpp/core/logging/AdvancedLogger.h"
+#include "minifi-cpp/core/logging/Logger.h"
#include "spdlog/common.h"
#include "spdlog/logger.h"
-#include "minifi-cpp/utils/gsl.h"
#include "utils/ConfigurationUtils.h"
#include "utils/Enum.h"
-#include "utils/GeneralUtils.h"
-#include "fmt/chrono.h"
-#include "fmt/std.h"
-#include "fmt/ostream.h"
-#include "minifi-cpp/core/logging/Logger.h"
namespace org::apache::nifi::minifi::core::logging {
@@ -111,7 +103,7 @@ inline LOG_LEVEL mapStringToLogLevel(const std::string& level_str) {
throw std::invalid_argument(fmt::format("Invalid LOG_LEVEL {}", level_str));
}
-class LoggerBase : public Logger {
+class LoggerBase : public AdvancedLogger {
public:
LoggerBase(LoggerBase const&) = delete;
LoggerBase& operator=(LoggerBase const&) = delete;
@@ -119,11 +111,11 @@ class LoggerBase : public Logger {
void set_max_log_size(int size) override {
max_log_size_ = size;
}
+ LOG_LEVEL level() const override;
virtual std::optional<std::string> get_id() = 0;
bool should_log(LOG_LEVEL level) override;
void log_string(LOG_LEVEL level, std::string str) override;
- LOG_LEVEL level() const override;
void setLogCallback(const std::function<void(LOG_LEVEL level, const std::string&)>& callback);
private:
diff --git a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h
index 182e6f15d..49beaae12 100644
--- a/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h
+++ b/libminifi/include/core/reporting/SiteToSiteProvenanceReportingTask.h
@@ -35,8 +35,7 @@ class SiteToSiteProvenanceReportingTask : public minifi::RemoteProcessGroupPort
public:
explicit SiteToSiteProvenanceReportingTask(std::shared_ptr<Configure> configure)
: minifi::RemoteProcessGroupPort(ReportTaskName, "", std::move(configure),
- utils::IdGenerator::getIdGenerator()->generate(), logging::LoggerFactory<SiteToSiteProvenanceReportingTask>::getLogger()) {
- this->setTriggerWhenEmpty(true);
+ utils::IdGenerator::getIdGenerator()->generate(), sitetosite::TransferDirection::SEND, logging::LoggerFactory<SiteToSiteProvenanceReportingTask>::getLogger()) {
batch_size_ = 100;
}
diff --git a/libminifi/include/utils/CControllerService.h b/libminifi/include/utils/CControllerService.h
index 6e000548d..54649969d 100644
--- a/libminifi/include/utils/CControllerService.h
+++ b/libminifi/include/utils/CControllerService.h
@@ -20,7 +20,7 @@
#include <string>
#include <vector>
-#include "minifi-c/minifi-c.h"
+#include "minifi-api.h"
#include "minifi-cpp/Exception.h"
#include "minifi-cpp/core/ControllerServiceMetadata.h"
#include "minifi-cpp/core/Property.h"
@@ -37,7 +37,7 @@ namespace org::apache::nifi::minifi::utils {
struct CControllerServiceClassDescription {
std::string full_name;
std::vector<core::Property> class_properties;
- MinifiControllerServiceCallbacks callbacks;
+ minifi_controller_service_callbacks callbacks;
};
class CControllerService final : public core::controller::ControllerServiceApi, public core::controller::ControllerServiceHandle {
@@ -45,11 +45,11 @@ class CControllerService final : public core::controller::ControllerServiceApi,
CControllerService(CControllerServiceClassDescription class_description, core::ControllerServiceMetadata metadata)
: class_description_(std::move(class_description)),
metadata_(std::move(metadata)) {
- MinifiControllerServiceMetadata c_metadata;
+ minifi_controller_service_metadata c_metadata;
auto uuid_str = metadata_.uuid.to_string();
- c_metadata.uuid = MinifiStringView{.data = uuid_str.data(), .length = uuid_str.length()};
- c_metadata.name = MinifiStringView{.data = metadata_.name.data(), .length = metadata_.name.length()};
- c_metadata.logger = reinterpret_cast<MinifiLogger*>(&metadata_.logger);
+ c_metadata.uuid = minifi_string_view{.data = uuid_str.data(), .length = uuid_str.length()};
+ c_metadata.name = minifi_string_view{.data = metadata_.name.data(), .length = metadata_.name.length()};
+ c_metadata.logger = reinterpret_cast<minifi_logger*>(&metadata_.logger);
impl_ = class_description_.callbacks.create(c_metadata);
}
CControllerService(CControllerServiceClassDescription class_description, core::ControllerServiceMetadata metadata, gsl::owner<void*> impl)
@@ -75,7 +75,7 @@ class CControllerService final : public core::controller::ControllerServiceApi,
void onEnable(core::controller::ControllerServiceContext& controller_service_context,
const std::shared_ptr<Configure>&,
const std::vector<std::shared_ptr<core::controller::ControllerServiceHandle>>&) override {
- const auto enable_status = class_description_.callbacks.enable(impl_, reinterpret_cast<MinifiControllerServiceContext*>(&controller_service_context));
+ const auto enable_status = class_description_.callbacks.enable(impl_, reinterpret_cast<minifi_controller_service_context*>(&controller_service_context));
if (enable_status != MINIFI_STATUS_SUCCESS) {
throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Could not enable controller service");
}
@@ -111,7 +111,7 @@ class CControllerService final : public core::controller::ControllerServiceApi,
minifi::core::ControllerServiceMetadata metadata_;
};
-void useCControllerServiceClassDescription(const MinifiControllerServiceClassDefinition& class_description,
+void useCControllerServiceClassDescription(std::string_view bundle_name, const minifi_controller_service_definition& class_description,
const std::function<void(const ClassDescription&, CControllerServiceClassDescription)>& fn);
} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/include/utils/CProcessor.h b/libminifi/include/utils/CProcessor.h
index cfa3ea2d6..329b5ce9c 100644
--- a/libminifi/include/utils/CProcessor.h
+++ b/libminifi/include/utils/CProcessor.h
@@ -20,10 +20,10 @@
#include <string>
#include <utility>
-#include "minifi-cpp/core/Annotation.h"
#include "core/ProcessorMetrics.h"
-#include "minifi-c/minifi-c.h"
+#include "minifi-api.h"
#include "minifi-cpp/agent/agent_docs.h"
+#include "minifi-cpp/core/Annotation.h"
#include "minifi-cpp/core/ProcessContext.h"
#include "minifi-cpp/core/ProcessorApi.h"
#include "minifi-cpp/core/ProcessorDescriptor.h"
@@ -56,7 +56,7 @@ struct CProcessorClassDescription {
minifi::core::annotation::Input input_requirement;
bool is_single_threaded;
- MinifiProcessorCallbacks callbacks;
+ minifi_processor_callbacks callbacks;
};
class CProcessor : public minifi::core::ProcessorApi {
@@ -65,11 +65,11 @@ class CProcessor : public minifi::core::ProcessorApi {
: class_description_(std::move(class_description)),
metadata_(std::move(metadata)),
metrics_extension_(std::make_shared<CProcessorMetricsWrapper>(*this)) {
- MinifiProcessorMetadata c_metadata;
+ minifi_processor_metadata c_metadata;
const auto uuid_str = metadata.uuid.to_string();
- c_metadata.uuid = MinifiStringView{.data = uuid_str.data(), .length = uuid_str.length()};
- c_metadata.name = MinifiStringView{.data = metadata_.name.data(), .length = metadata_.name.length()};
- c_metadata.logger = reinterpret_cast<MinifiLogger*>(&metadata_.logger);
+ c_metadata.uuid = minifi_string_view{.data = uuid_str.data(), .length = uuid_str.length()};
+ c_metadata.name = minifi_string_view{.data = metadata_.name.data(), .length = metadata_.name.length()};
+ c_metadata.logger = reinterpret_cast<minifi_logger*>(&metadata_.logger);
impl_ = class_description_.callbacks.create(c_metadata);
}
CProcessor(CProcessorClassDescription class_description, minifi::core::ProcessorMetadata metadata, gsl::owner<void*> impl)
@@ -114,13 +114,9 @@ class CProcessor : public minifi::core::ProcessorApi {
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));
+ auto status = class_description_.callbacks.trigger(impl_, reinterpret_cast<minifi_process_context*>(&process_context), reinterpret_cast<minifi_process_session*>(&process_session));
if (status == MINIFI_STATUS_PROCESSOR_YIELD) {
process_context.yield();
return;
@@ -132,18 +128,18 @@ class CProcessor : public minifi::core::ProcessorApi {
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));
+ auto status = class_description_.callbacks.schedule(impl_, reinterpret_cast<minifi_process_context*>(&process_context));
if (status != MINIFI_STATUS_SUCCESS) {
throw minifi::Exception(minifi::ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Error while scheduling processor");
}
}
void onUnSchedule() override {
- class_description_.callbacks.onUnSchedule(impl_);
+ class_description_.callbacks.unschedule(impl_);
}
void notifyStop() override {
- class_description_.callbacks.onUnSchedule(impl_);
+ class_description_.callbacks.unschedule(impl_);
}
minifi::core::annotation::Input getInputRequirement() const override {
@@ -166,13 +162,21 @@ class CProcessor : public minifi::core::ProcessorApi {
return metadata_.uuid;
}
+ void reportMetrics(std::vector<minifi::state::PublishedMetric> metrics) {
+ std::lock_guard guard(metrics_mutex_);
+ reported_metrics_ = std::move(metrics);
+ }
+
private:
CProcessorClassDescription class_description_;
gsl::owner<void*> impl_;
minifi::core::ProcessorMetadata metadata_;
gsl::not_null<std::shared_ptr<minifi::core::ProcessorMetricsExtension>> metrics_extension_;
+
+ mutable std::mutex metrics_mutex_;
+ std::vector<minifi::state::PublishedMetric> reported_metrics_;
};
-void useCProcessorClassDescription(const MinifiProcessorClassDefinition& class_description, const std::function<void(ClassDescription, CProcessorClassDescription)>& fn);
+void useCProcessorClassDescription(const minifi_processor_definition& class_description, const std::function<void(ClassDescription, CProcessorClassDescription)>& fn);
} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/src/RemoteProcessGroupPort.cpp b/libminifi/src/RemoteProcessGroupPort.cpp
index 500fa4d4c..684c00aef 100644
--- a/libminifi/src/RemoteProcessGroupPort.cpp
+++ b/libminifi/src/RemoteProcessGroupPort.cpp
@@ -128,6 +128,12 @@ void RemoteProcessGroupPort::initialize() {
}
void RemoteProcessGroupPort::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) {
+ if (direction_ == sitetosite::TransferDirection::RECEIVE) {
+ context.setTriggerWhenEmpty(true);
+ } else {
+ context.setTriggerWhenEmpty(false);
+ }
+
if (auto protocol_uuid = context.getProperty(portUUID)) {
protocol_uuid_ = *protocol_uuid;
}
diff --git a/libminifi/src/c2/HeartbeatLogger.cpp b/libminifi/src/c2/HeartbeatLogger.cpp
index f4380ba86..5ec1350c0 100644
--- a/libminifi/src/c2/HeartbeatLogger.cpp
+++ b/libminifi/src/c2/HeartbeatLogger.cpp
@@ -16,13 +16,15 @@
*/
#include "c2/HeartbeatLogger.h"
+
#include "core/Resource.h"
+#include "minifi-cpp/core/logging/AdvancedLogger.h"
namespace org::apache::nifi::minifi::c2 {
HeartbeatLogger::HeartbeatLogger(std::string_view name, const utils::Identifier& id)
: HeartbeatReporter(name, id) {
- logger_->set_max_log_size(-1); // log however huge the heartbeat is
+ gsl_Assert(core::logging::AdvancedLogger::setMaxLogSize(logger_.get(), -1));
}
int16_t HeartbeatLogger::heartbeat(const C2Payload &heartbeat) {
diff --git a/libminifi/src/core/ProcessContextImpl.cpp b/libminifi/src/core/ProcessContextImpl.cpp
index 841b74ca5..c732f32bc 100644
--- a/libminifi/src/core/ProcessContextImpl.cpp
+++ b/libminifi/src/core/ProcessContextImpl.cpp
@@ -178,6 +178,9 @@ StateManager* ProcessContextImpl::getStateManager() {
void ProcessContextImpl::setSessionStateManager(std::unique_ptr<StateManager> state_manager) {
session_state_manager_ = std::move(state_manager);
}
+void ProcessContextImpl::setTriggerWhenEmpty(bool trigger_when_empty) {
+ processor_.setTriggerWhenEmpty(trigger_when_empty);
+}
bool ProcessContextImpl::hasIncomingConnections() const {
return getProcessor().hasIncomingConnections();
diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp
index bd59eeaaa..b3bf3f16b 100644
--- a/libminifi/src/core/ProcessSession.cpp
+++ b/libminifi/src/core/ProcessSession.cpp
@@ -34,7 +34,7 @@
#include "core/Processor.h"
#include "io/StreamPipe.h"
#include "io/StreamSlice.h"
-#include "minifi-c/minifi-c.h"
+#include "minifi-api.h"
#include "minifi-cpp/utils/gsl.h"
/* This implementation is only for native Windows systems. */
diff --git a/libminifi/src/core/Processor.cpp b/libminifi/src/core/Processor.cpp
index 72fca50a6..108769c8b 100644
--- a/libminifi/src/core/Processor.cpp
+++ b/libminifi/src/core/Processor.cpp
@@ -59,6 +59,7 @@ Processor::Processor(std::string_view name, std::unique_ptr<ProcessorApi> impl)
run_duration_(DEFAULT_RUN_DURATION),
yield_period_(DEFAULT_YIELD_PERIOD_SECONDS),
active_tasks_(0),
+ trigger_when_empty_(false),
logger_(logging::LoggerFactory<Processor>::getLogger(uuid_)),
metrics_(gsl::make_not_null(std::make_shared<ProcessorMetrics>(*this))),
impl_(std::move(impl)) {
@@ -78,6 +79,7 @@ Processor::Processor(std::string_view name, const utils::Identifier& uuid, std::
run_duration_(DEFAULT_RUN_DURATION),
yield_period_(DEFAULT_YIELD_PERIOD_SECONDS),
active_tasks_(0),
+ trigger_when_empty_(false),
logger_(logging::LoggerFactory<Processor>::getLogger(uuid_)),
metrics_(gsl::make_not_null(std::make_shared<ProcessorMetrics>(*this))),
impl_(std::move(impl)) {
@@ -448,8 +450,12 @@ std::string Processor::getProcessorType() const {
return impl_->getProcessorType();
}
+void Processor::setTriggerWhenEmpty(bool trigger_when_empty) {
+ trigger_when_empty_ = trigger_when_empty;
+}
+
bool Processor::getTriggerWhenEmpty() const {
- return impl_->getTriggerWhenEmpty();
+ return trigger_when_empty_;
}
uint8_t Processor::getActiveTasks() const {
diff --git a/libminifi/src/core/extension/ApiVersion.cpp b/libminifi/src/core/extension/ApiVersion.cpp
index bf5259420..e7930fc11 100644
--- a/libminifi/src/core/extension/ApiVersion.cpp
+++ b/libminifi/src/core/extension/ApiVersion.cpp
@@ -16,7 +16,8 @@
*/
#include "core/extension/ApiVersion.h"
-#include "minifi-c/minifi-c.h"
+
+#include "minifi-api.h"
namespace org::apache::nifi::minifi::core::extension {
diff --git a/libminifi/src/core/extension/Extension.cpp b/libminifi/src/core/extension/Extension.cpp
index 1d210d3b6..bb76ae3f2 100644
--- a/libminifi/src/core/extension/Extension.cpp
+++ b/libminifi/src/core/extension/Extension.cpp
@@ -33,13 +33,13 @@
#define RTLD_LOCAL (1 << 2)
#endif
+#include "core/ClassLoader.h"
#include "core/extension/Extension.h"
-#include "utils/GeneralUtils.h"
#include "core/logging/LoggerFactory.h"
-#include "minifi-c/minifi-c.h"
+#include "minifi-api.h"
#include "minifi-cpp/agent/agent_docs.h"
+#include "utils/GeneralUtils.h"
#include "utils/RegexUtils.h"
-#include "core/ClassLoader.h"
namespace org::apache::nifi::minifi::core::extension {
@@ -61,17 +61,17 @@ bool Extension::load(bool global) {
return false;
}
logger_->log_trace("Dlopen succeeded for extension '{}' at '{}'", library_name_, library_path_);
- if (findSymbol("MinifiInitCppExtension")) {
+ if (findSymbol("minifi_init_cpp_extension")) {
logger_->log_trace("Loaded cpp extension '{}' at '{}'", library_name_, library_path_);
return true;
}
- if (!findSymbol("MinifiInitExtension")) {
+ if (!findSymbol("minifi_init_extension")) {
logger_->log_error("Failed to load as c extension '{}' at '{}': No initializer found", library_name_, library_path_);
return false;
}
- auto api_version_ptr = reinterpret_cast<const uint32_t*>(findSymbol("MinifiApiVersion"));
+ auto api_version_ptr = reinterpret_cast<const uint32_t*>(findSymbol("minifi_api_version"));
if (!api_version_ptr) {
- logger_->log_error("Failed to load c extension '{}' at '{}': No MinifiApiVersion symbol found", library_name_, library_path_);
+ logger_->log_error("Failed to load c extension '{}' at '{}': No minifi_api_version symbol found", library_name_, library_path_);
return false;
}
api_version_ = *api_version_ptr;
@@ -147,9 +147,9 @@ Extension::~Extension() {
bool Extension::initialize(const std::shared_ptr<minifi::Configure>& configure) {
logger_->log_trace("Initializing extension '{}'", library_name_);
- void* init_symbol_ptr = findSymbol("MinifiInitCppExtension");
+ void* init_symbol_ptr = findSymbol("minifi_init_cpp_extension");
if (!init_symbol_ptr) {
- init_symbol_ptr = findSymbol("MinifiInitExtension");
+ init_symbol_ptr = findSymbol("minifi_init_extension");
}
if (!init_symbol_ptr) {
logger_->log_error("No initializer for '{}'", library_name_);
@@ -157,7 +157,7 @@ bool Extension::initialize(const std::shared_ptr<minifi::Configure>& configure)
}
logger_->log_debug("Found initializer for '{}'", library_name_);
- auto init_fn = reinterpret_cast<void(*)(MinifiExtensionContext*)>(init_symbol_ptr);
+ auto init_fn = reinterpret_cast<void(*)(minifi_extension_context*)>(init_symbol_ptr);
Context extension_context{
.config = configure,
.create = [&] (Info info) -> Extension* {
@@ -168,7 +168,7 @@ bool Extension::initialize(const std::shared_ptr<minifi::Configure>& configure)
return this;
}
};
- init_fn(reinterpret_cast<MinifiExtensionContext*>(&extension_context));
+ init_fn(reinterpret_cast<minifi_extension_context*>(&extension_context));
if (!info_) {
logger_->log_error("Failed to initialize extension '{}'", library_name_);
return false;
diff --git a/libminifi/src/core/flow/StructuredConfiguration.cpp b/libminifi/src/core/flow/StructuredConfiguration.cpp
index e64b5a34d..04c123141 100644
--- a/libminifi/src/core/flow/StructuredConfiguration.cpp
+++ b/libminifi/src/core/flow/StructuredConfiguration.cpp
@@ -739,10 +739,9 @@ void StructuredConfiguration::parseRPGPort(const Node& port_node, core::ProcessG
uuid = portId;
auto port_impl = std::make_unique<minifi::RemoteProcessGroupPort>(
- nameStr, parent->getURL(), this->configuration_, uuid);
+ nameStr, parent->getURL(), this->configuration_, uuid, direction);
auto* port = port_impl.get();
auto port_wrapper = std::make_unique<core::Processor>(nameStr, uuid, std::move(port_impl));
- port->setDirection(direction);
port->setTimeout(std::chrono::milliseconds(parent->getTimeout()));
port->setTransmitting(true);
port_wrapper->setYieldPeriodMsec(parent->getYieldPeriodMsec());
diff --git a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
index 363fddede..c967e1aea 100644
--- a/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
+++ b/libminifi/src/core/reporting/SiteToSiteProvenanceReportingTask.cpp
@@ -151,7 +151,8 @@ void SiteToSiteProvenanceReportingTask::getJsonReport(core::ProcessContext&, cor
report = buffer.GetString();
}
-void SiteToSiteProvenanceReportingTask::onSchedule(core::ProcessContext&, core::ProcessSessionFactory&) {
+void SiteToSiteProvenanceReportingTask::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) {
+ context.setTriggerWhenEmpty(true);
}
void SiteToSiteProvenanceReportingTask::onTrigger(core::ProcessContext& context, core::ProcessSession& session) {
diff --git a/libminifi/src/core/state/nodes/AgentInformation.cpp b/libminifi/src/core/state/nodes/AgentInformation.cpp
index 7a81b7f39..ec4183c64 100644
--- a/libminifi/src/core/state/nodes/AgentInformation.cpp
+++ b/libminifi/src/core/state/nodes/AgentInformation.cpp
@@ -21,7 +21,7 @@
#include "core/Resource.h"
#include "core/state/nodes/SchedulingNodes.h"
#include "core/state/nodes/SupportedOperations.h"
-#include "minifi-c/minifi-c.h"
+#include "minifi-api.h"
#include "minifi-cpp/agent/agent_version.h"
#include "minifi-cpp/controllers/AttributeProviderService.h"
#include "minifi-cpp/controllers/ProxyConfigurationServiceInterface.h"
diff --git a/libminifi/src/minifi-c.cpp b/libminifi/src/minifi-api.cpp
similarity index 68%
rename from libminifi/src/minifi-c.cpp
rename to libminifi/src/minifi-api.cpp
index 68617d522..1b484341c 100644
--- a/libminifi/src/minifi-c.cpp
+++ b/libminifi/src/minifi-api.cpp
@@ -15,17 +15,16 @@
* limitations under the License.
*/
-#include "minifi-c/minifi-c.h"
-
#include <memory>
#include <vector>
#include "agent/agent_docs.h"
+#include "core/Processor.h"
#include "core/ProcessorMetrics.h"
#include "core/extension/ExtensionManager.h"
-#include "minifi-cpp/Exception.h"
-#include "minifi-cpp/controllers/SSLContextServiceInterface.h"
+#include "minifi-api.h"
#include "minifi-cpp/controllers/ProxyConfigurationServiceInterface.h"
+#include "minifi-cpp/controllers/SSLContextServiceInterface.h"
#include "minifi-cpp/core/Annotation.h"
#include "minifi-cpp/core/ClassLoader.h"
#include "minifi-cpp/core/ProcessContext.h"
@@ -45,19 +44,19 @@ namespace minifi = org::apache::nifi::minifi;
namespace {
-MinifiStringView minifiStringView(const std::string_view s) {
- return MinifiStringView{.data = s.data(), .length = s.size()};
+minifi_string_view minifiStringView(const std::string_view s) {
+ return minifi_string_view{.data = s.data(), .length = s.size()};
}
-std::string toString(MinifiStringView sv) {
+std::string toString(minifi_string_view sv) {
return {sv.data, sv.length};
}
-std::string_view toStringView(MinifiStringView sv) {
+std::string_view toStringView(minifi_string_view sv) {
return {sv.data, sv.length};
}
-minifi::core::annotation::Input toInputRequirement(MinifiInputRequirement req) {
+minifi::core::annotation::Input toInputRequirement(minifi_input_requirement 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;
@@ -66,7 +65,7 @@ minifi::core::annotation::Input toInputRequirement(MinifiInputRequirement req) {
gsl_FailFast();
}
-minifi::core::logging::LOG_LEVEL toLogLevel(MinifiLogLevel lvl) {
+minifi::core::logging::LOG_LEVEL toLogLevel(minifi_log_level lvl) {
switch (lvl) {
case MINIFI_LOG_LEVEL_TRACE: return minifi::core::logging::trace;
case MINIFI_LOG_LEVEL_DEBUG: return minifi::core::logging::debug;
@@ -79,7 +78,7 @@ minifi::core::logging::LOG_LEVEL toLogLevel(MinifiLogLevel lvl) {
gsl_FailFast();
}
-gsl::not_null<const minifi::core::PropertyValidator*> toPropertyValidator(MinifiValidator validator) {
+gsl::not_null<const minifi::core::PropertyValidator*> toPropertyValidator(minifi_validator validator) {
switch (validator) {
case MINIFI_VALIDATOR_ALWAYS_VALID: return gsl::make_not_null(&minifi::core::StandardPropertyValidators::ALWAYS_VALID_VALIDATOR);
case MINIFI_VALIDATOR_NON_BLANK: return gsl::make_not_null(&minifi::core::StandardPropertyValidators::NON_BLANK_VALIDATOR);
@@ -93,7 +92,7 @@ gsl::not_null<const minifi::core::PropertyValidator*> toPropertyValidator(Minifi
gsl_FailFast();
}
-minifi::core::Property createProperty(const MinifiPropertyDefinition* property_description) {
+minifi::core::Property createProperty(const minifi_property_definition* property_description) {
gsl_Expects(property_description);
std::vector<std::string_view> allowed_values;
allowed_values.reserve(property_description->allowed_values_count);
@@ -101,8 +100,8 @@ minifi::core::Property createProperty(const MinifiPropertyDefinition* property_d
allowed_values.push_back(toStringView(property_description->allowed_values_ptr[i]));
}
std::vector<std::string_view> allowed_types;
- if (property_description->type) {
- allowed_types.push_back(toStringView(*property_description->type));
+ if (property_description->allowed_type) {
+ allowed_types.push_back(toStringView(*property_description->allowed_type));
}
std::optional<std::string_view> default_value;
if (property_description->default_value) {
@@ -181,12 +180,12 @@ class CControllerServiceFactory : public minifi::core::controller::ControllerSer
minifi::utils::CControllerServiceClassDescription class_description_;
};
-MinifiProxyType minifiProxyType(const minifi::controllers::ProxyType& proxy_type) {
+minifi_proxy_type minifiProxyType(const minifi::controllers::ProxyType& proxy_type) {
switch (proxy_type) {
case minifi::controllers::ProxyType::DIRECT:
- return MinifiProxyType::MINIFI_PROXY_TYPE_DIRECT;
+ return minifi_proxy_type::MINIFI_PROXY_TYPE_DIRECT;
case minifi::controllers::ProxyType::HTTP:
- return MinifiProxyType::MINIFI_PROXY_TYPE_HTTP;
+ return minifi_proxy_type::MINIFI_PROXY_TYPE_HTTP;
}
std::unreachable();
}
@@ -195,11 +194,11 @@ MinifiProxyType minifiProxyType(const minifi::controllers::ProxyType& proxy_type
namespace org::apache::nifi::minifi::utils {
-void useCProcessorClassDescription(const MinifiProcessorClassDefinition& class_description, const std::function<void(minifi::ClassDescription, minifi::utils::CProcessorClassDescription)>& fn) {
+void useCProcessorClassDescription(const minifi_processor_definition& 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]));
+ properties.reserve(class_description.properties_count);
+ for (size_t i = 0; i < class_description.properties_count; ++i) {
+ properties.push_back(createProperty(&class_description.properties_ptr[i]));
}
std::vector<minifi::core::DynamicProperty> dynamic_properties;
dynamic_properties.reserve(class_description.dynamic_properties_count);
@@ -212,11 +211,11 @@ void useCProcessorClassDescription(const MinifiProcessorClassDefinition& class_d
});
}
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.reserve(class_description.relationships_count);
+ for (size_t i = 0; i < class_description.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)
+ toString(class_description.relationships_ptr[i].name),
+ toString(class_description.relationships_ptr[i].description)
});
}
std::vector<minifi::core::OutputAttribute> output_attributes;
@@ -226,7 +225,7 @@ void useCProcessorClassDescription(const MinifiProcessorClassDefinition& class_d
output_attribute.description = toString(class_description.output_attributes_ptr[attribute_idx].description);
for (size_t rel_idx = 0; rel_idx < class_description.output_attributes_ptr[attribute_idx].relationships_count; ++rel_idx) {
auto output_attribute_rel_name = toStringView(class_description.output_attributes_ptr[attribute_idx].relationships_ptr[rel_idx]);
- auto rel_it = std::find(relationships.begin(), relationships.end(), minifi::core::Relationship{std::string{output_attribute_rel_name}, ""});
+ auto rel_it = std::ranges::find(relationships, minifi::core::Relationship{std::string{output_attribute_rel_name}, ""});
gsl_Assert(rel_it != relationships.end());
output_attribute.relationships.push_back(*rel_it);
}
@@ -266,12 +265,12 @@ void useCProcessorClassDescription(const MinifiProcessorClassDefinition& class_d
fn(description, c_class_description);
}
-void useCControllerServiceClassDescription(const MinifiControllerServiceClassDefinition& class_description,
+void useCControllerServiceClassDescription(const std::string_view bundle_name, const minifi_controller_service_definition& class_description,
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) {
- properties.push_back(createProperty(&class_description.class_properties_ptr[i]));
+ properties.reserve(class_description.properties_count);
+ for (size_t i = 0; i < class_description.properties_count; ++i) {
+ properties.push_back(createProperty(&class_description.properties_ptr[i]));
}
auto name_segments = minifi::utils::string::split(toStringView(class_description.full_name), "::");
@@ -281,7 +280,10 @@ void useCControllerServiceClassDescription(const MinifiControllerServiceClassDef
implements_apis.reserve(class_description.provided_interfaces_count);
for (size_t i = 0; i < class_description.provided_interfaces_count; ++i) {
auto api_segments = string::split(toStringView(class_description.provided_interfaces_ptr[i]), "::");
- implements_apis.push_back(core::ControllerServiceType::minifiSystemControllerServiceType(string::join(".", api_segments)));
+ std::string type_name = string::join(".", api_segments);
+ api_segments.pop_back();
+ std::string group_name = string::join(".", api_segments);
+ implements_apis.emplace_back(std::string{bundle_name}, std::move(group_name), std::move(type_name));
}
ClassDescription description{
@@ -304,7 +306,7 @@ void useCControllerServiceClassDescription(const MinifiControllerServiceClassDef
extern "C" {
-MinifiExtension* MinifiRegisterExtension(MinifiExtensionContext* extension_context, const MinifiExtensionDefinition* extension_definition) {
+minifi_extension* minifi_register_extension(minifi_extension_context* extension_context, const minifi_extension_definition* extension_definition) {
gsl_Assert(extension_context);
gsl_Assert(extension_definition);
auto* extension = reinterpret_cast<org::apache::nifi::minifi::core::extension::Extension::Context*>(extension_context)->create(org::apache::nifi::minifi::core::extension::Extension::Info{
@@ -314,12 +316,12 @@ MinifiExtension* MinifiRegisterExtension(MinifiExtensionContext* extension_conte
.user_data = extension_definition->user_data
});
if (extension) {
- return reinterpret_cast<MinifiExtension*>(extension);
+ return reinterpret_cast<minifi_extension*>(extension);
}
- return MINIFI_NULL;
+ return nullptr;
}
-MinifiStatus MinifiRegisterProcessor(MinifiExtension* extension, const MinifiProcessorClassDefinition* processor) {
+minifi_status minifi_register_processor(minifi_extension* extension, const minifi_processor_definition* processor) {
gsl_Assert(extension);
gsl_Assert(processor);
auto* extension_ = reinterpret_cast<minifi::core::extension::Extension*>(extension);
@@ -342,7 +344,7 @@ MinifiStatus MinifiRegisterProcessor(MinifiExtension* extension, const MinifiPro
return MINIFI_STATUS_SUCCESS;
}
-MinifiStatus MinifiRegisterControllerService(MinifiExtension* extension, const MinifiControllerServiceClassDefinition* controller_service) {
+minifi_status minifi_register_controller_service(minifi_extension* extension, const minifi_controller_service_definition* controller_service) {
gsl_Assert(extension);
gsl_Assert(controller_service);
auto extension_info = reinterpret_cast<minifi::core::extension::Extension*>(extension)->getInfo();
@@ -354,7 +356,7 @@ MinifiStatus MinifiRegisterControllerService(MinifiExtension* extension, const M
.version = extension_info->version
};
auto& bundle_components = minifi::ClassDescriptionRegistry::getMutableClassDescriptions()[bundle];
- minifi::utils::useCControllerServiceClassDescription(*controller_service, [&] (const auto& description, const auto& c_class_description) {
+ minifi::utils::useCControllerServiceClassDescription(bundle.name, *controller_service, [&] (const auto& description, const auto& c_class_description) {
minifi::core::ClassLoader::getDefaultClassLoader().getClassLoader(extension_info->name).registerClass(
description.short_name_,
std::make_unique<CControllerServiceFactory>(extension_info->name, toString(controller_service->full_name), c_class_description));
@@ -364,15 +366,37 @@ MinifiStatus MinifiRegisterControllerService(MinifiExtension* extension, const M
}
-MinifiExtension* MINIFI_REGISTER_EXTENSION_FN(MinifiExtensionContext* extension_context, const MinifiExtensionDefinition* extension_definition) {
- return MinifiRegisterExtension(extension_context, extension_definition);
+minifi_extension* MINIFI_REGISTER_EXTENSION_FN(minifi_extension_context* extension_context, const minifi_extension_definition* extension_definition) {
+ return minifi_register_extension(extension_context, extension_definition);
}
-MinifiStatus MinifiProcessContextGetProperty(MinifiProcessContext* context, MinifiStringView property_name, MinifiFlowFile* flowfile,
- void (*result_cb)(void* user_ctx, MinifiStringView result), void* user_ctx) {
- gsl_Assert(context != MINIFI_NULL);
+minifi_status minifi_process_context_set_trigger_when_empty(minifi_process_context* context, bool trigger_when_empty) {
+ gsl_Assert(context);
+ reinterpret_cast<minifi::core::ProcessContext*>(context)->getProcessor().setTriggerWhenEmpty(trigger_when_empty);
+ return MINIFI_STATUS_SUCCESS;
+}
+
+minifi_status minifi_process_context_report_metrics(minifi_process_context* context, const size_t count, const minifi_string_view* metric_names, const double* metric_values) {
+ gsl_Assert(context);
+ std::vector<minifi::state::PublishedMetric> metrics;
+ metrics.reserve(count);
+ for (size_t i = 0; i < count; i++) {
+ metrics.emplace_back(minifi::state::PublishedMetric{toString(metric_names[i]), metric_values[i], {}});
+ }
+ auto& processor = reinterpret_cast<minifi::core::ProcessContext*>(context)->getProcessor();
+ auto& impl = processor.getImpl();
+ if (const auto c_processor = dynamic_cast<minifi::utils::CProcessor*>(&impl)) {
+ c_processor->reportMetrics(std::move(metrics));
+ return MINIFI_STATUS_SUCCESS;
+ }
+ return MINIFI_STATUS_UNKNOWN_ERROR;
+}
+
+minifi_status minifi_process_context_get_property(minifi_process_context* context, minifi_string_view property_name, minifi_flow_file* flowfile,
+ void (*result_cb)(void* user_ctx, minifi_string_view result), void* user_ctx) {
+ gsl_Assert(context);
auto result = reinterpret_cast<minifi::core::ProcessContext*>(context)->getProperty(toStringView(property_name),
- flowfile != MINIFI_NULL ? reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile)->get() : nullptr);
+ flowfile ? reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile)->get() : nullptr);
if (result) {
result_cb(user_ctx, minifiStringView(result.value()));
return MINIFI_STATUS_SUCCESS;
@@ -386,12 +410,7 @@ MinifiStatus MinifiProcessContextGetProperty(MinifiProcessContext* context, Mini
}
}
-MinifiBool MinifiProcessContextHasNonEmptyProperty(MinifiProcessContext* context, MinifiStringView property_name) {
- gsl_Assert(context != MINIFI_NULL);
- return reinterpret_cast<minifi::core::ProcessContext*>(context)->hasNonEmptyProperty(toString(property_name));
-}
-
-void MinifiConfigGet(MinifiExtensionContext* extension_context, MinifiStringView key, void(*cb)(void* user_ctx, MinifiStringView result), void* user_ctx) {
+void minifi_config_get(minifi_extension_context* extension_context, minifi_string_view key, void(*cb)(void* user_ctx, minifi_string_view result), void* user_ctx) {
gsl_Assert(extension_context);
auto value = reinterpret_cast<org::apache::nifi::minifi::core::extension::Extension::Context*>(extension_context)->config->get(toString(key));
if (value) {
@@ -399,65 +418,37 @@ void MinifiConfigGet(MinifiExtensionContext* extension_context, MinifiStringView
}
}
-void MinifiLoggerSetMaxLogSize(MinifiLogger* logger, int32_t max_size) {
- gsl_Assert(logger != MINIFI_NULL);
- (*reinterpret_cast<std::shared_ptr<minifi::core::logging::Logger>*>(logger))->set_max_log_size(max_size);
-}
-
-void MinifiLoggerLogString(MinifiLogger* logger, MinifiLogLevel level, MinifiStringView msg) {
- gsl_Assert(logger != MINIFI_NULL);
- (*reinterpret_cast<std::shared_ptr<minifi::core::logging::Logger>*>(logger))->log_string(toLogLevel(level), toString(msg));
+void minifi_logger_log_string(minifi_logger* logger, minifi_log_level level, minifi_string_view message) {
+ gsl_Assert(logger);
+ (*reinterpret_cast<std::shared_ptr<minifi::core::logging::Logger>*>(logger))->log_string(toLogLevel(level), toString(message));
}
-MinifiBool MinifiLoggerShouldLog(MinifiLogger* logger, MinifiLogLevel level) {
- gsl_Assert(logger != MINIFI_NULL);
+bool minifi_logger_should_log(minifi_logger* logger, minifi_log_level level) {
+ gsl_Assert(logger);
return (*reinterpret_cast<std::shared_ptr<minifi::core::logging::Logger>*>(logger))->should_log(toLogLevel(level));
}
-MinifiLogLevel MinifiLoggerLevel(MinifiLogger* logger) {
- gsl_Assert(logger != MINIFI_NULL);
- switch ((*reinterpret_cast<std::shared_ptr<minifi::core::logging::Logger>*>(logger))->level()) {
- case minifi::core::logging::LOG_LEVEL::trace: return MINIFI_LOG_LEVEL_TRACE;
- case minifi::core::logging::LOG_LEVEL::debug: return MINIFI_LOG_LEVEL_DEBUG;
- case minifi::core::logging::LOG_LEVEL::info: return MINIFI_LOG_LEVEL_INFO;
- case minifi::core::logging::LOG_LEVEL::warn: return MINIFI_LOG_LEVEL_WARNING;
- case minifi::core::logging::LOG_LEVEL::err: return MINIFI_LOG_LEVEL_ERROR;
- case minifi::core::logging::LOG_LEVEL::critical: return MINIFI_LOG_LEVEL_CRITICAL;
- case minifi::core::logging::LOG_LEVEL::off: return MINIFI_LOG_LEVEL_OFF;
- }
- gsl_FailFast();
-}
-
-MINIFI_OWNED gsl::owner<MinifiPublishedMetrics*> MinifiPublishedMetricsCreate(const size_t count, const MinifiStringView* metric_names, const double* metric_values) {
- gsl::owner<std::vector<minifi::state::PublishedMetric>*> metrics = new std::vector<minifi::state::PublishedMetric>(); // NOLINT(modernize-use-auto, cppcoreguidelines-owning-memory)
- metrics->reserve(count);
- for (size_t i = 0; i < count; i++) {
- metrics->emplace_back(minifi::state::PublishedMetric{toString(metric_names[i]), metric_values[i], {}});
- }
- return reinterpret_cast<MinifiPublishedMetrics*>(metrics);
-}
-
-MINIFI_OWNED MinifiFlowFile* MinifiProcessSessionGet(MinifiProcessSession* session) {
- gsl_Assert(session != MINIFI_NULL);
+MINIFI_OWNED minifi_flow_file* minifi_process_session_get(minifi_process_session* session) {
+ gsl_Assert(session);
if (const auto ff = reinterpret_cast<minifi::core::ProcessSession*>(session)->get()) {
- return reinterpret_cast<MinifiFlowFile*>(new std::shared_ptr<minifi::core::FlowFile>(ff));
+ return reinterpret_cast<minifi_flow_file*>(new std::shared_ptr<minifi::core::FlowFile>(ff));
}
- return MINIFI_NULL;
+ return nullptr;
}
-MINIFI_OWNED MinifiFlowFile* MinifiProcessSessionCreate(MinifiProcessSession* session, MinifiFlowFile* parent) {
- gsl_Assert(session != MINIFI_NULL);
- if (const auto ff = reinterpret_cast<minifi::core::ProcessSession*>(session)->create(parent != MINIFI_NULL
+MINIFI_OWNED minifi_flow_file* minifi_process_session_create(minifi_process_session* session, minifi_flow_file* parent) {
+ gsl_Assert(session);
+ if (const auto ff = reinterpret_cast<minifi::core::ProcessSession*>(session)->create(parent != nullptr
? reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(parent)->get()
: nullptr)) {
- return reinterpret_cast<MinifiFlowFile*>(new std::shared_ptr<minifi::core::FlowFile>(ff));
+ return reinterpret_cast<minifi_flow_file*>(new std::shared_ptr<minifi::core::FlowFile>(ff));
}
- return MINIFI_NULL;
+ return nullptr;
}
-MinifiStatus MinifiProcessSessionPenalize(MinifiProcessSession* session, MinifiFlowFile* flowfile) {
- gsl_Assert(session != MINIFI_NULL);
- gsl_Assert(flowfile != MINIFI_NULL);
+minifi_status minifi_process_session_penalize(minifi_process_session* session, minifi_flow_file* flowfile) {
+ gsl_Assert(session);
+ gsl_Assert(flowfile);
try {
reinterpret_cast<minifi::core::ProcessSession*>(session)->penalize(
*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile));
@@ -467,9 +458,9 @@ MinifiStatus MinifiProcessSessionPenalize(MinifiProcessSession* session, MinifiF
}
}
-MinifiStatus MinifiProcessSessionTransfer(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile, MinifiStringView relationship_name) {
- gsl_Assert(session != MINIFI_NULL);
- gsl_Assert(flowfile != MINIFI_NULL);
+minifi_status minifi_process_session_transfer(minifi_process_session* session, MINIFI_OWNED minifi_flow_file* flowfile, minifi_string_view relationship_name) {
+ gsl_Assert(session);
+ gsl_Assert(flowfile);
try {
reinterpret_cast<minifi::core::ProcessSession*>(session)->transfer(
*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile), minifi::core::Relationship{toString(relationship_name), ""});
@@ -479,9 +470,9 @@ MinifiStatus MinifiProcessSessionTransfer(MinifiProcessSession* session, MINIFI_
}
}
-MinifiStatus MinifiProcessSessionRemove(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile) {
- gsl_Assert(session != MINIFI_NULL);
- gsl_Assert(flowfile != MINIFI_NULL);
+minifi_status minifi_process_session_remove(minifi_process_session* session, MINIFI_OWNED minifi_flow_file* flowfile) {
+ gsl_Assert(session);
+ gsl_Assert(flowfile);
try {
reinterpret_cast<minifi::core::ProcessSession*>(session)->remove(*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile));
return MINIFI_STATUS_SUCCESS;
@@ -490,12 +481,12 @@ MinifiStatus MinifiProcessSessionRemove(MinifiProcessSession* session, MINIFI_OW
}
}
-MinifiStatus MinifiProcessSessionRead(MinifiProcessSession* session, MinifiFlowFile* flowfile, int64_t(*cb)(void* user_ctx, MinifiInputStream*), void* user_ctx) {
- gsl_Assert(session != MINIFI_NULL);
- gsl_Assert(flowfile != MINIFI_NULL);
+minifi_status minifi_process_session_read(minifi_process_session* session, minifi_flow_file* flowfile, int64_t(*cb)(void* user_ctx, minifi_input_stream*), void* user_ctx) {
+ gsl_Assert(session);
+ gsl_Assert(flowfile);
try {
reinterpret_cast<minifi::core::ProcessSession*>(session)->read(*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile), [&] (auto& input_stream) -> minifi::io::IoResult {
- const int64_t cb_result = cb(user_ctx, reinterpret_cast<MinifiInputStream*>(input_stream.get()));
+ const int64_t cb_result = cb(user_ctx, reinterpret_cast<minifi_input_stream*>(input_stream.get()));
return minifi::io::IoResult::from(cb_result);
});
return MINIFI_STATUS_SUCCESS;
@@ -504,12 +495,12 @@ MinifiStatus MinifiProcessSessionRead(MinifiProcessSession* session, MinifiFlowF
}
}
-MinifiStatus MinifiProcessSessionWrite(MinifiProcessSession* session, MinifiFlowFile* ff, int64_t(*cb)(void* user_ctx, MinifiOutputStream*), void* user_ctx) {
- gsl_Assert(session != MINIFI_NULL);
- gsl_Assert(ff != MINIFI_NULL);
+minifi_status minifi_process_session_write(minifi_process_session* session, minifi_flow_file* ff, int64_t(*cb)(void* user_ctx, minifi_output_stream*), void* user_ctx) {
+ gsl_Assert(session);
+ gsl_Assert(ff);
try {
reinterpret_cast<minifi::core::ProcessSession*>(session)->write(*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(ff), [&] (auto& output_stream) -> minifi::io::IoResult {
- const int64_t cb_result = cb(user_ctx, reinterpret_cast<MinifiOutputStream*>(output_stream.get()));
+ const int64_t cb_result = cb(user_ctx, reinterpret_cast<minifi_output_stream*>(output_stream.get()));
return minifi::io::IoResult::from(cb_result);
});
return MINIFI_STATUS_SUCCESS;
@@ -518,37 +509,40 @@ MinifiStatus MinifiProcessSessionWrite(MinifiProcessSession* session, MinifiFlow
}
}
-size_t MinifiInputStreamSize(MinifiInputStream* stream) {
- gsl_Assert(stream != MINIFI_NULL);
+size_t minifi_input_stream_size(minifi_input_stream* stream) {
+ gsl_Assert(stream);
return reinterpret_cast<minifi::io::InputStream*>(stream)->size();
}
-int64_t MinifiInputStreamRead(MinifiInputStream* stream, char* buffer, size_t size) {
- gsl_Assert(stream != MINIFI_NULL);
+int64_t minifi_input_stream_read(minifi_input_stream* stream, char* buffer, size_t size) {
+ gsl_Assert(stream);
return gsl::narrow<int64_t>(reinterpret_cast<minifi::io::InputStream*>(stream)->read(std::span(reinterpret_cast<std::byte*>(buffer), size)));
}
-int64_t MinifiOutputStreamWrite(MinifiOutputStream* stream, const char* data, size_t size) {
- gsl_Assert(stream != MINIFI_NULL);
+int64_t minifi_output_stream_write(minifi_output_stream* stream, const char* data, size_t size) {
+ gsl_Assert(stream);
return gsl::narrow<int64_t>(reinterpret_cast<minifi::io::OutputStream*>(stream)->write(as_bytes(std::span(data, size))));
}
-MinifiStatus MinifiProcessSessionSetFlowFileAttribute(MinifiProcessSession* session, MinifiFlowFile* flowfile, MinifiStringView attribute_name, const MinifiStringView* attribute_value) {
- gsl_Assert(session != MINIFI_NULL);
- gsl_Assert(flowfile != MINIFI_NULL);
+minifi_status minifi_process_session_set_flow_file_attribute(minifi_process_session* session, minifi_flow_file* flowfile,
+ minifi_string_view attribute_name, const minifi_string_view* attribute_value) {
+ gsl_Assert(session);
+ gsl_Assert(flowfile);
if (attribute_value == nullptr) {
- reinterpret_cast<minifi::core::ProcessSession*>(session)->removeAttribute(**reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile), toString(attribute_name));
+ reinterpret_cast<minifi::core::ProcessSession*>(session)->removeAttribute(**reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile),
+ toString(attribute_name));
} else {
- reinterpret_cast<minifi::core::ProcessSession*>(session)->putAttribute(
- **reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile), toString(attribute_name), toString(*attribute_value));
+ reinterpret_cast<minifi::core::ProcessSession*>(session)->putAttribute(**reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile),
+ toString(attribute_name),
+ toString(*attribute_value));
}
return MINIFI_STATUS_SUCCESS;
}
-MinifiBool MinifiProcessSessionGetFlowFileAttribute(MinifiProcessSession* session, MinifiFlowFile* flowfile, MinifiStringView attribute_name,
- void(*cb)(void* user_ctx, MinifiStringView attribute_value), void* user_ctx) {
- gsl_Assert(session != MINIFI_NULL);
- gsl_Assert(flowfile != MINIFI_NULL);
+bool minifi_process_session_get_flow_file_attribute(minifi_process_session* session, minifi_flow_file* flowfile, minifi_string_view attribute_name,
+ void(*cb)(void* user_ctx, minifi_string_view attribute_value), void* user_ctx) {
+ gsl_Assert(session);
+ gsl_Assert(flowfile);
auto value = (*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile))->getAttribute(toString(attribute_name));
if (!value.has_value()) {
return false;
@@ -557,32 +551,32 @@ MinifiBool MinifiProcessSessionGetFlowFileAttribute(MinifiProcessSession* sessio
return true;
}
-void MinifiProcessSessionGetFlowFileAttributes(MinifiProcessSession* session, MinifiFlowFile* flowfile,
- void(*cb)(void* user_ctx, MinifiStringView attribute_name, MinifiStringView attribute_value), void* user_ctx) {
- gsl_Assert(session != MINIFI_NULL);
- gsl_Assert(flowfile != MINIFI_NULL);
+void minifi_process_session_get_flow_file_attributes(minifi_process_session* session, minifi_flow_file* flowfile,
+ void(*cb)(void* user_ctx, minifi_string_view attribute_name, minifi_string_view attribute_value), void* user_ctx) {
+ gsl_Assert(session);
+ gsl_Assert(flowfile);
for (auto& [key, value] : (*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile))->getAttributes()) {
cb(user_ctx, minifiStringView(key), minifiStringView(value));
}
}
-uint64_t MinifiProcessSessionGetFlowFileSize(MinifiProcessSession* session, MinifiFlowFile* flowfile) {
- gsl_Assert(session != MINIFI_NULL);
- gsl_Assert(flowfile != MINIFI_NULL);
+uint64_t minifi_process_session_get_flow_file_size(minifi_process_session* session, minifi_flow_file* flowfile) {
+ gsl_Assert(session);
+ gsl_Assert(flowfile);
return (*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile))->getSize();
}
-MinifiStatus MinifiProcessSessionGetFlowFileId(MinifiProcessSession* session, MinifiFlowFile* flowfile, void(*cb)(void* user_ctx, MinifiStringView flow_file_id), void* user_ctx) {
- gsl_Assert(session != MINIFI_NULL);
- gsl_Assert(flowfile != MINIFI_NULL);
+minifi_status minifi_process_session_get_flow_file_id(minifi_process_session* session, minifi_flow_file* flowfile, void(*cb)(void* user_ctx, minifi_string_view flow_file_id), void* user_ctx) {
+ gsl_Assert(session);
+ gsl_Assert(flowfile);
const auto uuid_small_str = (*reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(flowfile))->getUUIDStr();
cb(user_ctx, minifiStringView(uuid_small_str.view()));
return MINIFI_STATUS_SUCCESS;
}
-MinifiStatus MinifiControllerServiceContextGetProperty(MinifiControllerServiceContext* context, MinifiStringView property_name,
- void (*result_cb)(void* user_ctx, MinifiStringView result), void* user_ctx) {
- gsl_Assert(context != MINIFI_NULL);
+minifi_status minifi_controller_service_context_get_property(minifi_controller_service_context* context, minifi_string_view property_name,
+ void (*result_cb)(void* user_ctx, minifi_string_view result), void* user_ctx) {
+ gsl_Assert(context);
auto result = reinterpret_cast<minifi::core::controller::ControllerServiceContext*>(context)->getProperty(toStringView(property_name));
if (result) {
result_cb(user_ctx, minifiStringView(result.value()));
@@ -597,16 +591,16 @@ MinifiStatus MinifiControllerServiceContextGetProperty(MinifiControllerServiceCo
}
}
-MinifiStatus MinifiProcessContextGetControllerServiceFromProperty(
- MinifiProcessContext* process_context,
- const MinifiStringView property_name,
- const MinifiStringView controller_service_type,
- MinifiControllerService** controller_service_out) {
+minifi_status minifi_process_context_get_controller_service_from_property(
+ minifi_process_context* process_context,
+ const minifi_string_view property_name,
+ const minifi_string_view controller_service_type,
+ minifi_controller_service** controller_service_out) {
if (!controller_service_out) {
return MINIFI_STATUS_UNKNOWN_ERROR;
}
- gsl_Assert(process_context != MINIFI_NULL);
+ gsl_Assert(process_context);
const auto context = reinterpret_cast<minifi::core::ProcessContext*>(process_context);
const auto property_name_str = std::string{toStringView(property_name)};
const auto name_str = context->getProperty(property_name_str, nullptr);
@@ -618,7 +612,7 @@ MinifiStatus MinifiProcessContextGetControllerServiceFromProperty(
if (const minifi::utils::CControllerService* c_controller_service = dynamic_cast<minifi::utils::CControllerService*>(&*service_shared_ptr)) {
const auto class_description = c_controller_service->getClassDescription();
if (class_description.full_name == toStringView(controller_service_type)) {
- *controller_service_out = static_cast<MinifiControllerService*>(c_controller_service->getImpl());
+ *controller_service_out = static_cast<minifi_controller_service*>(c_controller_service->getImpl());
return MINIFI_STATUS_SUCCESS;
}
if (class_description.callbacks.get_interface) {
@@ -626,25 +620,25 @@ MinifiStatus MinifiProcessContextGetControllerServiceFromProperty(
if (!interface_res) {
return MINIFI_STATUS_VALIDATION_FAILED;
}
- *controller_service_out = static_cast<MinifiControllerService*>(interface_res);
+ *controller_service_out = static_cast<minifi_controller_service*>(interface_res);
return MINIFI_STATUS_SUCCESS;
}
}
return MINIFI_STATUS_VALIDATION_FAILED;
}
-void MinifiProcessContextGetDynamicProperties(MinifiProcessContext* context, MinifiFlowFile* minifi_flow_file,
- void (*cb)(void* user_ctx, MinifiStringView dynamic_property_name, MinifiStringView dynamic_property_value), void* user_ctx) {
- gsl_Assert(context != MINIFI_NULL);
- auto flow_file = minifi_flow_file != MINIFI_NULL ? reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(minifi_flow_file)->get() : nullptr;
+void minifi_process_context_get_dynamic_properties(minifi_process_context* context, minifi_flow_file* minifi_flow_file,
+ void (*cb)(void* user_ctx, minifi_string_view dynamic_property_name, minifi_string_view dynamic_property_value), void* user_ctx) {
+ gsl_Assert(context);
+ auto flow_file = minifi_flow_file ? reinterpret_cast<std::shared_ptr<minifi::core::FlowFile>*>(minifi_flow_file)->get() : nullptr;
for (auto& [key, value] : reinterpret_cast<minifi::core::ProcessContext*>(context)->getDynamicProperties(flow_file)) {
cb(user_ctx, minifiStringView(key), minifiStringView(value));
}
}
-MinifiStatus MinifiProcessContextGetSslDataFromProperty(MinifiProcessContext* process_context, MinifiStringView property_name,
- void (*cb)(void* user_ctx, const MinifiSslData* ssl_data), void* user_ctx) {
- gsl_Assert(process_context != MINIFI_NULL);
+enum minifi_status minifi_process_context_get_ssl_data_from_property(struct minifi_process_context* process_context, struct minifi_string_view property_name,
+ void (*cb)(void* user_ctx, const struct minifi_ssl_data* ssl_data), void* user_ctx) {
+ gsl_Assert(process_context);
try {
const auto context = reinterpret_cast<minifi::core::ProcessContext*>(process_context);
const auto property_name_str = std::string{toStringView(property_name)};
@@ -658,7 +652,7 @@ MinifiStatus MinifiProcessContextGetSslDataFromProperty(MinifiProcessContext* pr
const std::string cert_file = ssl_context_service->getCertificateFile().string();
const std::string private_key_file = ssl_context_service->getPrivateKeyFile().string();
- MinifiSslData ssl_data{
+ minifi_ssl_data ssl_data{
.ca_certificate_file = minifiStringView(ca_cert_file),
.certificate_file = minifiStringView(cert_file),
.private_key_file = minifiStringView(private_key_file),
@@ -674,9 +668,9 @@ MinifiStatus MinifiProcessContextGetSslDataFromProperty(MinifiProcessContext* pr
}
-MinifiStatus MinifiProcessContextGetProxyDataFromProperty(MinifiProcessContext* process_context, MinifiStringView property_name,
- void (*cb)(void* user_ctx, const MinifiProxyData* proxy_data), void* user_ctx) {
- gsl_Assert(process_context != MINIFI_NULL);
+minifi_status minifi_process_context_get_proxy_data_from_property(minifi_process_context* process_context, minifi_string_view property_name,
+ void (*cb)(void* user_ctx, const minifi_proxy_data* proxy_data), void* user_ctx) {
+ gsl_Assert(process_context);
const auto context = reinterpret_cast<minifi::core::ProcessContext*>(process_context);
const auto property_name_str = std::string{toStringView(property_name)};
const auto name_str = context->getProperty(property_name_str, nullptr);
@@ -686,15 +680,15 @@ MinifiStatus MinifiProcessContextGetProxyDataFromProperty(MinifiProcessContext*
if (const auto proxy_service = dynamic_cast<minifi::controllers::ProxyConfigurationServiceInterface*>(service_shared_ptr.get())) {
const std::string hostname = proxy_service->getHost();
const auto basic_auth_data = proxy_service->getProxyCredentials();
- MinifiStringView username_holder = basic_auth_data ? minifiStringView(basic_auth_data->username) : MinifiStringView{};
- MinifiStringView password_holder = basic_auth_data ? minifiStringView(basic_auth_data->password) : MinifiStringView{};
+ minifi_string_view username_holder = basic_auth_data ? minifiStringView(basic_auth_data->username) : minifi_string_view{};
+ minifi_string_view password_holder = basic_auth_data ? minifiStringView(basic_auth_data->password) : minifi_string_view{};
- MinifiProxyData proxy_data{
+ minifi_proxy_data proxy_data{
+ .proxy_type = minifiProxyType(proxy_service->getProxyType()),
.hostname = minifiStringView(hostname),
.port = proxy_service->getPort(),
.username = basic_auth_data ? &username_holder : nullptr,
.password = basic_auth_data ? &password_holder : nullptr,
- .proxy_type = minifiProxyType(proxy_service->getProxyType()),
};
cb(user_ctx, &proxy_data);
return MINIFI_STATUS_SUCCESS;
diff --git a/libminifi/src/utils/CProcessor.cpp b/libminifi/src/utils/CProcessor.cpp
index 6285cb22e..69f6be69f 100644
--- a/libminifi/src/utils/CProcessor.cpp
+++ b/libminifi/src/utils/CProcessor.cpp
@@ -20,11 +20,8 @@
namespace org::apache::nifi::minifi::utils {
std::vector<minifi::state::PublishedMetric> CProcessor::getCustomMetrics() const {
- if (class_description_.callbacks.calculateMetrics == nullptr) {
- return {};
- }
- std::unique_ptr<std::vector<minifi::state::PublishedMetric>> metrics{reinterpret_cast<std::vector<minifi::state::PublishedMetric>*>(class_description_.callbacks.calculateMetrics(impl_))};
- return *metrics;
+ std::lock_guard guard(metrics_mutex_);
+ return reported_metrics_;
}
std::vector<minifi::state::response::SerializedResponseNode> CProcessorMetricsWrapper::serialize() {
diff --git a/libminifi/test/integration/extension-verification-test/CApiExtension.cpp b/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
index 13bec7e0c..d95ac09f5 100644
--- a/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
+++ b/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
@@ -40,17 +40,17 @@ class DummyCProcessor : public minifi::api::core::ProcessorImpl {
static constexpr bool IsSingleThreaded = true;
};
-CEXTENSIONAPI const uint32_t MinifiApiVersion = MINIFI_TEST_API_VERSION;
+CEXTENSIONAPI const uint32_t minifi_api_version = MINIFI_TEST_API_VERSION;
-CEXTENSIONAPI void MinifiInitExtension(MinifiExtensionContext* extension_context) {
- MinifiExtensionDefinition extension_definition{
+CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* extension_context) {
+ minifi_extension_definition extension_definition{
.name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
.version = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_VERSION)),
.deinit = nullptr,
.user_data = nullptr
};
- auto extension = MinifiRegisterExtension(extension_context, &extension_definition);
- minifi::api::core::useProcessorClassDefinition<DummyCProcessor>([&] (const MinifiProcessorClassDefinition& definition) {
- MinifiRegisterProcessor(extension, &definition);
+ auto extension = minifi_register_extension(extension_context, &extension_definition);
+ minifi::api::core::useProcessorClassDefinition<DummyCProcessor>([&] (const minifi_processor_definition& definition) {
+ minifi_register_processor(extension, &definition);
});
}
diff --git a/libminifi/test/integration/extension-verification-test/CppApiExtension.cpp b/libminifi/test/integration/extension-verification-test/CppApiExtension.cpp
index e765c0793..1b34fbc1d 100644
--- a/libminifi/test/integration/extension-verification-test/CppApiExtension.cpp
+++ b/libminifi/test/integration/extension-verification-test/CppApiExtension.cpp
@@ -24,8 +24,8 @@
namespace minifi = org::apache::nifi::minifi;
-extern "C" void MinifiInitCppExtension(MinifiExtensionContext* extension_context) {
- MinifiExtensionDefinition extension_definition{
+extern "C" void minifi_init_cpp_extension(minifi_extension_context* extension_context) {
+ minifi_extension_definition extension_definition{
.name = minifi::utils::toStringView(MAKESTRING(MODULE_NAME)),
.version = minifi::utils::toStringView("1.0.0"),
.deinit = nullptr,
diff --git a/libminifi/test/integration/extension-verification-test/CreateNotCalled.cpp b/libminifi/test/integration/extension-verification-test/CreateNotCalled.cpp
index ba2252baf..e4d2294a0 100644
--- a/libminifi/test/integration/extension-verification-test/CreateNotCalled.cpp
+++ b/libminifi/test/integration/extension-verification-test/CreateNotCalled.cpp
@@ -16,11 +16,11 @@
* limitations under the License.
*/
-#include "minifi-c.h"
#include "api/utils/minifi-c-utils.h"
+#include "minifi-api.h"
-CEXTENSIONAPI const uint32_t MinifiApiVersion = MINIFI_TEST_API_VERSION;
+CEXTENSIONAPI const uint32_t minifi_api_version = MINIFI_TEST_API_VERSION;
-CEXTENSIONAPI void MinifiInitExtension(MinifiExtensionContext* /*extension_context*/) {
+CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* /*extension_context*/) {
// NOT CALLING CREATE
}
diff --git a/libminifi/test/integration/extension-verification-test/ExtensionVerificationTests.cpp b/libminifi/test/integration/extension-verification-test/ExtensionVerificationTests.cpp
index a94474923..ab4814c7e 100644
--- a/libminifi/test/integration/extension-verification-test/ExtensionVerificationTests.cpp
+++ b/libminifi/test/integration/extension-verification-test/ExtensionVerificationTests.cpp
@@ -28,8 +28,8 @@
extern "C" {
-MinifiExtension* MinifiRegisterCppExtension_MatchingBuildId(MinifiExtensionContext* extension_context, const MinifiExtensionDefinition* extension_definition) {
- return MinifiRegisterExtension(extension_context, extension_definition);
+minifi_extension* MinifiRegisterCppExtension_MatchingBuildId(minifi_extension_context* extension_context, const minifi_extension_definition* extension_definition) {
+ return minifi_register_extension(extension_context, extension_definition);
}
} // extern "C"
@@ -88,12 +88,12 @@ TEST_CASE("Can't load c-api extensions with smaller version") {
REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(0s, "Api version is no longer supported, application supports 5-10 while extension is 4"));
}
-TEST_CASE("Can't load c-api extensions with no MinifiInitExtension function") {
+TEST_CASE("Can't load c-api extensions with no minifi_init_extension function") {
ExtensionLoadingTestController controller{"*test-extension-loading-missing-init*"};
REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(0s, "Failed to load as c extension 'test-extension-loading-missing-init'"));
}
-TEST_CASE("Can't load c-api extensions with no MinifiRegisterExtension call") {
+TEST_CASE("Can't load c-api extensions with no minifi_register_extension call") {
ExtensionLoadingTestController controller{"*test-extension-loading-create-not-called*"};
REQUIRE(minifi::test::utils::verifyLogLinePresenceInPollTime(0s, "Failed to initialize extension 'test-extension-loading-create-not-called'"));
}
diff --git a/libminifi/test/libtest/unit/ContentRepositoryDependentTests.h b/libminifi/test/libtest/unit/ContentRepositoryDependentTests.h
index 6db0790b8..053bb8a15 100644
--- a/libminifi/test/libtest/unit/ContentRepositoryDependentTests.h
+++ b/libminifi/test/libtest/unit/ContentRepositoryDependentTests.h
@@ -26,7 +26,7 @@
#include "core/ProcessSession.h"
#include "core/Processor.h"
#include "io/StreamPipe.h"
-#include "minifi-c/minifi-c.h"
+#include "minifi-api.h"
#include "unit/DummyProcessor.h"
#include "unit/TestBase.h"
diff --git a/libminifi/test/libtest/unit/MockClasses.h b/libminifi/test/libtest/unit/MockClasses.h
index a83160719..fd3e64d36 100644
--- a/libminifi/test/libtest/unit/MockClasses.h
+++ b/libminifi/test/libtest/unit/MockClasses.h
@@ -71,9 +71,7 @@ class MockControllerService : public minifi::core::controller::ControllerService
class MockProcessor : public minifi::core::ProcessorImpl {
public:
explicit MockProcessor(minifi::core::ProcessorMetadata metadata)
- : ProcessorImpl(metadata) {
- setTriggerWhenEmpty(true);
- }
+ : ProcessorImpl(metadata) {}
~MockProcessor() override = default;
@@ -97,6 +95,10 @@ class MockProcessor : public minifi::core::ProcessorImpl {
setSupportedProperties(Properties);
}
+ void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&) override {
+ context.setTriggerWhenEmpty(true);
+ }
+
void onTrigger(minifi::core::ProcessContext& context, minifi::core::ProcessSession& session) override {
auto flow_file = session.get();
if (!flow_file) {
diff --git a/minifi-api/CMakeLists.txt b/minifi-api/CMakeLists.txt
index a242392f4..e64b2a371 100644
--- a/minifi-api/CMakeLists.txt
+++ b/minifi-api/CMakeLists.txt
@@ -9,7 +9,7 @@ target_link_libraries(minifi-api INTERFACE minifi-api-common)
target_compile_definitions(minifi-api INTERFACE MINIFI_REGISTER_EXTENSION_FN=MinifiRegisterCppExtension_${BUILD_IDENTIFIER})
add_library(minifi-c-api INTERFACE)
-target_include_directories(minifi-c-api INTERFACE include/minifi-c)
+target_include_directories(minifi-c-api INTERFACE include)
if (APPLE)
target_link_options(minifi-c-api INTERFACE -undefined dynamic_lookup)
endif()
@@ -17,7 +17,7 @@ endif()
if (WIN32)
set(IMPORT_LIB_FILE ${CMAKE_CURRENT_BINARY_DIR}/minifi-c-api.lib)
add_custom_target(minifi_c_api_import_lib
- COMMAND lib.exe /def:${CMAKE_CURRENT_SOURCE_DIR}/minifi-c-api.def /out:${IMPORT_LIB_FILE} /machine:x64
+ COMMAND lib.exe /def:${CMAKE_CURRENT_SOURCE_DIR}/minifi-api.def /out:${IMPORT_LIB_FILE} /machine:x64
BYPRODUCTS ${IMPORT_LIB_FILE}
)
diff --git a/minifi-api/common/include/minifi-cpp/core/logging/Logger.h b/minifi-api/common/include/minifi-cpp/core/logging/Logger.h
index e09aa5fbd..b04ec967a 100644
--- a/minifi-api/common/include/minifi-cpp/core/logging/Logger.h
+++ b/minifi-api/common/include/minifi-cpp/core/logging/Logger.h
@@ -84,10 +84,8 @@ class Logger {
log(LOG_LEVEL::trace, std::move(fmt), std::forward<Args>(args)...);
}
- virtual void set_max_log_size(int size) = 0;
virtual void log_string(LOG_LEVEL level, std::string str) = 0;
virtual bool should_log(LOG_LEVEL level) = 0;
- [[nodiscard]] virtual LOG_LEVEL level() const = 0;
virtual ~Logger() = default;
diff --git a/minifi-api/common/include/minifi-cpp/io/StreamCallback.h b/minifi-api/common/include/minifi-cpp/io/StreamCallback.h
index ec4fcf59e..e500fd7e0 100644
--- a/minifi-api/common/include/minifi-cpp/io/StreamCallback.h
+++ b/minifi-api/common/include/minifi-cpp/io/StreamCallback.h
@@ -20,7 +20,7 @@
#include <functional>
#include <memory>
-#include "../../minifi-api/include/minifi-c/minifi-c.h"
+#include "minifi-api.h"
#include "Stream.h"
#include "minifi-cpp/utils/gsl.h"
#include "utils/expected.h"
@@ -51,7 +51,7 @@ class IoResult {
if constexpr (std::is_same_v<T, int64_t>) {
if (val < 0) {
- return IoResult(std::unexpected{static_cast<MinifiIoStatus>(val)});
+ return IoResult(std::unexpected{static_cast<minifi_io_status>(val)});
}
} else if constexpr (std::is_same_v<T, size_t>) {
if (isError(val)) {
@@ -75,12 +75,12 @@ class IoResult {
uint64_t operator*() const { return *result_; }
- std::expected<uint64_t, MinifiIoStatus> inner() const { return result_; }
+ std::expected<uint64_t, minifi_io_status> inner() const { return result_; }
private:
- explicit IoResult(std::expected<uint64_t, MinifiIoStatus> result) : result_(std::move(result)) {}
+ explicit IoResult(std::expected<uint64_t, minifi_io_status> result) : result_(std::move(result)) {}
- std::expected<uint64_t, MinifiIoStatus> result_;
+ std::expected<uint64_t, minifi_io_status> result_;
};
class ReadWriteResult {
@@ -111,9 +111,9 @@ class ReadWriteResult {
uint64_t bytes_read;
uint64_t bytes_written;
};
- explicit ReadWriteResult(std::expected<ReadWrite, MinifiIoStatus> result) : result_(std::move(result)) {}
+ explicit ReadWriteResult(std::expected<ReadWrite, minifi_io_status> result) : result_(std::move(result)) {}
- std::expected<ReadWrite, MinifiIoStatus> result_;
+ std::expected<ReadWrite, minifi_io_status> result_;
};
using OutputStreamCallback = std::function<IoResult(const std::shared_ptr<OutputStream>& output_stream)>;
diff --git a/minifi-api/include/minifi-api.h b/minifi-api/include/minifi-api.h
new file mode 100644
index 000000000..2ee6f87ab
--- /dev/null
+++ b/minifi-api/include/minifi-api.h
@@ -0,0 +1,306 @@
+/**
+ * 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.
+ */
+
+#ifndef MINIFI_API_INCLUDE_MINIFI_API_H_
+#define MINIFI_API_INCLUDE_MINIFI_API_H_
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stddef.h>
+#include <stdint.h>
+// This API targets C23. This header will fail to build with older language versions.
+
+#define MINIFI_OWNED
+#define MINIFI_NULLABLE
+
+#ifndef MINIFI_REGISTER_EXTENSION_FN
+#define MINIFI_REGISTER_EXTENSION_FN minifi_register_extension
+#endif
+
+/// To declare a processor property that expects an SSLContextService,
+/// use MINIFI_SSL_CONTEXT_SERVICE_INTERFACE_PROPERTY_TYPE in the type field of the property definition (minifi_property_definition::allowed_type)
+#define MINIFI_SSL_CONTEXT_SERVICE_INTERFACE_PROPERTY_TYPE "org.apache.nifi.minifi.controllers.SSLContextServiceInterface"
+
+/// To declare a processor property that expects an ProxyConfigurationService,
+/// use MINIFI_PROXY_CONFIGURATION_SERVICE_INTERFACE_PROPERTY_TYPE in the type field of the property definition
+/// (minifi_property_definition::allowed_type)
+#define MINIFI_PROXY_CONFIGURATION_SERVICE_INTERFACE_PROPERTY_TYPE "org.apache.nifi.minifi.controllers.ProxyConfigurationServiceInterface"
+
+enum : uint32_t {
+ MINIFI_API_VERSION = 1
+};
+
+enum minifi_io_status : int64_t {
+ MINIFI_IO_ERROR = -1,
+ MINIFI_IO_CANCEL = -125
+};
+
+enum minifi_input_requirement : uint32_t {
+ MINIFI_INPUT_REQUIRED = 0,
+ MINIFI_INPUT_ALLOWED = 1,
+ MINIFI_INPUT_FORBIDDEN = 2
+};
+
+/// Represents a non-owning read-only view to a sized, not necessarily null-terminated string.
+/// invariant: [data, data + length) is a valid range
+struct minifi_string_view {
+ /// nullable, non-owning pointer to the beginning of a continuous character sequence
+ const char* data;
+ /// the length of the buffer pointed-to by data
+ size_t length;
+};
+
+/// Represents an output relationship of a processor, part of the processor metadata
+struct minifi_relationship_definition {
+ /// Name, processors use this to transfer flow files to the connections connected to this relationship using MinifiProcessSessionTransfer.
+ struct minifi_string_view name;
+ /// Human-readable description of what flow files get routed to this relationship. Included in C2 manifest and generated processor docs.
+ struct minifi_string_view description;
+};
+
+struct minifi_output_attribute_definition {
+ struct minifi_string_view name;
+ size_t relationships_count;
+ const struct minifi_string_view* relationships_ptr;
+ struct minifi_string_view description;
+};
+
+/// Describes what the dynamic properties can be used for
+struct minifi_dynamic_property_definition {
+ struct minifi_string_view name;
+ struct minifi_string_view value;
+ struct minifi_string_view description;
+ bool supports_expression_language;
+};
+
+struct minifi_flow_file;
+struct minifi_logger;
+struct minifi_process_context;
+struct minifi_process_session;
+struct minifi_controller_service;
+struct minifi_controller_service_context;
+struct minifi_input_stream;
+struct minifi_output_stream;
+struct minifi_extension;
+struct minifi_extension_context;
+
+enum minifi_status : uint32_t {
+ MINIFI_STATUS_SUCCESS = 0,
+ MINIFI_STATUS_UNKNOWN_ERROR = 1,
+ MINIFI_STATUS_NOT_SUPPORTED_PROPERTY = 2,
+ MINIFI_STATUS_DYNAMIC_PROPERTIES_NOT_SUPPORTED = 3,
+ MINIFI_STATUS_PROPERTY_NOT_SET = 4,
+ MINIFI_STATUS_VALIDATION_FAILED = 5,
+ MINIFI_STATUS_PROCESSOR_YIELD = 6,
+};
+
+enum minifi_validator : uint32_t {
+ MINIFI_VALIDATOR_ALWAYS_VALID = 0,
+ MINIFI_VALIDATOR_NON_BLANK = 1,
+ MINIFI_VALIDATOR_TIME_PERIOD = 2,
+ MINIFI_VALIDATOR_BOOLEAN = 3,
+ MINIFI_VALIDATOR_INTEGER = 4,
+ MINIFI_VALIDATOR_UNSIGNED_INTEGER = 5,
+ MINIFI_VALIDATOR_DATA_SIZE = 6,
+ MINIFI_VALIDATOR_PORT = 7,
+};
+
+struct minifi_property_definition {
+ struct minifi_string_view name;
+ struct minifi_string_view display_name;
+ struct minifi_string_view description;
+ bool is_required;
+ bool is_sensitive;
+
+ const struct minifi_string_view* default_value;
+ size_t allowed_values_count;
+ const struct minifi_string_view* allowed_values_ptr;
+ enum minifi_validator validator;
+
+ const struct minifi_string_view* allowed_type;
+ bool supports_expression_language;
+};
+
+enum minifi_log_level : uint32_t {
+ MINIFI_LOG_LEVEL_TRACE = 0,
+ MINIFI_LOG_LEVEL_DEBUG = 1,
+ MINIFI_LOG_LEVEL_INFO = 2,
+ MINIFI_LOG_LEVEL_WARNING = 3,
+ MINIFI_LOG_LEVEL_ERROR = 4,
+ MINIFI_LOG_LEVEL_CRITICAL = 5,
+ MINIFI_LOG_LEVEL_OFF = 6
+};
+
+struct minifi_processor_metadata {
+ struct minifi_string_view uuid;
+ struct minifi_string_view name;
+ struct minifi_logger* logger; // borrowed non-null reference, live until the processor is live
+};
+
+struct minifi_controller_service_metadata {
+ struct minifi_string_view uuid;
+ struct minifi_string_view name;
+ struct minifi_logger* logger; // borrowed non-null reference, live until the controller service is live
+};
+
+struct minifi_processor_callbacks {
+ MINIFI_OWNED void* (*create)(struct minifi_processor_metadata);
+ void (*destroy)(MINIFI_OWNED void*);
+ enum minifi_status (*trigger)(void*, struct minifi_process_context*, struct minifi_process_session*);
+ enum minifi_status (*schedule)(void*, struct minifi_process_context*);
+ void (*unschedule)(void*);
+};
+
+struct minifi_controller_service_callbacks {
+ MINIFI_OWNED void* (*create)(struct minifi_controller_service_metadata);
+ void (*destroy)(MINIFI_OWNED void*);
+ enum minifi_status (*enable)(void*, struct minifi_controller_service_context*);
+ void (*disable)(void*);
+ void* (*get_interface)(void* ctx, struct minifi_string_view interface_type);
+};
+
+struct minifi_processor_definition {
+ struct minifi_string_view full_name; // '::'-delimited fully qualified name e.g. 'org::apache::nifi::minifi::GenerateFlowFile'
+ struct minifi_string_view description;
+ size_t properties_count;
+ const struct minifi_property_definition* properties_ptr;
+ size_t dynamic_properties_count;
+ const struct minifi_dynamic_property_definition* dynamic_properties_ptr;
+ size_t relationships_count;
+ const struct minifi_relationship_definition* relationships_ptr;
+ size_t output_attributes_count;
+ const struct minifi_output_attribute_definition* output_attributes_ptr;
+ bool supports_dynamic_properties;
+ bool supports_dynamic_relationships;
+ enum minifi_input_requirement input_requirement;
+ bool is_single_threaded;
+
+ struct minifi_processor_callbacks callbacks;
+};
+
+struct minifi_controller_service_definition {
+ struct minifi_string_view full_name; // '::'-delimited fully qualified name e.g. 'org::apache::nifi::minifi::processors::GetFile
+ struct minifi_string_view description;
+ size_t properties_count;
+ const struct minifi_property_definition* properties_ptr;
+ size_t provided_interfaces_count;
+ const struct minifi_string_view* provided_interfaces_ptr;
+
+ struct minifi_controller_service_callbacks callbacks;
+};
+
+struct minifi_extension_definition {
+ struct minifi_string_view name;
+ struct minifi_string_view version;
+ void (*deinit)(void* user_data);
+ void* user_data;
+};
+
+// When directly linking against the agent library (legacy c++ extension) this declares a build identifier dependent function
+// which prevents loading extensions from different builds (e.g. the agent provides MinifiRegisterCppExtension_123 but the extension
+// expects MinifiRegisterCppExtension_567). Otherwise, it declares minifi_register_extension.
+struct minifi_extension* MINIFI_REGISTER_EXTENSION_FN(struct minifi_extension_context* extension_context,
+ const struct minifi_extension_definition* extension_definition);
+
+enum minifi_status minifi_register_processor(struct minifi_extension* extension, const struct minifi_processor_definition* processor);
+enum minifi_status minifi_register_controller_service(struct minifi_extension* extension,
+ const struct minifi_controller_service_definition* controller_service);
+
+enum minifi_status minifi_process_context_set_trigger_when_empty(struct minifi_process_context*, bool trigger_when_empty);
+enum minifi_status minifi_process_context_report_metrics(struct minifi_process_context*, size_t count, const struct minifi_string_view* metric_names,
+ const double* metric_values);
+
+enum minifi_status minifi_process_context_get_property(struct minifi_process_context* context, struct minifi_string_view property_name,
+ MINIFI_NULLABLE struct minifi_flow_file* flowfile, void (*cb)(void* user_ctx, struct minifi_string_view property_value), void* user_ctx);
+
+enum minifi_status minifi_process_context_get_controller_service_from_property(struct minifi_process_context* process_context,
+ struct minifi_string_view property_name, struct minifi_string_view controller_service_type,
+ struct minifi_controller_service** controller_service_out);
+void minifi_process_context_get_dynamic_properties(struct minifi_process_context* context, MINIFI_NULLABLE struct minifi_flow_file* minifi_flow_file,
+ void (*cb)(void* user_ctx, struct minifi_string_view dynamic_property_name, struct minifi_string_view dynamic_property_value), void* user_ctx);
+
+void minifi_logger_log_string(struct minifi_logger*, enum minifi_log_level, struct minifi_string_view message);
+bool minifi_logger_should_log(struct minifi_logger*, enum minifi_log_level);
+
+MINIFI_OWNED struct minifi_flow_file* minifi_process_session_get(struct minifi_process_session*);
+MINIFI_OWNED struct minifi_flow_file* minifi_process_session_create(struct minifi_process_session* session,
+ MINIFI_NULLABLE struct minifi_flow_file* parent_flowfile);
+
+enum minifi_status minifi_process_session_penalize(struct minifi_process_session* session, struct minifi_flow_file* flowfile);
+enum minifi_status minifi_process_session_transfer(struct minifi_process_session* session, MINIFI_OWNED struct minifi_flow_file* flowfile,
+ struct minifi_string_view relationship_name);
+enum minifi_status minifi_process_session_remove(struct minifi_process_session* session, MINIFI_OWNED struct minifi_flow_file* flowfile);
+
+enum minifi_status minifi_process_session_read(struct minifi_process_session*, struct minifi_flow_file*,
+ int64_t (*cb)(void* user_ctx, struct minifi_input_stream*), void* user_ctx);
+enum minifi_status minifi_process_session_write(struct minifi_process_session*, struct minifi_flow_file*,
+ int64_t (*cb)(void* user_ctx, struct minifi_output_stream*), void* user_ctx);
+
+void minifi_config_get(struct minifi_extension_context* extension_context, struct minifi_string_view config_key,
+ void (*cb)(void* user_ctx, struct minifi_string_view config_value), void* user_ctx);
+
+size_t minifi_input_stream_size(struct minifi_input_stream*);
+
+int64_t minifi_input_stream_read(struct minifi_input_stream* stream, char* buffer, size_t buffer_size);
+int64_t minifi_output_stream_write(struct minifi_output_stream* stream, const char* data, size_t data_size);
+
+enum minifi_status minifi_process_session_set_flow_file_attribute(struct minifi_process_session* session, struct minifi_flow_file* flowfile,
+ struct minifi_string_view attribute_name, const struct minifi_string_view* attribute_value);
+bool minifi_process_session_get_flow_file_attribute(struct minifi_process_session* session, struct minifi_flow_file* flowfile,
+ struct minifi_string_view attribute_name, void (*cb)(void* user_ctx, struct minifi_string_view attribute_value), void* user_ctx);
+void minifi_process_session_get_flow_file_attributes(struct minifi_process_session* session, struct minifi_flow_file* flowfile,
+ void (*cb)(void* user_ctx, struct minifi_string_view attribute_name, struct minifi_string_view attribute_value), void* user_ctx);
+uint64_t minifi_process_session_get_flow_file_size(struct minifi_process_session* session, struct minifi_flow_file* flowfile);
+enum minifi_status minifi_process_session_get_flow_file_id(struct minifi_process_session* session, struct minifi_flow_file* flowfile,
+ void (*cb)(void* user_ctx, struct minifi_string_view flow_file_id), void* user_ctx);
+
+enum minifi_status minifi_controller_service_context_get_property(struct minifi_controller_service_context* context,
+ struct minifi_string_view property_name, void (*cb)(void* user_ctx, struct minifi_string_view property_value), void* user_ctx);
+
+struct minifi_ssl_data {
+ struct minifi_string_view ca_certificate_file;
+ struct minifi_string_view certificate_file;
+ struct minifi_string_view private_key_file;
+ struct minifi_string_view passphrase;
+};
+
+enum minifi_status minifi_process_context_get_ssl_data_from_property(struct minifi_process_context* process_context,
+ struct minifi_string_view property_name, void (*cb)(void* user_ctx, const struct minifi_ssl_data* ssl_data), void* user_ctx);
+
+enum minifi_proxy_type : uint8_t {
+ MINIFI_PROXY_TYPE_DIRECT,
+ MINIFI_PROXY_TYPE_HTTP
+};
+
+struct minifi_proxy_data {
+ enum minifi_proxy_type proxy_type;
+ struct minifi_string_view hostname;
+ uint16_t port;
+ struct minifi_string_view* username;
+ struct minifi_string_view* password;
+};
+
+enum minifi_status minifi_process_context_get_proxy_data_from_property(struct minifi_process_context* process_context,
+ struct minifi_string_view property_name, void (*cb)(void* user_ctx, const struct minifi_proxy_data* proxy_data), void* user_ctx);
+
+#ifdef __cplusplus
+} // extern "C"
+#endif // __cplusplus
+
+#endif // MINIFI_API_INCLUDE_MINIFI_API_H_
diff --git a/minifi-api/include/minifi-c/minifi-c.h b/minifi-api/include/minifi-c/minifi-c.h
deleted file mode 100644
index 1fdfe953c..000000000
--- a/minifi-api/include/minifi-c/minifi-c.h
+++ /dev/null
@@ -1,314 +0,0 @@
-/**
- * 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.
- */
-
-#ifndef MINIFI_API_INCLUDE_MINIFI_C_MINIFI_C_H_
-#define MINIFI_API_INCLUDE_MINIFI_C_MINIFI_C_H_
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
-#include <stddef.h>
-#include <stdint.h>
-#if __STDC_VERSION__ < 202311l
-# include <stdbool.h>
-#endif // < C23
-
-#define MINIFI_PRIVATE_STRINGIFY_HELPER(X) #X
-#define MINIFI_PRIVATE_STRINGIFY(X) MINIFI_PRIVATE_STRINGIFY_HELPER(X)
-
-#define MINIFI_PRIVATE_JOIN_HELPER(X, Y) X ## _ ## Y
-#define MINIFI_PRIVATE_JOIN(X, Y) MINIFI_PRIVATE_JOIN_HELPER(X, Y)
-
-#define MINIFI_NULL nullptr
-#define MINIFI_OWNED
-
-#ifndef MINIFI_REGISTER_EXTENSION_FN
-#define MINIFI_REGISTER_EXTENSION_FN MinifiRegisterExtension
-#endif
-
-/// To declare a processor property that expects an SSLContextService,
-/// use MINIFI_SSL_CONTEXT_SERVICE_PROPERTY_TYPE in the type field of the property definition (MinifiPropertyDefinition::type)
-#define MINIFI_SSL_CONTEXT_SERVICE_PROPERTY_TYPE "org.apache.nifi.minifi.controllers.SSLContextServiceInterface"
-
-/// To declare a processor property that expects an ProxyConfigurationService,
-/// use MINIFI_PROXY_CONFIGURATION_SERVICE_PROPERTY_TYPE in the type field of the property definition (MinifiPropertyDefinition::type)
-#define MINIFI_PROXY_CONFIGURATION_SERVICE_PROPERTY_TYPE "org.apache.nifi.minifi.controllers.ProxyConfigurationServiceInterface"
-
-
-enum : uint32_t {
- MINIFI_API_VERSION = 3
-};
-
-typedef bool MinifiBool;
-
-typedef enum MinifiIoStatus : int64_t {
- MINIFI_IO_ERROR = -1,
- MINIFI_IO_CANCEL = -125
-} MinifiIoStatus;
-
-typedef enum MinifiInputRequirement : uint32_t {
- MINIFI_INPUT_REQUIRED = 0,
- MINIFI_INPUT_ALLOWED = 1,
- MINIFI_INPUT_FORBIDDEN = 2
-} MinifiInputRequirement;
-
-/// Represents a non-owning read-only view to a sized, not necessarily null-terminated string.
-/// invariant: [data, data + length) is a valid range
-typedef struct MinifiStringView {
- /// nullable, non-owning pointer to the beginning of a continuous character sequence
- const char* data;
- /// the length of the buffer pointed-to by data
- size_t length;
-} MinifiStringView;
-
-/// Represents an output relationship of a processor, part of the processor metadata
-typedef struct MinifiRelationshipDefinition {
- /// Name, processors use this to transfer flow files to the connections connected to this relationship using MinifiProcessSessionTransfer.
- MinifiStringView name;
- /// Human-readable description of what flow files get routed to this relationship. Included in C2 manifest and generated processor docs.
- MinifiStringView description;
-} MinifiRelationshipDefinition;
-
-typedef struct MinifiOutputAttributeDefinition {
- MinifiStringView name;
- size_t relationships_count;
- const MinifiStringView* relationships_ptr;
- MinifiStringView description;
-} MinifiOutputAttributeDefinition;
-
-typedef struct MinifiDynamicPropertyDefinition {
- MinifiStringView name;
- MinifiStringView value;
- MinifiStringView description;
- MinifiBool supports_expression_language;
-} MinifiDynamicPropertyDefinition;
-
-typedef struct MinifiFlowFile MinifiFlowFile;
-typedef struct MinifiLogger MinifiLogger;
-typedef struct MinifiProcessContext MinifiProcessContext;
-typedef struct MinifiProcessSession MinifiProcessSession;
-typedef struct MinifiInputStream MinifiInputStream;
-typedef struct MinifiOutputStream MinifiOutputStream;
-typedef struct MinifiExtension MinifiExtension;
-typedef struct MinifiExtensionContext MinifiExtensionContext;
-typedef struct MinifiPublishedMetrics MinifiPublishedMetrics;
-typedef struct MinifiAgent MinifiAgent;
-
-typedef struct MinifiControllerService MinifiControllerService;
-typedef struct MinifiControllerServiceContext MinifiControllerServiceContext;
-
-typedef enum MinifiStatus : uint32_t {
- MINIFI_STATUS_SUCCESS = 0,
- MINIFI_STATUS_UNKNOWN_ERROR = 1,
- MINIFI_STATUS_NOT_SUPPORTED_PROPERTY = 2,
- MINIFI_STATUS_DYNAMIC_PROPERTIES_NOT_SUPPORTED = 3,
- MINIFI_STATUS_PROPERTY_NOT_SET = 4,
- MINIFI_STATUS_VALIDATION_FAILED = 5,
- MINIFI_STATUS_PROCESSOR_YIELD = 6
-} MinifiStatus;
-
-typedef enum MinifiValidator : uint32_t {
- MINIFI_VALIDATOR_ALWAYS_VALID = 0,
- MINIFI_VALIDATOR_NON_BLANK = 1,
- MINIFI_VALIDATOR_TIME_PERIOD = 2,
- MINIFI_VALIDATOR_BOOLEAN = 3,
- MINIFI_VALIDATOR_INTEGER = 4,
- MINIFI_VALIDATOR_UNSIGNED_INTEGER = 5,
- MINIFI_VALIDATOR_DATA_SIZE = 6,
- MINIFI_VALIDATOR_PORT = 7
-} MinifiValidator;
-
-typedef struct MinifiPropertyDefinition {
- MinifiStringView name;
- MinifiStringView display_name;
- MinifiStringView description;
- MinifiBool is_required;
- MinifiBool is_sensitive;
-
- const MinifiStringView* default_value;
- size_t allowed_values_count;
- const MinifiStringView* allowed_values_ptr;
- MinifiValidator validator;
-
- const MinifiStringView* type;
- MinifiBool supports_expression_language;
-} MinifiPropertyDefinition;
-
-typedef enum MinifiLogLevel : uint32_t {
- MINIFI_LOG_LEVEL_TRACE = 0,
- MINIFI_LOG_LEVEL_DEBUG = 1,
- MINIFI_LOG_LEVEL_INFO = 2,
- MINIFI_LOG_LEVEL_WARNING = 3,
- MINIFI_LOG_LEVEL_ERROR = 4,
- MINIFI_LOG_LEVEL_CRITICAL = 5,
- MINIFI_LOG_LEVEL_OFF = 6
-} MinifiLogLevel;
-
-typedef struct MinifiProcessorMetadata {
- MinifiStringView uuid;
- MinifiStringView name;
- MinifiLogger* logger; // borrowed non-null reference, live until the processor is live
-} MinifiProcessorMetadata;
-
-typedef struct MinifiControllerServiceMetadata {
- MinifiStringView uuid;
- MinifiStringView name;
- MinifiLogger* logger; // borrowed non-null reference, live until the controller service is live
-} MinifiControllerServiceMetadata;
-
-typedef struct MinifiProcessorCallbacks {
- MINIFI_OWNED void*(*create)(MinifiProcessorMetadata);
- void(*destroy)(MINIFI_OWNED void*);
- MinifiBool(*getTriggerWhenEmpty)(void*);
- MinifiStatus(*onTrigger)(void*, MinifiProcessContext*, MinifiProcessSession*);
- MinifiStatus(*onSchedule)(void*, MinifiProcessContext*);
- void(*onUnSchedule)(void*);
- MINIFI_OWNED MinifiPublishedMetrics*(*calculateMetrics)(void*);
-} MinifiProcessorCallbacks;
-
-typedef struct MinifiControllerServiceCallbacks {
- MINIFI_OWNED void*(*create)(MinifiControllerServiceMetadata);
- void(*destroy)(MINIFI_OWNED void*);
- MinifiStatus(*enable)(void*, MinifiControllerServiceContext*);
- void(*disable)(void*);
- void* (*get_interface)(void* ctx, MinifiStringView interface_type);
-} MinifiControllerServiceCallbacks;
-
-typedef struct MinifiProcessorClassDefinition {
- MinifiStringView full_name; // '::'-delimited fully qualified name e.g. 'org::apache::nifi::minifi::GenerateFlowFile'
- MinifiStringView description;
- size_t class_properties_count;
- const MinifiPropertyDefinition* class_properties_ptr;
- size_t dynamic_properties_count;
- const MinifiDynamicPropertyDefinition* dynamic_properties_ptr;
- size_t class_relationships_count;
- const MinifiRelationshipDefinition* class_relationships_ptr;
- size_t output_attributes_count;
- const MinifiOutputAttributeDefinition* output_attributes_ptr;
- MinifiBool supports_dynamic_properties;
- MinifiBool supports_dynamic_relationships;
- MinifiInputRequirement input_requirement;
- MinifiBool is_single_threaded;
-
- MinifiProcessorCallbacks callbacks;
-} MinifiProcessorClassDefinition;
-
-typedef struct MinifiControllerServiceClassDefinition {
- MinifiStringView full_name; // '::'-delimited fully qualified name e.g. 'org::apache::nifi::minifi::extensions::gcp::GCPCredentialsControllerService
- MinifiStringView description;
- size_t class_properties_count;
- const MinifiPropertyDefinition* class_properties_ptr;
- size_t provided_interfaces_count;
- const MinifiStringView* provided_interfaces_ptr;
-
- MinifiControllerServiceCallbacks callbacks;
-} MinifiControllerServiceClassDefinition;
-
-typedef struct MinifiExtensionDefinition {
- MinifiStringView name;
- MinifiStringView version;
- void(*deinit)(void* user_data);
- void* user_data;
-} MinifiExtensionDefinition;
-
-// When directly linking against the agent library (legacy c++ extension) this declares a build identifier dependent function
-// which prevents loading extensions from different builds (e.g. the agent provides MinifiRegisterCppExtension_123 but the extension
-// expects MinifiRegisterCppExtension_567). Otherwise, it declares MinifiRegisterExtension.
-MinifiExtension* MINIFI_REGISTER_EXTENSION_FN(MinifiExtensionContext* extension_context, const MinifiExtensionDefinition* extension_definition);
-
-MinifiStatus MinifiRegisterProcessor(MinifiExtension* extension, const MinifiProcessorClassDefinition* processor);
-
-MinifiStatus MinifiRegisterControllerService(MinifiExtension* extension, const MinifiControllerServiceClassDefinition* controller_service);
-
-MINIFI_OWNED MinifiPublishedMetrics* MinifiPublishedMetricsCreate(size_t count, const MinifiStringView* metric_names, const double* metric_values);
-
-MinifiStatus MinifiProcessContextGetProperty(MinifiProcessContext* context, MinifiStringView property_name, MinifiFlowFile* flowfile,
- void(*cb)(void* user_ctx, MinifiStringView property_value), void* user_ctx);
-MinifiBool MinifiProcessContextHasNonEmptyProperty(MinifiProcessContext* context, MinifiStringView property_name);
-
-MinifiStatus MinifiProcessContextGetControllerServiceFromProperty(
- MinifiProcessContext* process_context, MinifiStringView property_name, MinifiStringView controller_service_type, MinifiControllerService** controller_service_out);
-void MinifiProcessContextGetDynamicProperties(MinifiProcessContext* context, MinifiFlowFile* minifi_flow_file,
- void (*cb)(void* user_ctx, MinifiStringView dynamic_property_name, MinifiStringView dynamic_property_value), void* user_ctx);
-
-void MinifiLoggerSetMaxLogSize(MinifiLogger*, int32_t);
-void MinifiLoggerLogString(MinifiLogger*, MinifiLogLevel, MinifiStringView);
-MinifiBool MinifiLoggerShouldLog(MinifiLogger*, MinifiLogLevel);
-MinifiLogLevel MinifiLoggerLevel(MinifiLogger*);
-
-MINIFI_OWNED MinifiFlowFile* MinifiProcessSessionGet(MinifiProcessSession*);
-MINIFI_OWNED MinifiFlowFile* MinifiProcessSessionCreate(MinifiProcessSession* session, MinifiFlowFile* parent_flowfile);
-
-MinifiStatus MinifiProcessSessionPenalize(MinifiProcessSession* session, MinifiFlowFile* flowfile);
-MinifiStatus MinifiProcessSessionTransfer(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile, MinifiStringView relationship_name);
-MinifiStatus MinifiProcessSessionRemove(MinifiProcessSession* session, MINIFI_OWNED MinifiFlowFile* flowfile);
-
-MinifiStatus MinifiProcessSessionRead(MinifiProcessSession*, MinifiFlowFile*, int64_t(*cb)(void* user_ctx, MinifiInputStream*), void* user_ctx);
-MinifiStatus MinifiProcessSessionWrite(MinifiProcessSession*, MinifiFlowFile*, int64_t(*cb)(void* user_ctx, MinifiOutputStream*), void* user_ctx);
-
-void MinifiConfigGet(MinifiExtensionContext* extension_context, MinifiStringView config_key, void(*cb)(void* user_ctx, MinifiStringView config_value), void* user_ctx);
-
-size_t MinifiInputStreamSize(MinifiInputStream*);
-
-int64_t MinifiInputStreamRead(MinifiInputStream* stream, char* buffer, size_t size);
-int64_t MinifiOutputStreamWrite(MinifiOutputStream* stream, const char* data, size_t size);
-
-MinifiStatus MinifiProcessSessionSetFlowFileAttribute(MinifiProcessSession* session, MinifiFlowFile* flowfile, MinifiStringView attribute_name, const MinifiStringView* attribute_value);
-MinifiBool MinifiProcessSessionGetFlowFileAttribute(MinifiProcessSession* session, MinifiFlowFile* flowfile, MinifiStringView attribute_name,
- void(*cb)(void* user_ctx, MinifiStringView attribute_value), void* user_ctx);
-void MinifiProcessSessionGetFlowFileAttributes(MinifiProcessSession* session, MinifiFlowFile* flowfile,
- void (*cb)(void* user_ctx, MinifiStringView attribute_name, MinifiStringView attribute_value), void* user_ctx);
-uint64_t MinifiProcessSessionGetFlowFileSize(MinifiProcessSession* session, MinifiFlowFile* flowfile);
-MinifiStatus MinifiProcessSessionGetFlowFileId(MinifiProcessSession* session, MinifiFlowFile* flowfile, void(*cb)(void* user_ctx, MinifiStringView flow_file_id), void* user_ctx);
-
-MinifiStatus MinifiControllerServiceContextGetProperty(MinifiControllerServiceContext* context,
- MinifiStringView property_name,
- void(*cb)(void* user_ctx, MinifiStringView property_value),
- void* user_ctx);
-
-typedef struct MinifiSslData {
- MinifiStringView ca_certificate_file;
- MinifiStringView certificate_file;
- MinifiStringView private_key_file;
- MinifiStringView passphrase;
-} MinifiSslData;
-
-MinifiStatus MinifiProcessContextGetSslDataFromProperty(MinifiProcessContext* process_context, MinifiStringView property_name,
- void (*cb)(void* user_ctx, const MinifiSslData* ssl_data), void* user_ctx);
-
-typedef enum MinifiProxyType : uint8_t {
- MINIFI_PROXY_TYPE_DIRECT,
- MINIFI_PROXY_TYPE_HTTP
-} MinifiProxyType;
-
-typedef struct MinifiProxyData {
- MinifiStringView hostname;
- uint16_t port;
- MinifiStringView* username;
- MinifiStringView* password;
- MinifiProxyType proxy_type;
-} MinifiProxyData;
-
-MinifiStatus MinifiProcessContextGetProxyDataFromProperty(MinifiProcessContext* process_context, MinifiStringView property_name,
- void (*cb)(void* user_ctx, const MinifiProxyData* proxy_data), void* user_ctx);
-
-#ifdef __cplusplus
-} // extern "C"
-#endif // __cplusplus
-
-#endif // MINIFI_API_INCLUDE_MINIFI_C_MINIFI_C_H_
diff --git a/minifi-api/include/minifi-cpp/core/ControllerServiceType.h b/minifi-api/include/minifi-cpp/core/ControllerServiceType.h
index f45a8e3fb..c29fb4ee0 100644
--- a/minifi-api/include/minifi-cpp/core/ControllerServiceType.h
+++ b/minifi-api/include/minifi-cpp/core/ControllerServiceType.h
@@ -27,16 +27,11 @@ struct ControllerServiceType {
std::string group;
std::string type;
- ControllerServiceType(const ControllerServiceTypeDefinition& definition) // NOLINT(runtime/explicit)
+ explicit ControllerServiceType(const ControllerServiceTypeDefinition& definition)
: artifact(definition.artifact),
group(definition.group),
type(definition.type) {}
- static ControllerServiceType minifiSystemControllerServiceType(std::string type) {
- return ControllerServiceType("minifi-system", "org.apache.nifi.minifi", std::move(type));
- }
-
- private:
ControllerServiceType(std::string artifact, std::string group, std::string type)
: artifact(std::move(artifact)),
group(std::move(group)),
diff --git a/minifi-api/include/minifi-cpp/core/ProcessContext.h b/minifi-api/include/minifi-cpp/core/ProcessContext.h
index 6fa663a7a..7cecd92f1 100644
--- a/minifi-api/include/minifi-cpp/core/ProcessContext.h
+++ b/minifi-api/include/minifi-cpp/core/ProcessContext.h
@@ -99,6 +99,7 @@ class ProcessContext : public virtual core::VariableRegistry, public virtual uti
virtual StateManager* getStateManager() = 0;
virtual void setSessionStateManager(std::unique_ptr<StateManager> state_manager) = 0;
virtual gsl::not_null<Configure*> getConfiguration() const = 0;
+ virtual void setTriggerWhenEmpty(bool trigger_when_empty) = 0;
};
} // namespace org::apache::nifi::minifi::core
diff --git a/minifi-api/include/minifi-cpp/core/ProcessorApi.h b/minifi-api/include/minifi-cpp/core/ProcessorApi.h
index 2ce94f853..f4ce64288 100644
--- a/minifi-api/include/minifi-cpp/core/ProcessorApi.h
+++ b/minifi-api/include/minifi-cpp/core/ProcessorApi.h
@@ -51,7 +51,6 @@ class ProcessorApi {
virtual void initialize(ProcessorDescriptor& descriptor) = 0;
virtual bool isSingleThreaded() const = 0;
virtual std::string getProcessorType() const = 0;
- virtual bool getTriggerWhenEmpty() const = 0;
virtual void onTrigger(ProcessContext&, ProcessSession&) = 0;
virtual void onSchedule(ProcessContext&, ProcessSessionFactory&) = 0;
virtual void onUnSchedule() = 0;
diff --git a/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h b/minifi-api/include/minifi-cpp/core/logging/AdvancedLogger.h
similarity index 50%
copy from extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
copy to minifi-api/include/minifi-cpp/core/logging/AdvancedLogger.h
index 5a4401344..638d568b9 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
+++ b/minifi-api/include/minifi-cpp/core/logging/AdvancedLogger.h
@@ -15,27 +15,29 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#pragma once
-#include <iostream>
-#include <string>
+#pragma once
-#include "minifi-c.h"
#include "minifi-cpp/core/logging/Logger.h"
-
-namespace org::apache::nifi::minifi::api::core::logging {
-
-class CffiLogger : public minifi::core::logging::Logger {
+namespace org::apache::nifi::minifi::core::logging {
+class AdvancedLogger : public Logger {
public:
- explicit CffiLogger(MinifiLogger* impl): impl_(impl) {}
+ virtual void set_max_log_size(int size) = 0;
+ [[nodiscard]] virtual LOG_LEVEL level() const = 0;
- void set_max_log_size(int size) override;
- void log_string(minifi::core::logging::LOG_LEVEL level, std::string str) override;
- bool should_log(minifi::core::logging::LOG_LEVEL level) override;
- [[nodiscard]] minifi::core::logging::LOG_LEVEL level() const override;
+ static std::expected<void, std::error_code> setMaxLogSize(Logger* logger, int size) {
+ if (const auto advanced_logger = dynamic_cast<AdvancedLogger*>(logger)) {
+ advanced_logger->set_max_log_size(size);
+ return {};
+ }
+ return std::unexpected{std::make_error_code(std::errc::invalid_argument)};
+ }
- private:
- MinifiLogger* impl_;
+ static std::expected<LOG_LEVEL, std::error_code> getLevel(Logger* logger) {
+ if (const auto advanced_logger = dynamic_cast<AdvancedLogger*>(logger)) {
+ return advanced_logger->level();
+ }
+ return std::unexpected{std::make_error_code(std::errc::invalid_argument)};
+ }
};
-
-} // namespace org::apache::nifi::minifi::api::core::logging
+} // namespace org::apache::nifi::minifi::core::logging
diff --git a/minifi-api/minifi-api.def b/minifi-api/minifi-api.def
new file mode 100644
index 000000000..a0a0b74ad
--- /dev/null
+++ b/minifi-api/minifi-api.def
@@ -0,0 +1,31 @@
+LIBRARY core-minifi.dll
+EXPORTS
+ minifi_config_get
+ minifi_controller_service_context_get_property
+ minifi_input_stream_read
+ minifi_input_stream_size
+ minifi_logger_log_string
+ minifi_logger_should_log
+ minifi_output_stream_write
+ minifi_process_context_get_controller_service_from_property
+ minifi_process_context_get_dynamic_properties
+ minifi_process_context_get_property
+ minifi_process_context_get_proxy_data_from_property
+ minifi_process_context_get_ssl_data_from_property
+ minifi_process_context_report_metrics
+ minifi_process_context_set_trigger_when_empty
+ minifi_process_session_create
+ minifi_process_session_get
+ minifi_process_session_get_flow_file_attribute
+ minifi_process_session_get_flow_file_attributes
+ minifi_process_session_get_flow_file_id
+ minifi_process_session_get_flow_file_size
+ minifi_process_session_penalize
+ minifi_process_session_read
+ minifi_process_session_remove
+ minifi_process_session_set_flow_file_attribute
+ minifi_process_session_transfer
+ minifi_process_session_write
+ minifi_register_controller_service
+ minifi_register_extension
+ minifi_register_processor
diff --git a/minifi-api/minifi-c-api.def b/minifi-api/minifi-c-api.def
deleted file mode 100644
index e74b9b9f1..000000000
--- a/minifi-api/minifi-c-api.def
+++ /dev/null
@@ -1,33 +0,0 @@
-LIBRARY core-minifi.dll
-EXPORTS
- MinifiRegisterExtension
- MinifiRegisterProcessor
- MinifiRegisterControllerService
- MinifiPublishedMetricsCreate
- MinifiProcessContextGetControllerServiceFromProperty
- MinifiProcessContextGetProperty
- MinifiProcessContextHasNonEmptyProperty
- MinifiControllerServiceContextGetProperty
- MinifiLoggerSetMaxLogSize
- MinifiLoggerLogString
- MinifiLoggerShouldLog
- MinifiLoggerLevel
- MinifiProcessSessionGet
- MinifiProcessSessionCreate
- MinifiProcessSessionTransfer
- MinifiProcessSessionRemove
- MinifiProcessSessionPenalize
- MinifiProcessSessionRead
- MinifiProcessSessionWrite
- MinifiConfigGet
- MinifiInputStreamSize
- MinifiInputStreamRead
- MinifiOutputStreamWrite
- MinifiProcessSessionSetFlowFileAttribute
- MinifiProcessSessionGetFlowFileAttribute
- MinifiProcessSessionGetFlowFileAttributes
- MinifiProcessSessionGetFlowFileSize
- MinifiProcessSessionGetFlowFileId
- MinifiProcessContextGetDynamicProperties
- MinifiProcessContextGetSslDataFromProperty
- MinifiProcessContextGetProxyDataFromProperty