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/06/17 16:16:02 UTC
(nifi-minifi-cpp) branch beke_2 created (now af14beba0)
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a change to branch beke_2
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
at af14beba0 changes found from rust wrapper rebase
This branch includes the following new commits:
new 91c53f458 MINIFICPP-2850 Finalize stable API before 1.0
new af14beba0 changes found from rust wrapper rebase
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(nifi-minifi-cpp) 02/02: changes found from rust wrapper rebase
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch beke_2
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit af14beba0b36f2f0f1d6d7c73a6f3404ea64c08f
Author: Martin Zink <ma...@apache.org>
AuthorDate: Wed Jun 17 17:45:02 2026 +0200
changes found from rust wrapper rebase
---
Extensions.md | 4 ++--
.../cpp-extension-lib/include/api/core/ProcessContext.h | 2 ++
.../cpp-extension-lib/mocklib/include/MockProcessContext.h | 2 ++
extensions/gcp/ExtensionInitializer.cpp | 2 +-
extensions/kafka/ConsumeKafka.cpp | 2 +-
extensions/kafka/ExtensionInitializer.cpp | 2 +-
extensions/llamacpp/ExtensionInitializer.cpp | 2 +-
extensions/stable-api-sandbox/ExtensionInitializer.cpp | 2 +-
libminifi/src/minifi-api.cpp | 4 ++--
.../integration/extension-verification-test/CApiExtension.cpp | 2 +-
.../extension-verification-test/ExtensionVerificationTests.cpp | 4 ++--
minifi-api/include/minifi-c/minifi-api.h | 8 ++++----
12 files changed, 20 insertions(+), 16 deletions(-)
diff --git a/Extensions.md b/Extensions.md
index 5fec72529..cebe83671 100644
--- a/Extensions.md
+++ b/Extensions.md
@@ -32,7 +32,7 @@ Unless otherwise specified, the following lifetime rules apply to all functions
* Created Resources: The lifetime of resources created within these functions (e.g., a handle returned by `MinifiProcessSessionGet` inside `MinifiProcessorCallbacks::onTrigger`)
is limited to the scope of the innermost callback.
-(the return value of `MinifiRegisterExtension` is only valid during the execution of `minifi_init_extension`).
+(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 `minifi_init_extension` call.
One possible example of this is:
@@ -47,7 +47,7 @@ extern "C" void minifi_init_extension(MinifiExtensionContext* extension_context)
.deinit = nullptr,
.user_data = nullptr
};
- auto* extension = MinifiRegisterExtension(extension_context, &extension_definition);
+ auto* extension = minifi_register_extension(extension_context, &extension_definition);
minifi::api::core::useProcessorClassDefinition<minifi::extensions::llamacpp::processors::RunLlamaCppInference>([&] (const MinifiProcessorClassDefinition& definition) {
MinifiRegisterProcessor(extension, &definition);
});
diff --git a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h b/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
index a1e528d88..e3f821cfd 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
@@ -48,6 +48,7 @@ class ProcessContext {
[[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 {
@@ -63,6 +64,7 @@ class CffiProcessContext : public ProcessContext {
[[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;
diff --git a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
index 153068646..f133a32c8 100644
--- a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
@@ -39,6 +39,8 @@ class MockProcessContext : public api::core::ProcessContext {
*last_published_metrics_ = metrics;
}
+ [[nodiscard]] std::expected<void, std::error_code> setTriggerWhenEmpty(bool) override {}
+
std::map<std::string, std::string, std::less<>> properties_;
std::unique_ptr<api::core::PublishedMetrics> last_published_metrics_ = std::make_unique<api::core::PublishedMetrics>();
diff --git a/extensions/gcp/ExtensionInitializer.cpp b/extensions/gcp/ExtensionInitializer.cpp
index 26014afb4..267c5fed6 100644
--- a/extensions/gcp/ExtensionInitializer.cpp
+++ b/extensions/gcp/ExtensionInitializer.cpp
@@ -37,7 +37,7 @@ CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* extension_con
.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/kafka/ConsumeKafka.cpp b/extensions/kafka/ConsumeKafka.cpp
index 3c1f9ae14..facc83cae 100644
--- a/extensions/kafka/ConsumeKafka.cpp
+++ b/extensions/kafka/ConsumeKafka.cpp
@@ -66,7 +66,7 @@ minifi_status ConsumeKafka::onScheduleImpl(api::core::ProcessContext& context) {
configureNewConnection(context);
if (commit_policy_ == consume_kafka::CommitPolicyEnum::CommitFromIncomingFlowFiles) {
- // TODO(beke) setTriggerWhenEmpty(true);
+ context.setTriggerWhenEmpty(true);
}
return MINIFI_STATUS_SUCCESS;
}
diff --git a/extensions/kafka/ExtensionInitializer.cpp b/extensions/kafka/ExtensionInitializer.cpp
index d95a3dbf7..d9d424b80 100644
--- a/extensions/kafka/ExtensionInitializer.cpp
+++ b/extensions/kafka/ExtensionInitializer.cpp
@@ -32,6 +32,6 @@ CEXTENSIONAPI void MinifiInitExtension(minifi_extension_context* extension_conte
.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/llamacpp/ExtensionInitializer.cpp b/extensions/llamacpp/ExtensionInitializer.cpp
index 1152aa7e6..e2ee8b9f6 100644
--- a/extensions/llamacpp/ExtensionInitializer.cpp
+++ b/extensions/llamacpp/ExtensionInitializer.cpp
@@ -33,6 +33,6 @@ CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* extension_con
.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/stable-api-sandbox/ExtensionInitializer.cpp b/extensions/stable-api-sandbox/ExtensionInitializer.cpp
index 4a9040acf..1bf0d123f 100644
--- a/extensions/stable-api-sandbox/ExtensionInitializer.cpp
+++ b/extensions/stable-api-sandbox/ExtensionInitializer.cpp
@@ -32,7 +32,7 @@ CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* extension_con
.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_sandbox::ZooProcessor>(extension);
minifi::api::core::registerControllerServices<minifi::api_sandbox::DogController, minifi::api_sandbox::DuckController>(extension);
}
diff --git a/libminifi/src/minifi-api.cpp b/libminifi/src/minifi-api.cpp
index 82cecd6d6..807e65df1 100644
--- a/libminifi/src/minifi-api.cpp
+++ b/libminifi/src/minifi-api.cpp
@@ -304,7 +304,7 @@ void useCControllerServiceClassDescription(const minifi_controller_service_class
extern "C" {
-minifi_extension* MinifiRegisterExtension(minifi_extension_context* extension_context, const minifi_extension_definition* 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{
@@ -365,7 +365,7 @@ minifi_status minifi_register_controller_service(minifi_extension* extension, co
minifi_extension* MINIFI_REGISTER_EXTENSION_FN(minifi_extension_context* extension_context, const minifi_extension_definition* extension_definition) {
- return MinifiRegisterExtension(extension_context, extension_definition);
+ return minifi_register_extension(extension_context, extension_definition);
}
minifi_status minifi_process_context_set_trigger_when_empty(minifi_process_context* context, bool trigger_when_empty) {
diff --git a/libminifi/test/integration/extension-verification-test/CApiExtension.cpp b/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
index a1ae261b0..efa5d2007 100644
--- a/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
+++ b/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
@@ -49,7 +49,7 @@ CEXTENSIONAPI void minifi_init_extension(minifi_extension_context* extension_con
.deinit = nullptr,
.user_data = nullptr
};
- auto extension = MinifiRegisterExtension(extension_context, &extension_definition);
+ auto extension = minifi_register_extension(extension_context, &extension_definition);
minifi::api::core::useProcessorClassDefinition<DummyCProcessor>([&] (const minifi_processor_class_definition& definition) {
minifi_register_processor(extension, &definition);
});
diff --git a/libminifi/test/integration/extension-verification-test/ExtensionVerificationTests.cpp b/libminifi/test/integration/extension-verification-test/ExtensionVerificationTests.cpp
index 40d358e0d..ab4814c7e 100644
--- a/libminifi/test/integration/extension-verification-test/ExtensionVerificationTests.cpp
+++ b/libminifi/test/integration/extension-verification-test/ExtensionVerificationTests.cpp
@@ -29,7 +29,7 @@
extern "C" {
minifi_extension* MinifiRegisterCppExtension_MatchingBuildId(minifi_extension_context* extension_context, const minifi_extension_definition* extension_definition) {
- return MinifiRegisterExtension(extension_context, extension_definition);
+ return minifi_register_extension(extension_context, extension_definition);
}
} // extern "C"
@@ -93,7 +93,7 @@ TEST_CASE("Can't load c-api extensions with no minifi_init_extension function")
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/minifi-api/include/minifi-c/minifi-api.h b/minifi-api/include/minifi-c/minifi-api.h
index 142c5652f..149e88806 100644
--- a/minifi-api/include/minifi-c/minifi-api.h
+++ b/minifi-api/include/minifi-c/minifi-api.h
@@ -38,7 +38,7 @@ extern "C" {
#define MINIFI_NULLABLE
#ifndef MINIFI_REGISTER_EXTENSION_FN
-#define MINIFI_REGISTER_EXTENSION_FN MinifiRegisterExtension
+#define MINIFI_REGISTER_EXTENSION_FN minifi_register_extension
#endif
/// To declare a processor property that expects an SSLContextService,
@@ -221,7 +221,7 @@ struct minifi_extension_definition {
// 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.
+// 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);
@@ -288,7 +288,7 @@ struct minifi_ssl_data {
};
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 minifi_ssl_data* ssl_data), void* user_ctx);
+ 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,
@@ -296,7 +296,7 @@ enum minifi_proxy_type : uint8_t {
};
struct minifi_proxy_data {
- minifi_proxy_type proxy_type;
+ enum minifi_proxy_type proxy_type;
struct minifi_string_view hostname;
uint16_t port;
struct minifi_string_view* username;
(nifi-minifi-cpp) 01/02: MINIFICPP-2850 Finalize stable API before 1.0
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch beke_2
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 91c53f45804244344e6c56a5c7af32a9a5f0d77c
Author: Martin Zink <ma...@apache.org>
AuthorDate: Tue Jun 2 09:57:36 2026 +0200
MINIFICPP-2850 Finalize stable API before 1.0
---
.clang-format | 9 +-
.github/workflows/create-release-artifacts.yml | 2 +-
Extensions.md | 12 +-
core-framework/include/core/ProcessorImpl.h | 2 +-
.../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 | 17 +-
.../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 | 6 +-
.../include/api/utils/minifi-c-utils.h | 31 +-
.../libtest/CProcessorTestUtils.h | 4 +-
.../mocklib/include/MockProcessContext.h | 8 +-
.../mocklib/include/MockProcessSession.h | 24 +-
.../mocklib/src/MockProcessContext.cpp | 6 +-
.../mocklib/src/MockProcessSession.cpp | 8 +-
.../mocklib/src/mock-minifi-api.cpp | 128 +++++++++
.../mocklib/src/mock-minifi-c.cpp | 142 ----------
.../src/core/ControllerServiceContext.cpp | 4 +-
.../src/core/ControllerServiceImpl.cpp | 2 +-
.../cpp-extension-lib/src/core/ProcessContext.cpp | 53 ++--
.../cpp-extension-lib/src/core/ProcessSession.cpp | 46 +--
.../cpp-extension-lib/src/core/ProcessorImpl.cpp | 5 +-
.../cpp-extension-lib/src/core/logging/Logger.cpp | 12 +-
.../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/gcp/ExtensionInitializer.cpp | 6 +-
.../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 | 2 +-
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 | 4 +-
extensions/kafka/KafkaProcessorBase.h | 2 +-
extensions/kafka/PublishKafka.cpp | 4 +-
extensions/kafka/PublishKafka.h | 4 +-
extensions/kafka/tests/PublishKafkaTests.cpp | 2 +-
extensions/llamacpp/ExtensionInitializer.cpp | 6 +-
.../llamacpp/processors/RunLlamaCppInference.cpp | 7 +-
.../llamacpp/processors/RunLlamaCppInference.h | 5 +-
.../python/pythonlibloader/PythonLibLoader.cpp | 16 +-
extensions/python/pythonloader/PyProcLoader.cpp | 14 +-
extensions/sftp/SFTPLoader.cpp | 10 +-
.../AnimalControllerServices.cpp | 2 +-
.../stable-api-sandbox/AnimalControllerServices.h | 4 +-
.../stable-api-sandbox/ExtensionInitializer.cpp | 6 +-
extensions/stable-api-sandbox/ZooProcessor.cpp | 4 +-
extensions/stable-api-sandbox/ZooProcessor.h | 4 +-
libminifi/include/core/Processor.h | 1 +
libminifi/include/core/extension/Extension.h | 4 +-
libminifi/include/utils/CControllerService.h | 16 +-
libminifi/include/utils/CProcessor.h | 34 ++-
libminifi/src/core/ProcessSession.cpp | 2 +-
libminifi/src/core/Processor.cpp | 4 +
libminifi/src/core/extension/ApiVersion.cpp | 3 +-
libminifi/src/core/extension/Extension.cpp | 18 +-
.../src/core/state/nodes/AgentInformation.cpp | 2 +-
libminifi/src/{minifi-c.cpp => minifi-api.cpp} | 298 +++++++++----------
libminifi/src/utils/CProcessor.cpp | 6 +-
.../extension-verification-test/CApiExtension.cpp | 10 +-
.../CppApiExtension.cpp | 4 +-
.../CreateNotCalled.cpp | 6 +-
.../ExtensionVerificationTests.cpp | 4 +-
.../libtest/unit/ContentRepositoryDependentTests.h | 2 +-
minifi-api/CMakeLists.txt | 2 +-
.../common/include/minifi-cpp/io/StreamCallback.h | 14 +-
minifi-api/include/minifi-c/minifi-api.h | 313 ++++++++++++++++++++
minifi-api/include/minifi-c/minifi-c.h | 314 ---------------------
minifi-api/include/minifi-cpp/core/ProcessorApi.h | 1 +
minifi-api/minifi-api.def | 32 +++
minifi-api/minifi-c-api.def | 33 ---
84 files changed, 992 insertions(+), 995 deletions(-)
diff --git a/.clang-format b/.clang-format
index 1207baf97..970b63023 100644
--- a/.clang-format
+++ b/.clang-format
@@ -5,10 +5,11 @@ AlignConsecutiveAssignments: None
AlignOperands: DontAlign
AllowAllParametersOfDeclarationOnNextLine: true
AllowAllConstructorInitializersOnNextLine: false
-AllowShortBlocksOnASingleLine: Always
-AllowShortCaseLabelsOnASingleLine: true
+AllowShortBlocksOnASingleLine: Empty
+AllowShortCaseLabelsOnASingleLine: false
+AllowShortEnumsOnASingleLine: false
AllowShortFunctionsOnASingleLine: Inline
-AllowShortIfStatementsOnASingleLine: Always
+AllowShortIfStatementsOnASingleLine: WithoutElse
AllowShortLambdasOnASingleLine: All
AllowShortLoopsOnASingleLine: false
AlwaysBreakAfterReturnType: None
@@ -57,4 +58,4 @@ UseTab: Never
BinPackArguments: false # Prevents arguments from being packed onto the same line
PenaltyBreakAssignment: 200
PenaltyBreakOpenParenthesis: 200
-PenaltyReturnTypeOnItsOwnLine: 200
+PenaltyReturnTypeOnItsOwnLine: 200
\ No newline at end of file
diff --git a/.github/workflows/create-release-artifacts.yml b/.github/workflows/create-release-artifacts.yml
index 87d52da9d..e0967a177 100644
--- a/.github/workflows/create-release-artifacts.yml
+++ b/.github/workflows/create-release-artifacts.yml
@@ -95,7 +95,7 @@ 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/minifi-api.def minifi-native-sdk/
cp minifi-api/include/minifi-c/minifi-c.h minifi-native-sdk/
zip -r minifi-native-sdk.zip minifi-native-sdk/
diff --git a/Extensions.md b/Extensions.md
index 0213d68f1..5fec72529 100644
--- a/Extensions.md
+++ b/Extensions.md
@@ -21,26 +21,26 @@ 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`
+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-c.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`, `MinifiProcessorCallbacks::onTrigger`, 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`)
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 `MinifiRegisterExtension` 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) {
+extern "C" void minifi_init_extension(MinifiExtensionContext* extension_context) {
MinifiExtensionDefinition extension_definition{
.name = minifi::api::utils::toStringView(MAKESTRING(EXTENSION_NAME)),
.version = minifi::api::utils::toStringView(MAKESTRING(EXTENSION_VERSION)),
diff --git a/core-framework/include/core/ProcessorImpl.h b/core-framework/include/core/ProcessorImpl.h
index aa4c34006..bceb39b60 100644
--- a/core-framework/include/core/ProcessorImpl.h
+++ b/core-framework/include/core/ProcessorImpl.h
@@ -87,7 +87,7 @@ class ProcessorImpl : public virtual ProcessorApi {
std::string getProcessorType() const override = 0;
- void setTriggerWhenEmpty(bool trigger_when_empty) {
+ void setTriggerWhenEmpty(const bool trigger_when_empty) override {
trigger_when_empty_ = trigger_when_empty;
}
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..a1e528d88 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,34 @@ 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;
};
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;
+
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 8cde8b52b..b0fd44dd1 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_class_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_class_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 {
const std::string_view name_view{interface_name.data, interface_name.length};
@@ -211,16 +198,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_class_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_class_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..abde9430e 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,14 +20,14 @@
#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;
@@ -35,7 +35,7 @@ class CffiLogger : public minifi::core::logging::Logger {
[[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..a8ed3953e 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_class_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,7 +44,7 @@ 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::api::core::useControllerServiceClassDefinition<T>([&](const minifi_controller_service_class_definition& definition) {
minifi::utils::useCControllerServiceClassDescription(definition, [&](const auto&, auto c_description) {
controller_service_impl = std::make_unique<minifi::utils::CControllerService>(std::move(c_description),
metadata,
diff --git a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
index cddcc9761..153068646 100644
--- a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
@@ -29,14 +29,18 @@ 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;
+ }
+
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..829b027e4
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-api.cpp
@@ -0,0 +1,128 @@
+/**
+ * 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_class_definition*) {
+ throw std::runtime_error("Not implemented");
+}
+
+minifi_status minifi_register_controller_service(minifi_extension*, const minifi_controller_service_class_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");
+}
+
+minifi_status minifi_process_context_get_ssl_data(minifi_process_context*, minifi_string_view,
+ void (*)(void* user_ctx, const minifi_ssl_data* ssl_data), 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");
+}
+} // 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..3980a019a 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
@@ -17,6 +17,7 @@
#include "api/core/ProcessContext.h"
+#include "../../../../cmake-build-debug/_deps/llamacpp-src/common/jinja/value.h"
#include "api/core/FlowFile.h"
#include "api/utils/minifi-c-utils.h"
@@ -29,25 +30,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 +59,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,12 +70,9 @@ 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) {
+ 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);
@@ -94,10 +88,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 +113,25 @@ 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 {};
+}
+
+
} // 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..439c31756 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;
@@ -34,7 +34,7 @@ MinifiLogLevel toCLogLevel(minifi::core::logging::LOG_LEVEL lvl) {
gsl_FailFast();
}
-minifi::core::logging::LOG_LEVEL toLogLevel(MinifiLogLevel level) {
+minifi::core::logging::LOG_LEVEL toLogLevel(minifi_log_level level) {
switch (level) {
case MINIFI_LOG_LEVEL_TRACE: return minifi::core::logging::trace;
case MINIFI_LOG_LEVEL_DEBUG: return minifi::core::logging::debug;
@@ -50,19 +50,19 @@ minifi::core::logging::LOG_LEVEL toLogLevel(MinifiLogLevel level) {
} // namespace
void CffiLogger::set_max_log_size(const int size) {
- MinifiLoggerSetMaxLogSize(impl_, size);
+ throw std::runtime_error("Unimplemented C Api");
}
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));
+ return minifi_logger_should_log(impl_, toCLogLevel(level));
}
[[nodiscard]] minifi::core::logging::LOG_LEVEL CffiLogger::level() const {
- return toLogLevel(MinifiLoggerLevel(impl_));
+ throw std::runtime_error("Unimplemented C API");
}
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..a199c9cb2 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-c/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..1fe381f9a 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-c/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 MinifiInitCppExtension(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..b15ebbf8c 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-c/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 MinifiInitCppExtension(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/gcp/ExtensionInitializer.cpp b/extensions/gcp/ExtensionInitializer.cpp
index 89b00e9ff..26014afb4 100644
--- a/extensions/gcp/ExtensionInitializer.cpp
+++ b/extensions/gcp/ExtensionInitializer.cpp
@@ -28,10 +28,10 @@
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,
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..b7b685368 100644
--- a/extensions/gcp/processors/GCSProcessor.h
+++ b/extensions/gcp/processors/GCSProcessor.h
@@ -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..3c1f9ae14 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);
+ // TODO(beke) 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..d95a3dbf7 100644
--- a/extensions/kafka/ExtensionInitializer.cpp
+++ b/extensions/kafka/ExtensionInitializer.cpp
@@ -27,8 +27,8 @@ namespace minifi = org::apache::nifi::minifi;
CEXTENSIONAPI const uint32_t MinifiApiVersion = MINIFI_API_VERSION;
-CEXTENSIONAPI void MinifiInitExtension(MinifiExtensionContext* extension_context) {
- const MinifiExtensionDefinition extension_definition{.name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
+CEXTENSIONAPI void MinifiInitExtension(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};
diff --git a/extensions/kafka/KafkaProcessorBase.h b/extensions/kafka/KafkaProcessorBase.h
index 6cdc2feec..e8ec63817 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 {
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/llamacpp/ExtensionInitializer.cpp b/extensions/llamacpp/ExtensionInitializer.cpp
index 09b142ff2..1152aa7e6 100644
--- a/extensions/llamacpp/ExtensionInitializer.cpp
+++ b/extensions/llamacpp/ExtensionInitializer.cpp
@@ -24,10 +24,10 @@
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,
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/pythonlibloader/PythonLibLoader.cpp b/extensions/python/pythonlibloader/PythonLibLoader.cpp
index e080da22a..6b05a84b3 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-c/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 MinifiInitCppExtension(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..1b8794122 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-c/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 MinifiInitCppExtension(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..ba323e513 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-c/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 MinifiInitCppExtension(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-sandbox/AnimalControllerServices.cpp b/extensions/stable-api-sandbox/AnimalControllerServices.cpp
index a8b1efe9d..4f39fed7a 100644
--- a/extensions/stable-api-sandbox/AnimalControllerServices.cpp
+++ b/extensions/stable-api-sandbox/AnimalControllerServices.cpp
@@ -20,7 +20,7 @@
#include "api/utils/ProcessorConfigUtils.h"
namespace org::apache::nifi::minifi::api_sandbox {
-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-sandbox/AnimalControllerServices.h b/extensions/stable-api-sandbox/AnimalControllerServices.h
index 1ab293396..63c388679 100644
--- a/extensions/stable-api-sandbox/AnimalControllerServices.h
+++ b/extensions/stable-api-sandbox/AnimalControllerServices.h
@@ -41,7 +41,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_; }
@@ -59,7 +59,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-sandbox/ExtensionInitializer.cpp b/extensions/stable-api-sandbox/ExtensionInitializer.cpp
index 52ed52192..4a9040acf 100644
--- a/extensions/stable-api-sandbox/ExtensionInitializer.cpp
+++ b/extensions/stable-api-sandbox/ExtensionInitializer.cpp
@@ -25,10 +25,10 @@
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};
diff --git a/extensions/stable-api-sandbox/ZooProcessor.cpp b/extensions/stable-api-sandbox/ZooProcessor.cpp
index 515712183..d60bad72a 100644
--- a/extensions/stable-api-sandbox/ZooProcessor.cpp
+++ b/extensions/stable-api-sandbox/ZooProcessor.cpp
@@ -22,7 +22,7 @@
namespace org::apache::nifi::minifi::api_sandbox {
-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_sandbox
diff --git a/extensions/stable-api-sandbox/ZooProcessor.h b/extensions/stable-api-sandbox/ZooProcessor.h
index 98ac0b8f5..d7046a03d 100644
--- a/extensions/stable-api-sandbox/ZooProcessor.h
+++ b/extensions/stable-api-sandbox/ZooProcessor.h
@@ -56,8 +56,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_sandbox
diff --git a/libminifi/include/core/Processor.h b/libminifi/include/core/Processor.h
index 5cb2ed9ce..ea43ae846 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();
diff --git a/libminifi/include/core/extension/Extension.h b/libminifi/include/core/extension/Extension.h
index 060c889f0..97b6b565d 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-c/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/utils/CControllerService.h b/libminifi/include/utils/CControllerService.h
index 6e000548d..91c813a44 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-c/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(const minifi_controller_service_class_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..5d3486718 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-c/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)
@@ -115,12 +115,16 @@ class CProcessor : public minifi::core::ProcessorApi {
}
bool getTriggerWhenEmpty() const override {
- return static_cast<bool>(class_description_.callbacks.getTriggerWhenEmpty(impl_));
+ return trigger_when_empty_;
+ }
+
+ void setTriggerWhenEmpty(bool set_trigger_when_empty) override {
+ trigger_when_empty_ = set_trigger_when_empty;
}
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 +136,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 +170,17 @@ class CProcessor : public minifi::core::ProcessorApi {
return metadata_.uuid;
}
+ void reportMetrics(std::vector<minifi::state::PublishedMetric> metrics) { 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_;
+ bool trigger_when_empty_ = false;
+ 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_class_definition& class_description, const std::function<void(ClassDescription, CProcessorClassDescription)>& fn);
} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp
index bd59eeaaa..a377f6206 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-c/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..d09b4df36 100644
--- a/libminifi/src/core/Processor.cpp
+++ b/libminifi/src/core/Processor.cpp
@@ -448,6 +448,10 @@ std::string Processor::getProcessorType() const {
return impl_->getProcessorType();
}
+void Processor::setTriggerWhenEmpty(bool trigger_when_empty) {
+ impl_->setTriggerWhenEmpty(trigger_when_empty);
+}
+
bool Processor::getTriggerWhenEmpty() const {
return impl_->getTriggerWhenEmpty();
}
diff --git a/libminifi/src/core/extension/ApiVersion.cpp b/libminifi/src/core/extension/ApiVersion.cpp
index bf5259420..e6fecb5c0 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-c/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..d0245974c 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-c/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 {
@@ -65,13 +65,13 @@ bool Extension::load(bool global) {
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;
@@ -149,7 +149,7 @@ bool Extension::initialize(const std::shared_ptr<minifi::Configure>& configure)
logger_->log_trace("Initializing extension '{}'", library_name_);
void* init_symbol_ptr = findSymbol("MinifiInitCppExtension");
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/state/nodes/AgentInformation.cpp b/libminifi/src/core/state/nodes/AgentInformation.cpp
index 64423140a..dba3c47ec 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-c/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 70%
rename from libminifi/src/minifi-c.cpp
rename to libminifi/src/minifi-api.cpp
index 68617d522..82cecd6d6 100644
--- a/libminifi/src/minifi-c.cpp
+++ b/libminifi/src/minifi-api.cpp
@@ -15,17 +15,17 @@
* 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-c/minifi-api.h"
#include "minifi-cpp/Exception.h"
-#include "minifi-cpp/controllers/SSLContextServiceInterface.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 +45,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 +66,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 +79,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 +93,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 +101,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 +181,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 +195,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_class_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 +212,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;
@@ -266,12 +266,12 @@ void useCProcessorClassDescription(const MinifiProcessorClassDefinition& class_d
fn(description, c_class_description);
}
-void useCControllerServiceClassDescription(const MinifiControllerServiceClassDefinition& class_description,
+void useCControllerServiceClassDescription(const minifi_controller_service_class_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), "::");
@@ -304,7 +304,7 @@ void useCControllerServiceClassDescription(const MinifiControllerServiceClassDef
extern "C" {
-MinifiExtension* MinifiRegisterExtension(MinifiExtensionContext* extension_context, const MinifiExtensionDefinition* extension_definition) {
+minifi_extension* MinifiRegisterExtension(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 +314,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_class_definition* processor) {
gsl_Assert(extension);
gsl_Assert(processor);
auto* extension_ = reinterpret_cast<minifi::core::extension::Extension*>(extension);
@@ -342,7 +342,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_class_definition* controller_service) {
gsl_Assert(extension);
gsl_Assert(controller_service);
auto extension_info = reinterpret_cast<minifi::core::extension::Extension*>(extension)->getInfo();
@@ -364,15 +364,37 @@ MinifiStatus MinifiRegisterControllerService(MinifiExtension* extension, const M
}
-MinifiExtension* MINIFI_REGISTER_EXTENSION_FN(MinifiExtensionContext* extension_context, const MinifiExtensionDefinition* extension_definition) {
+minifi_extension* MINIFI_REGISTER_EXTENSION_FN(minifi_extension_context* extension_context, const minifi_extension_definition* extension_definition) {
return MinifiRegisterExtension(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 +408,12 @@ MinifiStatus MinifiProcessContextGetProperty(MinifiProcessContext* context, Mini
}
}
-MinifiBool MinifiProcessContextHasNonEmptyProperty(MinifiProcessContext* context, MinifiStringView property_name) {
- gsl_Assert(context != MINIFI_NULL);
+bool minifi_process_context_has_non_empty_property(minifi_process_context* context, minifi_string_view property_name) {
+ gsl_Assert(context);
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 +421,42 @@ void MinifiConfigGet(MinifiExtensionContext* extension_context, MinifiStringView
}
}
-void MinifiLoggerSetMaxLogSize(MinifiLogger* logger, int32_t max_size) {
- gsl_Assert(logger != MINIFI_NULL);
+void MinifiLoggerSetMaxLogSize(minifi_logger* logger, int32_t max_size) {
+ gsl_Assert(logger);
(*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);
+void minifi_logger_log_string(minifi_logger* logger, minifi_log_level level, minifi_string_view msg) {
+ gsl_Assert(logger);
(*reinterpret_cast<std::shared_ptr<minifi::core::logging::Logger>*>(logger))->log_string(toLogLevel(level), toString(msg));
}
-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 +466,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 +478,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 +489,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 +503,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 +517,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 +559,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 +599,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 +620,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 +628,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);
+minifi_status minifi_process_context_get_ssl_data_from_property(minifi_process_context* process_context, minifi_string_view property_name,
+ void (*cb)(void* user_ctx, const 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 +660,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 +676,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 +688,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..75121c2ac 100644
--- a/libminifi/src/utils/CProcessor.cpp
+++ b/libminifi/src/utils/CProcessor.cpp
@@ -20,11 +20,7 @@
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;
+ 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..a1ae261b0 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);
+ minifi::api::core::useProcessorClassDefinition<DummyCProcessor>([&] (const minifi_processor_class_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..22e36ea09 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 MinifiInitCppExtension(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..40d358e0d 100644
--- a/libminifi/test/integration/extension-verification-test/ExtensionVerificationTests.cpp
+++ b/libminifi/test/integration/extension-verification-test/ExtensionVerificationTests.cpp
@@ -28,7 +28,7 @@
extern "C" {
-MinifiExtension* MinifiRegisterCppExtension_MatchingBuildId(MinifiExtensionContext* extension_context, const MinifiExtensionDefinition* extension_definition) {
+minifi_extension* MinifiRegisterCppExtension_MatchingBuildId(minifi_extension_context* extension_context, const minifi_extension_definition* extension_definition) {
return MinifiRegisterExtension(extension_context, extension_definition);
}
} // extern "C"
@@ -88,7 +88,7 @@ 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'"));
}
diff --git a/libminifi/test/libtest/unit/ContentRepositoryDependentTests.h b/libminifi/test/libtest/unit/ContentRepositoryDependentTests.h
index 6db0790b8..903b27a8c 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-c/minifi-api.h"
#include "unit/DummyProcessor.h"
#include "unit/TestBase.h"
diff --git a/minifi-api/CMakeLists.txt b/minifi-api/CMakeLists.txt
index a242392f4..fed77688a 100644
--- a/minifi-api/CMakeLists.txt
+++ b/minifi-api/CMakeLists.txt
@@ -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/io/StreamCallback.h b/minifi-api/common/include/minifi-cpp/io/StreamCallback.h
index ec4fcf59e..78bd3adce 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/include/minifi-c/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-c/minifi-api.h b/minifi-api/include/minifi-c/minifi-api.h
new file mode 100644
index 000000000..142c5652f
--- /dev/null
+++ b/minifi-api/include/minifi-c/minifi-api.h
@@ -0,0 +1,313 @@
+/**
+ * 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_API_H_
+#define MINIFI_API_INCLUDE_MINIFI_C_MINIFI_API_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_OWNED
+#define MINIFI_NULLABLE
+
+#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 (minifi_property_definition::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 (minifi_property_definition::type)
+#define MINIFI_PROXY_CONFIGURATION_SERVICE_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_class_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_class_definition {
+ struct minifi_string_view full_name; // '::'-delimited fully qualified name e.g. 'org::apache::nifi::minifi::extensions::gcp::GCPCredentialsControllerService
+ 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 MinifiRegisterExtension.
+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_class_definition* processor);
+enum minifi_status minifi_register_controller_service(struct minifi_extension* extension,
+ const struct minifi_controller_service_class_definition* controller_service);
+
+enum minifi_status minifi_process_context_set_trigger_when_empty(struct minifi_process_context*, bool);
+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);
+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 size);
+int64_t minifi_output_stream_write(struct minifi_output_stream* stream, const char* data, size_t 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 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 {
+ 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_C_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/ProcessorApi.h b/minifi-api/include/minifi-cpp/core/ProcessorApi.h
index 2ce94f853..7964679a2 100644
--- a/minifi-api/include/minifi-cpp/core/ProcessorApi.h
+++ b/minifi-api/include/minifi-cpp/core/ProcessorApi.h
@@ -51,6 +51,7 @@ class ProcessorApi {
virtual void initialize(ProcessorDescriptor& descriptor) = 0;
virtual bool isSingleThreaded() const = 0;
virtual std::string getProcessorType() const = 0;
+ virtual void setTriggerWhenEmpty(bool trigger_when_empty) = 0;
virtual bool getTriggerWhenEmpty() const = 0;
virtual void onTrigger(ProcessContext&, ProcessSession&) = 0;
virtual void onSchedule(ProcessContext&, ProcessSessionFactory&) = 0;
diff --git a/minifi-api/minifi-api.def b/minifi-api/minifi-api.def
new file mode 100644
index 000000000..dc966971a
--- /dev/null
+++ b/minifi-api/minifi-api.def
@@ -0,0 +1,32 @@
+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_get_ssl_data_from_property
+ minifi_process_context_has_non_empty_property
+ 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_published_metrics_create
+ minifi_register_controller_service
+ minifi_register_extension
+ minifi_register_processor
\ No newline at end of file
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