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/05/05 12:38:22 UTC
(nifi-minifi-cpp) 01/01: MINIFICPP-2798 Mock library for C extensions
This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch MINIFICPP-2798
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 5b9fc339645979bfaa902dec8f99cbc40ebb28b6
Author: Martin Zink <ma...@apache.org>
AuthorDate: Wed Apr 1 13:03:10 2026 +0200
MINIFICPP-2798 Mock library for C extensions
---
.../{ => common}/include/utils/AttributeErrors.h | 0
.../{ => common}/include/utils/RegexUtils.h | 0
.../common}/src/utils/AttributeErrors.cpp | 0
.../{ => common}/src/utils/RegexUtils.cpp | 0
.../cpp-extension-lib/CMakeLists.txt | 2 +-
.../include/api/core/ProcessContext.h | 43 +++++--
.../include/api/core/ProcessSession.h | 55 +++++++--
.../cpp-extension-lib/include/api/core/Resource.h | 50 +++++---
.../include/api/core/logging/Logger.h | 5 +-
.../include/api/utils/ProcessorConfigUtils.h | 32 ++---
.../api/{core/logging/Logger.h => utils/Ssl.h} | 45 ++++---
.../include/api/utils/minifi-c-utils.h | 22 ++--
.../cpp-extension-lib/{ => mocklib}/CMakeLists.txt | 11 +-
.../Logger.h => mocklib/include/MockLogger.h} | 28 ++---
.../mocklib/include/MockProcessContext.h | 46 +++++++
.../mocklib/include/MockProcessSession.h | 82 +++++++++++++
.../mocklib/include/MockStreams.h | 97 +++++++++++++++
.../Logger.h => mocklib/include/MockUtils.h} | 31 ++---
.../mocklib/src/MockProcessContext.cpp | 50 ++++++++
.../mocklib/src/MockProcessSession.cpp | 110 +++++++++++++++++
.../mocklib/src/mock-minifi-c.cpp | 135 +++++++++++++++++++++
.../src/core/ControllerServiceContext.cpp | 2 +-
.../cpp-extension-lib/src/core/ProcessContext.cpp | 67 +++++++---
.../cpp-extension-lib/src/core/ProcessSession.cpp | 48 +++++---
.../cpp-extension-lib/src/core/logging/Logger.cpp | 8 +-
.../extension-verification-test/CApiExtension.cpp | 4 +-
26 files changed, 810 insertions(+), 163 deletions(-)
diff --git a/core-framework/include/utils/AttributeErrors.h b/core-framework/common/include/utils/AttributeErrors.h
similarity index 100%
rename from core-framework/include/utils/AttributeErrors.h
rename to core-framework/common/include/utils/AttributeErrors.h
diff --git a/core-framework/include/utils/RegexUtils.h b/core-framework/common/include/utils/RegexUtils.h
similarity index 100%
rename from core-framework/include/utils/RegexUtils.h
rename to core-framework/common/include/utils/RegexUtils.h
diff --git a/libminifi/src/utils/AttributeErrors.cpp b/core-framework/common/src/utils/AttributeErrors.cpp
similarity index 100%
rename from libminifi/src/utils/AttributeErrors.cpp
rename to core-framework/common/src/utils/AttributeErrors.cpp
diff --git a/core-framework/src/utils/RegexUtils.cpp b/core-framework/common/src/utils/RegexUtils.cpp
similarity index 100%
rename from core-framework/src/utils/RegexUtils.cpp
rename to core-framework/common/src/utils/RegexUtils.cpp
diff --git a/extension-framework/cpp-extension-lib/CMakeLists.txt b/extension-framework/cpp-extension-lib/CMakeLists.txt
index 6b22dd980..f1161279b 100644
--- a/extension-framework/cpp-extension-lib/CMakeLists.txt
+++ b/extension-framework/cpp-extension-lib/CMakeLists.txt
@@ -24,4 +24,4 @@ target_include_directories(minifi-cpp-extension-lib PUBLIC include)
target_link_libraries(minifi-cpp-extension-lib PUBLIC minifi-core-framework-common minifi-c-api)
add_subdirectory(libtest)
-
+add_subdirectory(mocklib)
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 6f012d719..ee9eb0f72 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
@@ -1,5 +1,5 @@
/**
-* Licensed to the Apache Software Foundation (ASF) under one or more
+ * Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
@@ -20,23 +20,48 @@
#include <string>
#include <expected>
+#include "api/core/FlowFile.h"
+#include "api/utils/Ssl.h"
#include "minifi-c.h"
#include "minifi-cpp/core/PropertyDefinition.h"
-#include "api/core/FlowFile.h"
namespace org::apache::nifi::minifi::api::core {
class ProcessContext {
public:
- explicit ProcessContext(MinifiProcessContext* impl): impl_(impl) {}
+ virtual ~ProcessContext() noexcept = default;
+
+ ProcessContext() = default;
+ ProcessContext(const ProcessContext&) = delete;
+ ProcessContext(ProcessContext&&) = delete;
+ ProcessContext& operator=(const ProcessContext&) = delete;
+ ProcessContext& operator=(ProcessContext&&) = delete;
+
+ [[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(std::string_view name,
+ std::string_view type) const = 0;
+ [[nodiscard]] virtual bool hasNonEmptyProperty(std::string_view name) const = 0;
+ [[nodiscard]] virtual std::map<std::string, std::string> getDynamicProperties(const FlowFile* flow_file) const = 0;
+
+ [[nodiscard]] virtual std::expected<utils::net::SslData, std::error_code> getSslData(std::string_view name) const = 0;
+};
+
+class CffiProcessContext : public ProcessContext {
+ public:
+ explicit CffiProcessContext(MinifiProcessContext* impl) : impl_(impl) {}
- std::expected<std::string, std::error_code> getProperty(std::string_view name, const FlowFile* flow_file = nullptr) const;
- std::expected<std::string, std::error_code> getProperty(const minifi::core::PropertyReference& property_reference, const FlowFile* flow_file = nullptr) const {
- return getProperty(property_reference.name, flow_file);
- }
- [[nodiscard]] std::expected<MinifiControllerService*, std::error_code> getControllerService(std::string_view controller_service_name, std::string_view controller_service_class) const;
+ [[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(std::string_view name,
+ std::string_view type) 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]] bool hasNonEmptyProperty(std::string_view name) const;
+ [[nodiscard]] std::expected<utils::net::SslData, std::error_code> getSslData(std::string_view name) const override;
+
+ private:
+ [[nodiscard]] std::expected<std::string, std::error_code> getProperty(std::string_view name, const FlowFile* flow_file) const;
private:
MinifiProcessContext* impl_;
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 e3e812249..1ef50770d 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessSession.h
@@ -29,23 +29,54 @@ namespace org::apache::nifi::minifi::api::core {
class ProcessSession {
public:
- explicit ProcessSession(MinifiProcessSession* impl): impl_(impl) {}
+ virtual ~ProcessSession() = default;
- FlowFile create(const FlowFile* parent = nullptr);
- FlowFile get();
- void transfer(FlowFile ff, const minifi::core::Relationship& relationship);
- void remove(FlowFile ff);
- void write(FlowFile& flow, const io::OutputStreamCallback& callback);
- void read(FlowFile& flow, const io::InputStreamCallback& callback);
+ ProcessSession() = default;
- void setAttribute(FlowFile& ff, std::string_view key, std::string value);
- void removeAttribute(FlowFile& ff, std::string_view key);
- std::optional<std::string> getAttribute(FlowFile& ff, std::string_view key);
- std::map<std::string, std::string> getAttributes(FlowFile& ff);
+ ProcessSession(const ProcessSession&) = delete;
+ ProcessSession(ProcessSession&&) = delete;
+ ProcessSession& operator=(const ProcessSession&) = delete;
+ ProcessSession& operator=(ProcessSession&&) = delete;
+
+ virtual FlowFile create(const FlowFile* parent = nullptr) = 0;
+ virtual FlowFile get() = 0;
+
+ virtual void penalize(FlowFile& ff) = 0;
+ virtual void transfer(FlowFile ff, const minifi::core::Relationship& relationship) = 0;
+ virtual void remove(FlowFile ff) = 0;
+ virtual void write(FlowFile& flow, const io::OutputStreamCallback& callback) = 0;
+ virtual void read(FlowFile& flow, const io::InputStreamCallback& callback) = 0;
+
+ virtual void setAttribute(FlowFile& ff, std::string_view key, std::string value) = 0;
+ virtual void removeAttribute(FlowFile& ff, std::string_view key) = 0;
+ [[nodiscard]] virtual std::optional<std::string> getAttribute(FlowFile& ff, std::string_view key) = 0;
+ [[nodiscard]] virtual std::map<std::string, std::string> getAttributes(const FlowFile& ff) const = 0;
+ [[nodiscard]] virtual std::string getFlowFileId(const FlowFile& ff) const = 0;
+ [[nodiscard]] virtual uint64_t getFlowFileSize(const FlowFile& ff) const = 0;
void writeBuffer(FlowFile& flow_file, std::span<const char> buffer);
void writeBuffer(FlowFile& flow_file, std::span<const std::byte> buffer);
- std::vector<std::byte> readBuffer(FlowFile& flow_file);
+ [[nodiscard]] std::vector<std::byte> readBuffer(FlowFile& flow_file);
+};
+
+class CffiProcessSession : public ProcessSession {
+ public:
+ explicit CffiProcessSession(MinifiProcessSession* impl): impl_(impl) {}
+
+ FlowFile create(const FlowFile* parent = nullptr) override;
+ FlowFile get() override;
+ void penalize(FlowFile& ff) override;
+ void transfer(FlowFile ff, const minifi::core::Relationship& relationship) override;
+ void remove(FlowFile ff) override;
+ void write(FlowFile& flow, const io::OutputStreamCallback& callback) override;
+ void read(FlowFile& flow, const io::InputStreamCallback& callback) override;
+
+ void setAttribute(FlowFile& ff, std::string_view key, std::string value) override;
+ void removeAttribute(FlowFile& ff, std::string_view key) override;
+ [[nodiscard]] std::optional<std::string> getAttribute(FlowFile& ff, std::string_view key) override;
+ [[nodiscard]] std::map<std::string, std::string> getAttributes(const FlowFile& ff) const override;
+ [[nodiscard]] std::string getFlowFileId(const FlowFile& ff) const override;
+ [[nodiscard]] uint64_t getFlowFileSize(const FlowFile& ff) const override;
private:
MinifiProcessSession* impl_;
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 8187605e5..60766741c 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/Resource.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/Resource.h
@@ -49,17 +49,17 @@ void useProcessorClassDefinition(Fn&& fn) {
std::vector<MinifiDynamicPropertyDefinition> dynamic_properties;
for (auto& prop : Class::DynamicProperties) {
dynamic_properties.push_back(MinifiDynamicPropertyDefinition {
- .name = utils::toStringView(prop.name),
- .value = utils::toStringView(prop.value),
- .description = utils::toStringView(prop.description),
+ .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;
for (auto& rel : Class::Relationships) {
relationships.push_back(MinifiRelationshipDefinition{
- .name = utils::toStringView(rel.name),
- .description = utils::toStringView(rel.description)
+ .name = utils::minifiStringView(rel.name),
+ .description = utils::minifiStringView(rel.description)
});
}
std::vector<std::vector<MinifiStringView>> attribute_relationships_cache;
@@ -67,20 +67,20 @@ void useProcessorClassDefinition(Fn&& fn) {
for (auto& attr : Class::OutputAttributes) {
std::vector<MinifiStringView> rel_cache;
for (auto& rel : attr.relationships) {
- rel_cache.push_back(utils::toStringView(rel.name));
+ rel_cache.push_back(utils::minifiStringView(rel.name));
}
output_attributes.push_back(MinifiOutputAttributeDefinition {
- .name = utils::toStringView(attr.name),
+ .name = utils::minifiStringView(attr.name),
.relationships_count = gsl::narrow<uint32_t>(attr.relationships.size()),
.relationships_ptr = rel_cache.data(),
- .description = utils::toStringView(attr.description)
+ .description = utils::minifiStringView(attr.description)
});
attribute_relationships_cache.push_back(std::move(rel_cache));
}
MinifiProcessorClassDefinition definition{
- .full_name = utils::toStringView(full_name),
- .description = utils::toStringView(Class::Description),
+ .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(),
.dynamic_properties_count = gsl::narrow<uint32_t>(dynamic_properties.size()),
@@ -100,7 +100,7 @@ void useProcessorClassDefinition(Fn&& fn) {
return new Class{minifi::core::ProcessorMetadata{
.uuid = minifi::utils::Identifier::parse(std::string{metadata.uuid.data, metadata.uuid.length}).value(),
.name = std::string{metadata.name.data, metadata.name.length},
- .logger = std::make_shared<logging::Logger>(metadata.logger)}};
+ .logger = std::make_shared<logging::CffiLogger>(metadata.logger)}};
} catch (...) { return nullptr; }
},
.destroy = [] (MINIFI_OWNED void* self) -> void {
@@ -110,8 +110,8 @@ void useProcessorClassDefinition(Fn&& fn) {
return static_cast<Class*>(self)->getTriggerWhenEmpty();
},
.onTrigger = [] (void* self, MinifiProcessContext* context, MinifiProcessSession* session) -> MinifiStatus {
- ProcessContext context_wrapper(context);
- ProcessSession session_wrapper(session);
+ CffiProcessContext context_wrapper(context);
+ CffiProcessSession session_wrapper(session);
try {
return static_cast<Class*>(self)->onTrigger(context_wrapper, session_wrapper);
} catch (...) {
@@ -119,7 +119,7 @@ void useProcessorClassDefinition(Fn&& fn) {
}
},
.onSchedule = [] (void* self, MinifiProcessContext* context) -> MinifiStatus {
- ProcessContext context_wrapper(context);
+ CffiProcessContext context_wrapper(context);
try {
return static_cast<Class*>(self)->onSchedule(context_wrapper);
} catch (...) {
@@ -136,7 +136,7 @@ void useProcessorClassDefinition(Fn&& fn) {
std::vector<MinifiStringView> names;
std::vector<double> values;
for (auto& [name, val] : metrics) {
- names.push_back(utils::toStringView(name));
+ names.push_back(utils::minifiStringView(name));
values.push_back(val);
}
return MinifiPublishedMetricsCreate(gsl::narrow<uint32_t>(metrics.size()), names.data(), values.data());
@@ -155,8 +155,8 @@ void useControllerServiceClassDefinition(Fn&& fn) {
std::vector<MinifiPropertyDefinition> class_properties = utils::toProperties(Class::Properties, string_vector_cache);
- MinifiControllerServiceClassDefinition definition{.full_name = utils::toStringView(full_name),
- .description = utils::toStringView(Class::Description),
+ MinifiControllerServiceClassDefinition 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(),
@@ -166,7 +166,7 @@ void useControllerServiceClassDefinition(Fn&& fn) {
return new Class{minifi::core::ControllerServiceMetadata{
.uuid = minifi::utils::Identifier::parse(std::string{metadata.uuid.data, metadata.uuid.length}).value(),
.name = std::string{metadata.name.data, metadata.name.length},
- .logger = std::make_shared<logging::Logger>(metadata.logger)}};
+ .logger = std::make_shared<logging::CffiLogger>(metadata.logger)}};
} catch (...) { return nullptr; }
},
.destroy = [](MINIFI_OWNED void* self) -> void { delete static_cast<Class*>(self); },
@@ -186,4 +186,18 @@ void useControllerServiceClassDefinition(Fn&& fn) {
fn(definition);
}
+template <typename... Processors>
+void registerProcessors(MinifiExtension* extension) {
+ (core::useProcessorClassDefinition<Processors>([&](const MinifiProcessorClassDefinition& definition) {
+ MinifiRegisterProcessor(extension, &definition);
+ }), ...);
+}
+
+template <typename... ControllerServices>
+void registerControllerServices(MinifiExtension* extension) {
+ (core::useControllerServiceClassDefinition<ControllerServices>([&](const MinifiControllerServiceClassDefinition& definition) {
+ MinifiRegisterControllerService(extension, &definition);
+ }), ...);
+}
+
} // namespace org::apache::nifi::minifi::api::core
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 ada9e5178..5a4401344 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,15 +20,14 @@
#include <iostream>
#include <string>
-#include "fmt/chrono.h"
#include "minifi-c.h"
#include "minifi-cpp/core/logging/Logger.h"
namespace org::apache::nifi::minifi::api::core::logging {
-class Logger : public minifi::core::logging::Logger {
+class CffiLogger : public minifi::core::logging::Logger {
public:
- explicit Logger(MinifiLogger* impl): impl_(impl) {}
+ explicit CffiLogger(MinifiLogger* impl): impl_(impl) {}
void set_max_log_size(int size) override;
void log_string(minifi::core::logging::LOG_LEVEL level, std::string str) override;
diff --git a/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h b/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
index a73dd7549..db44e1cd5 100644
--- a/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
+++ b/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
@@ -32,47 +32,47 @@
namespace org::apache::nifi::minifi::api::utils {
inline std::string parseProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- return ctx.getProperty(property.name, flow_file)
+ return ctx.getProperty(property, flow_file)
| minifi::utils::orThrow(fmt::format("Expected valid value from \"{}\"", property.name));
}
inline bool parseBoolProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- return ctx.getProperty(property.name, flow_file)
+ return ctx.getProperty(property, flow_file)
| minifi::utils::andThen(parsing::parseBool)
| minifi::utils::orThrow(fmt::format("Expected parsable bool from \"{}\"", property.name));
}
inline uint64_t parseU64Property(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- return ctx.getProperty(property.name, flow_file)
+ return ctx.getProperty(property, flow_file)
| minifi::utils::andThen(parsing::parseIntegral<uint64_t>)
| minifi::utils::orThrow(fmt::format("Expected parsable uint64_t from \"{}\"", property.name));
}
inline int64_t parseI64Property(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- return ctx.getProperty(property.name, flow_file)
+ return ctx.getProperty(property, flow_file)
| minifi::utils::andThen(parsing::parseIntegral<int64_t>)
| minifi::utils::orThrow(fmt::format("Expected parsable int64_t from \"{}\"", property.name));
}
inline std::chrono::milliseconds parseDurationProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- return ctx.getProperty(property.name, flow_file)
+ return ctx.getProperty(property, flow_file)
| minifi::utils::andThen(parsing::parseDuration<std::chrono::milliseconds>)
| minifi::utils::orThrow(fmt::format("Expected parsable duration from \"{}\"", property.name));
}
inline uint64_t parseDataSizeProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- return ctx.getProperty(property.name, flow_file)
+ return ctx.getProperty(property, flow_file)
| minifi::utils::andThen(parsing::parseDataSize)
| minifi::utils::orThrow(fmt::format("Expected parsable data size from \"{}\"", property.name));
}
inline std::optional<std::string> parseOptionalProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- return ctx.getProperty(property.name, flow_file)
+ return ctx.getProperty(property, flow_file)
| minifi::utils::toOptional();
}
inline std::optional<bool> parseOptionalBoolProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- if (const auto property_str = ctx.getProperty(property.name, flow_file)) {
+ if (const auto property_str = ctx.getProperty(property, flow_file)) {
return parsing::parseBool(*property_str)
| minifi::utils::orThrow(fmt::format("Expected parsable bool from \"{}\"", property.name));
}
@@ -80,7 +80,7 @@ inline std::optional<bool> parseOptionalBoolProperty(const core::ProcessContext&
}
inline std::optional<uint64_t> parseOptionalU64Property(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- if (const auto property_str = ctx.getProperty(property.name, flow_file)) {
+ if (const auto property_str = ctx.getProperty(property, flow_file)) {
if (property_str->empty()) {
return std::nullopt;
}
@@ -92,7 +92,7 @@ inline std::optional<uint64_t> parseOptionalU64Property(const core::ProcessConte
}
inline std::optional<int64_t> parseOptionalI64Property(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- if (const auto property_str = ctx.getProperty(property.name, flow_file)) {
+ if (const auto property_str = ctx.getProperty(property, flow_file)) {
if (property_str->empty()) {
return std::nullopt;
}
@@ -106,7 +106,7 @@ inline std::optional<int64_t> parseOptionalI64Property(const core::ProcessContex
inline std::optional<std::chrono::milliseconds> parseOptionalDurationProperty(const core::ProcessContext& ctx,
const minifi::core::PropertyReference& property,
const core::FlowFile* flow_file = nullptr) {
- if (const auto property_str = ctx.getProperty(property.name, flow_file)) {
+ if (const auto property_str = ctx.getProperty(property, flow_file)) {
if (property_str->empty()) {
return std::nullopt;
}
@@ -118,7 +118,7 @@ inline std::optional<std::chrono::milliseconds> parseOptionalDurationProperty(co
}
inline std::optional<uint64_t> parseOptionalDataSizeProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- if (const auto property_str = ctx.getProperty(property.name, flow_file)) {
+ if (const auto property_str = ctx.getProperty(property, flow_file)) {
if (property_str->empty()) {
return std::nullopt;
}
@@ -130,7 +130,7 @@ inline std::optional<uint64_t> parseOptionalDataSizeProperty(const core::Process
}
inline std::optional<float> parseOptionalFloatProperty(const core::ProcessContext& ctx, const minifi::core::PropertyReference& property, const core::FlowFile* flow_file = nullptr) {
- if (const auto property_str = ctx.getProperty(property.name, flow_file)) {
+ if (const auto property_str = ctx.getProperty(property, flow_file)) {
if (property_str->empty()) {
return std::nullopt;
}
@@ -142,7 +142,7 @@ inline std::optional<float> parseOptionalFloatProperty(const core::ProcessContex
template<typename T>
T parseEnumProperty(const core::ProcessContext& context, const minifi::core::PropertyReference& prop, const core::FlowFile* flow_file = nullptr) {
- const auto enum_str = context.getProperty(prop.name, flow_file);
+ const auto enum_str = context.getProperty(prop, flow_file);
if (!enum_str) {
throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Property '" + std::string(prop.name) + "' is missing");
}
@@ -155,7 +155,7 @@ T parseEnumProperty(const core::ProcessContext& context, const minifi::core::Pro
template<typename T>
std::optional<T> parseOptionalEnumProperty(const core::ProcessContext& context, const minifi::core::PropertyReference& prop) {
- const auto enum_str = context.getProperty(prop.name);
+ const auto enum_str = context.getProperty(prop, nullptr);
if (!enum_str) {
return std::nullopt;
@@ -169,7 +169,7 @@ std::optional<T> parseOptionalEnumProperty(const core::ProcessContext& context,
template<typename ControllerServiceType>
ControllerServiceType* parseOptionalControllerService(const core::ProcessContext& context, const minifi::core::PropertyReference& prop) {
- const auto controller_service_name = context.getProperty(prop.name);
+ const auto controller_service_name = context.getProperty(prop, nullptr);
if (!controller_service_name || controller_service_name->empty()) {
return nullptr;
}
diff --git a/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h b/extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
similarity index 53%
copy from extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
copy to extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
index ada9e5178..eba0e1670 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
+++ b/extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
@@ -1,5 +1,4 @@
/**
- *
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
@@ -17,26 +16,40 @@
*/
#pragma once
-#include <iostream>
#include <string>
+#include <memory>
+#include <optional>
+#include <filesystem>
+
+#include "utils/Enum.h"
+
+namespace org::apache::nifi::minifi::api::utils::net {
-#include "fmt/chrono.h"
-#include "minifi-c.h"
-#include "minifi-cpp/core/logging/Logger.h"
+enum class ClientAuthOption {
+ NONE,
+ WANT,
+ REQUIRED
+};
-namespace org::apache::nifi::minifi::api::core::logging {
+struct SslData {
+ std::filesystem::path ca_loc;
+ std::filesystem::path cert_loc;
+ std::filesystem::path key_loc;
+ std::string key_pw;
-class Logger : public minifi::core::logging::Logger {
- public:
- explicit Logger(MinifiLogger* impl): impl_(impl) {}
+ bool isValid() const {
+ return !cert_loc.empty() && !key_loc.empty();
+ }
+};
- void set_max_log_size(int size) override;
- void log_string(minifi::core::logging::LOG_LEVEL level, std::string str) override;
- bool should_log(minifi::core::logging::LOG_LEVEL level) override;
- [[nodiscard]] minifi::core::logging::LOG_LEVEL level() const override;
+struct SslServerOptions {
+ SslData cert_data;
+ ClientAuthOption client_auth_option;
- private:
- MinifiLogger* impl_;
+ SslServerOptions(SslData cert_data, ClientAuthOption client_auth_option)
+ : cert_data(cert_data),
+ client_auth_option(client_auth_option) {}
};
-} // namespace org::apache::nifi::minifi::api::core::logging
+
+} // namespace org::apache::nifi::minifi::api::utils::net
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 43fb28306..227ad4c80 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
@@ -32,10 +32,18 @@
namespace org::apache::nifi::minifi::api::utils {
-inline MinifiStringView toStringView(std::string_view str) {
+inline MinifiStringView minifiStringView(const std::string_view str) {
return MinifiStringView{.data = str.data(), .length = str.length()};
}
+inline std::string toString(const MinifiStringView sv) {
+ return {sv.data, sv.length};
+}
+
+inline std::string_view toStringView(const MinifiStringView sv) {
+ return {sv.data, sv.length};
+}
+
template<typename T>
std::string classNameWithDots() {
std::string class_name{minifi::core::className<T>()};
@@ -59,27 +67,27 @@ inline std::vector<MinifiPropertyDefinition> toProperties(std::span<const minifi
std::vector<MinifiStringView> sv_cache;
const size_t allowed_values_begin = sv_cache.size();
for (auto& allowed_value : prop.allowed_values) {
- sv_cache.emplace_back(toStringView(allowed_value));
+ sv_cache.emplace_back(minifiStringView(allowed_value));
}
const std::optional<size_t> allowed_types_begin = [&] () -> std::optional<size_t> {
if (prop.allowed_types.empty()) {
return std::nullopt;
}
gsl_Expects(prop.allowed_types.size() == 1);
- sv_cache.emplace_back(toStringView(prop.allowed_types[0]));
+ sv_cache.emplace_back(minifiStringView(prop.allowed_types[0]));
return sv_cache.size() - 1;
}();
const std::optional<size_t> default_value_begin = [&] () -> std::optional<size_t> {
if (!prop.default_value) {
return std::nullopt;
}
- sv_cache.emplace_back(toStringView(*prop.default_value));
+ sv_cache.emplace_back(minifiStringView(*prop.default_value));
return sv_cache.size() - 1;
}();
properties.push_back(MinifiPropertyDefinition{
- .name = toStringView(prop.name),
- .display_name = toStringView(prop.display_name),
- .description = toStringView(prop.description),
+ .name = minifiStringView(prop.name),
+ .display_name = minifiStringView(prop.display_name),
+ .description = minifiStringView(prop.description),
.is_required = prop.is_required,
.is_sensitive = prop.is_sensitive,
diff --git a/extension-framework/cpp-extension-lib/CMakeLists.txt b/extension-framework/cpp-extension-lib/mocklib/CMakeLists.txt
similarity index 70%
copy from extension-framework/cpp-extension-lib/CMakeLists.txt
copy to extension-framework/cpp-extension-lib/mocklib/CMakeLists.txt
index 6b22dd980..f93870f37 100644
--- a/extension-framework/cpp-extension-lib/CMakeLists.txt
+++ b/extension-framework/cpp-extension-lib/mocklib/CMakeLists.txt
@@ -17,11 +17,8 @@
# under the License.
#
-file(GLOB SOURCES src/core/*.cpp src/core/logging/*.cpp src/utils/*.cpp)
-
-add_minifi_library(minifi-cpp-extension-lib STATIC ${SOURCES})
-target_include_directories(minifi-cpp-extension-lib PUBLIC include)
-target_link_libraries(minifi-cpp-extension-lib PUBLIC minifi-core-framework-common minifi-c-api)
-
-add_subdirectory(libtest)
+file(GLOB SRC_FILES CONFIGURE_DEPENDS "src/*.cpp")
+add_library(libmock-minifi STATIC ${SRC_FILES})
+target_link_libraries(libmock-minifi PUBLIC minifi-cpp-extension-lib minifi-core-framework-common minifi-c-api)
+target_include_directories(libmock-minifi PUBLIC include)
diff --git a/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h b/extension-framework/cpp-extension-lib/mocklib/include/MockLogger.h
similarity index 57%
copy from extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
copy to extension-framework/cpp-extension-lib/mocklib/include/MockLogger.h
index ada9e5178..56be27b96 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockLogger.h
@@ -1,5 +1,4 @@
/**
- *
* 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.
@@ -15,28 +14,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
#pragma once
-#include <iostream>
+#include <map>
#include <string>
-#include "fmt/chrono.h"
-#include "minifi-c.h"
#include "minifi-cpp/core/logging/Logger.h"
-namespace org::apache::nifi::minifi::api::core::logging {
-
-class Logger : public minifi::core::logging::Logger {
+namespace org::apache::nifi::minifi::mock {
+class MockLogger : public core::logging::Logger {
public:
- explicit Logger(MinifiLogger* impl): impl_(impl) {}
+ void set_max_log_size(int) override {}
+ void log_string(const core::logging::LOG_LEVEL level, std::string s) override { logs_[level].emplace_back(std::move(s)); }
+ [[nodiscard]] bool should_log(const core::logging::LOG_LEVEL level) override { return level > log_level_; }
+ [[nodiscard]] core::logging::LOG_LEVEL level() const override { return log_level_; }
- void set_max_log_size(int size) override;
- void log_string(minifi::core::logging::LOG_LEVEL level, std::string str) override;
- bool should_log(minifi::core::logging::LOG_LEVEL level) override;
- [[nodiscard]] minifi::core::logging::LOG_LEVEL level() const override;
-
- private:
- MinifiLogger* impl_;
+ core::logging::LOG_LEVEL log_level_ = core::logging::LOG_LEVEL::trace;
+ std::map<core::logging::LOG_LEVEL, std::vector<std::string>> logs_;
};
-
-} // namespace org::apache::nifi::minifi::api::core::logging
+} // namespace org::apache::nifi::minifi::mock
diff --git a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
new file mode 100644
index 000000000..62a0bc57a
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+#include <map>
+#include <system_error>
+
+#include "MockUtils.h"
+#include "api/core/ProcessContext.h"
+
+namespace org::apache::nifi::minifi::mock {
+class MockProcessContext : public api::core::ProcessContext {
+ public:
+ using ProcessContext::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(std::string_view controller_service_name,
+ std::string_view controller_service_class) 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<api::utils::net::SslData, std::error_code> getSslData(std::string_view name) const override;
+
+ std::map<std::string, std::string, std::less<>> properties_;
+
+ private:
+ [[nodiscard]] std::expected<std::string, std::error_code> getProperty(std::string_view name,
+ const api::core::FlowFile* flow_file) const;
+};
+
+} // namespace org::apache::nifi::minifi::mock
diff --git a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h
new file mode 100644
index 000000000..fa53671a9
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h
@@ -0,0 +1,82 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+#include <map>
+#include <ranges>
+
+#include "MockUtils.h"
+#include "api/core/ProcessSession.h"
+#include "range/v3/range/conversion.hpp"
+#include "range/v3/view/transform.hpp"
+
+namespace org::apache::nifi::minifi::mock {
+struct MockFlowFileData {
+ explicit MockFlowFileData(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>>();
+ }
+
+ ~MockFlowFileData() = default;
+
+ MockFlowFileData() = default;
+ MockFlowFileData(const MockFlowFileData&) = default;
+ MockFlowFileData(MockFlowFileData&&) = default;
+ MockFlowFileData& operator=(const MockFlowFileData&) = default;
+ MockFlowFileData& operator=(MockFlowFileData&&) = default;
+
+ std::map<std::string, std::string> attributes;
+ std::vector<std::byte> content;
+ bool is_penalized = false;
+ std::string id;
+};
+
+class MockProcessSession : public api::core::ProcessSession {
+ public:
+ MockProcessSession() = default;
+ MockProcessSession(const MockProcessSession&) = delete;
+ MockProcessSession& operator=(const MockProcessSession&) = delete;
+ MockProcessSession(const MockProcessSession&&) = delete;
+ MockProcessSession& operator=(const MockProcessSession&&) = delete;
+
+ ~MockProcessSession() override;
+
+ api::core::FlowFile create(const api::core::FlowFile* parent) override;
+ api::core::FlowFile get() override;
+ void penalize(api::core::FlowFile& ff) override;
+ void transfer(api::core::FlowFile ff, const minifi::core::Relationship& relationship) override;
+ void remove(api::core::FlowFile ff) override;
+ void write(api::core::FlowFile& ff, const io::OutputStreamCallback& callback) override;
+ void read(api::core::FlowFile& ff, const io::InputStreamCallback& callback) override;
+ void setAttribute(api::core::FlowFile& ff, std::string_view key, std::string value) override;
+ void removeAttribute(api::core::FlowFile& ff, std::string_view key) override;
+ std::optional<std::string> getAttribute(api::core::FlowFile& ff, std::string_view key) override;
+ [[nodiscard]] std::map<std::string, std::string> getAttributes(const api::core::FlowFile& ff) const override;
+ [[nodiscard]] std::string getFlowFileId(const api::core::FlowFile& ff) const override;
+ [[nodiscard]] uint64_t getFlowFileSize(const api::core::FlowFile& ff) const override;
+
+ void addInputFlowFile(MockFlowFileData flow_file_data);
+
+ private:
+ std::vector<api::core::FlowFile> input_flow_files_;
+ std::map<MinifiFlowFile*, MockFlowFileData> flow_file_datas;
+ std::map<std::string, std::vector<api::core::FlowFile>> transferred_flow_files_;
+ std::vector<api::core::FlowFile> removed_flow_files_;
+};
+
+} // namespace org::apache::nifi::minifi::mock
diff --git a/extension-framework/cpp-extension-lib/mocklib/include/MockStreams.h b/extension-framework/cpp-extension-lib/mocklib/include/MockStreams.h
new file mode 100644
index 000000000..73352daf7
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockStreams.h
@@ -0,0 +1,97 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+#pragma once
+
+#include <algorithm>
+#include <cstddef>
+#include <span>
+#include <vector>
+
+#include "minifi-cpp/io/InputStream.h"
+#include "minifi-cpp/io/OutputStream.h"
+
+namespace org::apache::nifi::minifi::mock {
+
+class MockOutputStream : public io::OutputStream {
+ public:
+ explicit MockOutputStream(std::vector<std::byte>& container) : container_(container) {}
+
+ size_t write(const uint8_t* value, const size_t len) override {
+ if (len == 0 || value == nullptr) { return 0; }
+
+ if (offset_ + len > container_.size()) { container_.resize(offset_ + len); }
+
+ std::memcpy(container_.data() + offset_, value, len);
+ offset_ += len;
+ return len;
+ }
+
+ void seek(const size_t offset) override { offset_ = offset; }
+
+ [[nodiscard]] size_t tell() const override { return offset_; }
+
+ [[nodiscard]] std::span<const std::byte> getBuffer() const override { return {container_.data(), container_.size()}; }
+
+ int initialize() override {
+ offset_ = 0;
+ return 0;
+ }
+
+ void close() override {}
+
+ private:
+ std::vector<std::byte>& container_;
+ size_t offset_ = 0;
+};
+
+class MockInputStream : public io::InputStream {
+ public:
+ explicit MockInputStream(const std::vector<std::byte>& container) : container_(container) {}
+
+ size_t read(std::span<std::byte> out_buffer) override {
+ if (out_buffer.empty() || offset_ >= container_.size()) { return 0; }
+
+ const size_t readable = container_.size() - offset_;
+ const size_t len = std::min(out_buffer.size(), readable);
+
+ std::memcpy(out_buffer.data(), container_.data() + offset_, len);
+ offset_ += len;
+ return len;
+ }
+
+ void seek(const size_t offset) override { offset_ = std::min(offset, container_.size()); }
+
+ [[nodiscard]] size_t tell() const override { return offset_; }
+
+ [[nodiscard]] size_t size() const override { return container_.size(); }
+
+ [[nodiscard]] std::span<const std::byte> getBuffer() const override { return {container_.data(), container_.size()}; }
+
+ int initialize() override {
+ offset_ = 0;
+ return 0;
+ }
+
+ void close() override {}
+
+ private:
+ const std::vector<std::byte>& container_;
+ size_t offset_ = 0;
+};
+
+} // namespace org::apache::nifi::minifi::mock
diff --git a/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h b/extension-framework/cpp-extension-lib/mocklib/include/MockUtils.h
similarity index 56%
copy from extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
copy to extension-framework/cpp-extension-lib/mocklib/include/MockUtils.h
index ada9e5178..471df8189 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/logging/Logger.h
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockUtils.h
@@ -1,5 +1,4 @@
/**
- *
* 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.
@@ -15,28 +14,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+
#pragma once
-#include <iostream>
#include <string>
-#include "fmt/chrono.h"
-#include "minifi-c.h"
-#include "minifi-cpp/core/logging/Logger.h"
-
-namespace org::apache::nifi::minifi::api::core::logging {
-
-class Logger : public minifi::core::logging::Logger {
- public:
- explicit Logger(MinifiLogger* impl): impl_(impl) {}
+#include "MockLogger.h"
+#include "minifi-cpp/core/ProcessorMetadata.h"
- void set_max_log_size(int size) override;
- void log_string(minifi::core::logging::LOG_LEVEL level, std::string str) override;
- bool should_log(minifi::core::logging::LOG_LEVEL level) override;
- [[nodiscard]] minifi::core::logging::LOG_LEVEL level() const override;
+struct MinifiFlowFile{};
- private:
- MinifiLogger* impl_;
-};
+namespace org::apache::nifi::minifi::mock {
+inline core::ProcessorMetadata getMockMetadata() {
+ return core::ProcessorMetadata{.uuid = utils::Identifier{}, .name = "Processor", .logger = std::make_shared<MockLogger>()};
+}
-} // namespace org::apache::nifi::minifi::api::core::logging
+inline core::ProcessorMetadata metadataWithLogger(std::shared_ptr<core::logging::Logger> logger) {
+ return core::ProcessorMetadata{.uuid = utils::Identifier{}, .name = "Processor", .logger = std::move(logger)};
+}
+} // 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
new file mode 100644
index 000000000..b8bb1e5fc
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp
@@ -0,0 +1,50 @@
+/**
+ * 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 "MockProcessContext.h"
+
+#include "utils/PropertyErrors.h"
+
+namespace org::apache::nifi::minifi::mock {
+std::expected<std::string, std::error_code> MockProcessContext::getProperty(const std::string_view name, const api::core::FlowFile*) const {
+ if (!properties_.contains(name)) { return std::unexpected{make_error_code(core::PropertyErrorCode::PropertyNotSet)}; }
+ return properties_.at(std::string(name));
+}
+
+std::expected<std::string, std::error_code> MockProcessContext::getProperty(const minifi::core::PropertyReference& property_reference,
+ const api::core::FlowFile* flow_file) const {
+ if (auto property = getProperty(property_reference.name, flow_file)) { return property; }
+ if (property_reference.default_value) { return std::string{*property_reference.default_value}; }
+ return std::unexpected{make_error_code(core::PropertyErrorCode::PropertyNotSet)};
+}
+
+std::expected<MinifiControllerService*, std::error_code> MockProcessContext::getControllerService(std::string_view, std::string_view) const {
+ return nullptr;
+}
+
+std::map<std::string, std::string> MockProcessContext::getDynamicProperties(const api::core::FlowFile*) const {
+ return {};
+}
+
+bool MockProcessContext::hasNonEmptyProperty(const std::string_view name) const {
+ return properties_.contains(name);
+}
+
+std::expected<api::utils::net::SslData, std::error_code> MockProcessContext::getSslData(std::string_view) const {
+ return api::utils::net::SslData{};
+}
+} // namespace org::apache::nifi::minifi::mock
diff --git a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp
new file mode 100644
index 000000000..2329808f7
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp
@@ -0,0 +1,110 @@
+/**
+ * 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 "MockProcessSession.h"
+
+#include "MockStreams.h"
+
+namespace org::apache::nifi::minifi::mock {
+
+MockProcessSession::~MockProcessSession() {
+ // api::core::FlowFile requires that somebody takes ownership before it goes out of scope
+ // the tested processor should either remove or transfer all FlowFiles it handles
+ for (auto& ff : removed_flow_files_) {
+ delete ff.release(); // NOLINT(cppcoreguidelines-owning-memory)
+ }
+ for (auto& ffs : transferred_flow_files_ | std::views::values) {
+ for (auto& ff : ffs) {
+ delete ff.release(); // NOLINT(cppcoreguidelines-owning-memory)
+ }
+ }
+}
+
+api::core::FlowFile MockProcessSession::create(const api::core::FlowFile*) {
+ auto new_ff = api::core::FlowFile{new MinifiFlowFile}; // NOLINT(clang-analyzer-cplusplus.NewDeleteLeaks)
+ flow_file_datas[new_ff.get()];
+ return new_ff;
+}
+api::core::FlowFile MockProcessSession::get() {
+ if (input_flow_files_.empty()) { return nullptr; }
+
+ auto ff = std::move(input_flow_files_.back());
+ input_flow_files_.pop_back();
+ flow_file_datas[ff.get()];
+
+ return ff;
+}
+
+void MockProcessSession::penalize(api::core::FlowFile& ff) {
+ auto& flow_file_data = flow_file_datas.at(ff.get());
+ flow_file_data.is_penalized = true;
+}
+
+void MockProcessSession::transfer(api::core::FlowFile ff, const minifi::core::Relationship& relationship) {
+ transferred_flow_files_[relationship.getName()].push_back(std::move(ff));
+}
+
+void MockProcessSession::remove(api::core::FlowFile ff) {
+ removed_flow_files_.push_back(std::move(ff));
+}
+
+void MockProcessSession::write(api::core::FlowFile& ff, const io::OutputStreamCallback& callback) {
+ auto& flow_file_data = flow_file_datas.at(ff.get());
+ const auto stream = std::make_shared<MockOutputStream>(flow_file_data.content);
+ callback(stream);
+}
+
+void MockProcessSession::read(api::core::FlowFile& ff, const io::InputStreamCallback& callback) {
+ auto& flow_file_data = flow_file_datas.at(ff.get());
+ const auto stream = std::make_shared<MockInputStream>(flow_file_data.content);
+ callback(stream);
+}
+
+void MockProcessSession::setAttribute(api::core::FlowFile& ff, std::string_view key, std::string value) {
+ auto& attributes = flow_file_datas.at(ff.get()).attributes;
+ attributes[std::string(key)] = std::move(value);
+}
+
+void MockProcessSession::removeAttribute(api::core::FlowFile& ff, std::string_view key) {
+ auto& attributes = flow_file_datas.at(ff.get()).attributes;
+ attributes.erase(std::string(key));
+}
+
+std::optional<std::string> MockProcessSession::getAttribute(api::core::FlowFile& ff, std::string_view key) {
+ auto& attributes = flow_file_datas.at(ff.get()).attributes;
+ if (const auto it = attributes.find(std::string(key)); it != attributes.end()) { return it->second; }
+ return std::nullopt;
+}
+
+std::map<std::string, std::string> MockProcessSession::getAttributes(const api::core::FlowFile& ff) const {
+ return flow_file_datas.at(ff.get()).attributes;
+}
+
+std::string MockProcessSession::getFlowFileId(const api::core::FlowFile& ff) const {
+ return flow_file_datas.at(ff.get()).id;
+}
+
+uint64_t MockProcessSession::getFlowFileSize(const api::core::FlowFile& ff) const {
+ return flow_file_datas.at(ff.get()).content.size();
+}
+
+void MockProcessSession::addInputFlowFile(MockFlowFileData flow_file_data) {
+ auto new_ff = api::core::FlowFile{new MinifiFlowFile}; // NOLINT(clang-analyzer-cplusplus.NewDeleteLeaks)
+ flow_file_datas.emplace(new_ff.get(), std::move(flow_file_data));
+ input_flow_files_.push_back(std::move(new_ff));
+}
+} // namespace org::apache::nifi::minifi::mock
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
new file mode 100644
index 000000000..8b48d0d57
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-c.cpp
@@ -0,0 +1,135 @@
+/**
+ * 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"
+
+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 MinifiProcessContextGetControllerService(MinifiProcessContext*, MinifiStringView, MinifiStringView, MinifiControllerService**) {
+ throw std::runtime_error("Not implemented");
+}
+void MinifiProcessContextGetDynamicProperties(MinifiProcessContext*,
+ void (*)(void* user_ctx, MinifiStringView dynamic_property_name, MinifiStringView dynamic_property_value), void*) {
+ throw std::runtime_error("Not implemented");
+}
+
+MinifiStatus MinifiProcessContextGetSslData(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");
+}
diff --git a/extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp b/extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp
index 089112c04..be5104c3b 100644
--- a/extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ControllerServiceContext.cpp
@@ -22,7 +22,7 @@ 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::toStringView(name),
+ const MinifiStatus status = MinifiControllerServiceContextGetProperty(impl_, utils::minifiStringView(name),
[] (void* data, const MinifiStringView 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/ProcessContext.cpp b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
index d260b9c0c..a4dd9505c 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
@@ -1,5 +1,5 @@
/**
-* Licensed to the Apache Software Foundation (ASF) under one or more
+ * Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
@@ -16,34 +16,40 @@
*/
#include "api/core/ProcessContext.h"
-#include "api/utils/minifi-c-utils.h"
+
#include "api/core/FlowFile.h"
+#include "api/utils/minifi-c-utils.h"
namespace org::apache::nifi::minifi::api::core {
-std::expected<std::string, std::error_code> ProcessContext::getProperty(std::string_view name, const FlowFile* flow_file) const {
+std::expected<std::string, std::error_code> CffiProcessContext::getProperty(const minifi::core::PropertyReference& property_reference,
+ const FlowFile* flow_file) const {
+ return getProperty(property_reference.name, flow_file);
+}
+
+std::expected<std::string, std::error_code> CffiProcessContext::getProperty(std::string_view name, const FlowFile* flow_file) const {
std::optional<std::string> value;
- MinifiStatus status = MinifiProcessContextGetProperty(impl_, utils::toStringView(name), flow_file ? flow_file->get() : MINIFI_NULL,
- [] (void* data, MinifiStringView result) {
- (*static_cast<std::optional<std::string>*>(data)) = std::string(result.data, result.length);
- }, &value);
+ const MinifiStatus status = MinifiProcessContextGetProperty(
+ 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); },
+ &value);
- if (!value) {
- return std::unexpected{utils::make_error_code(status)};
- }
+ if (!value) { return std::unexpected{utils::make_error_code(status)}; }
return value.value();
}
-bool ProcessContext::hasNonEmptyProperty(std::string_view name) const {
- return MinifiProcessContextHasNonEmptyProperty(impl_, utils::toStringView(name));
+bool CffiProcessContext::hasNonEmptyProperty(std::string_view name) const {
+ return MinifiProcessContextHasNonEmptyProperty(impl_, utils::minifiStringView(name));
}
-std::expected<MinifiControllerService*, std::error_code> ProcessContext::getControllerService(const std::string_view controller_service_name,
- const std::string_view controller_service_class) const {
+std::expected<MinifiControllerService*, std::error_code> CffiProcessContext::getControllerService(const std::string_view name,
+ const std::string_view type) const {
MinifiControllerService* controller_service = nullptr;
if (const MinifiStatus status = MinifiProcessContextGetControllerService(impl_,
- utils::toStringView(controller_service_name),
- utils::toStringView(controller_service_class),
+ utils::minifiStringView(name),
+ utils::minifiStringView(type),
&controller_service);
status != MINIFI_STATUS_SUCCESS) {
return std::unexpected{utils::make_error_code(status)};
@@ -51,4 +57,33 @@ std::expected<MinifiControllerService*, std::error_code> ProcessContext::getCont
return controller_service;
}
+std::map<std::string, std::string> CffiProcessContext::getDynamicProperties(const FlowFile* flow_file) const {
+ std::map<std::string, std::string> result;
+ MinifiProcessContextGetDynamicProperties(
+ impl_,
+ flow_file ? flow_file->get() : MINIFI_NULL,
+ [](void* user_ctx, const MinifiStringView key, const MinifiStringView value) {
+ static_cast<std::map<std::string, std::string>*>(user_ctx)->emplace(utils::toString(key), utils::toString(value));
+ },
+ &result);
+ return result;
+}
+
+std::expected<utils::net::SslData, std::error_code> CffiProcessContext::getSslData(const std::string_view name) const {
+ auto ssl_data = utils::net::SslData{};
+
+ if (const auto status = MinifiProcessContextGetSslData(impl_, utils::minifiStringView(name), [](void* data, const MinifiSslData* minifi_ssl_data) {
+ auto* my_ssl_data = static_cast<utils::net::SslData*>(data);
+ my_ssl_data->ca_loc = utils::toString(minifi_ssl_data->ca_certificate_file);
+ my_ssl_data->cert_loc = utils::toString(minifi_ssl_data->certificate_file);
+ my_ssl_data->key_loc = utils::toString(minifi_ssl_data->private_key_file);
+ my_ssl_data->key_pw = utils::toString(minifi_ssl_data->passphrase);
+ }, &ssl_data);
+ status != MINIFI_STATUS_SUCCESS) {
+ return std::unexpected{utils::make_error_code(status)};
+ }
+
+ return ssl_data;
+}
+
} // 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 8b13b09c2..5bf12b1d6 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ProcessSession.cpp
@@ -69,28 +69,34 @@ class MinifiInputStreamWrapper : public io::InputStreamImpl {
} // namespace
-FlowFile ProcessSession::get() {
+FlowFile CffiProcessSession::get() {
return FlowFile{MinifiProcessSessionGet(impl_)};
}
-FlowFile ProcessSession::create(const FlowFile* parent) {
+FlowFile CffiProcessSession::create(const FlowFile* parent) {
return FlowFile{MinifiProcessSessionCreate(impl_, parent ? parent->get() : MINIFI_NULL)};
}
-void ProcessSession::transfer(FlowFile ff, const minifi::core::Relationship& relationship) {
+void CffiProcessSession::penalize(FlowFile& ff) {
+ if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionPenalize(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::toStringView(rel_name))) {
+ if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionTransfer(impl_, ff.release(), utils::minifiStringView(rel_name))) {
throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to transfer flowfile");
}
}
-void ProcessSession::remove(FlowFile ff) {
+void CffiProcessSession::remove(FlowFile ff) {
if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionRemove(impl_, ff.release())) {
throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to remove flowfile");
}
}
-void ProcessSession::write(FlowFile& flow_file, const io::OutputStreamCallback& callback) {
+void CffiProcessSession::write(FlowFile& flow_file, const io::OutputStreamCallback& callback) {
const auto status = MinifiProcessSessionWrite(
impl_,
flow_file.get(),
@@ -103,7 +109,7 @@ void ProcessSession::write(FlowFile& flow_file, const io::OutputStreamCallback&
if (status != MINIFI_STATUS_SUCCESS) { throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to process flowfile content"); }
}
-void ProcessSession::read(FlowFile& flow_file, const io::InputStreamCallback& callback) {
+void CffiProcessSession::read(FlowFile& flow_file, const io::InputStreamCallback& callback) {
const auto status = MinifiProcessSessionRead(
impl_,
flow_file.get(),
@@ -116,28 +122,28 @@ void ProcessSession::read(FlowFile& flow_file, const io::InputStreamCallback& ca
if (status != MINIFI_STATUS_SUCCESS) { throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to process flowfile content"); }
}
-void ProcessSession::setAttribute(FlowFile& ff, const std::string_view key, std::string value) { // NOLINT(performance-unnecessary-value-param)
- const MinifiStringView value_ref = utils::toStringView(value);
- if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionSetFlowFileAttribute(impl_, ff.get(), utils::toStringView(key), &value_ref)) {
+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)) {
throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to set attribute");
}
}
-void ProcessSession::removeAttribute(FlowFile& ff, const std::string_view key) {
- if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionSetFlowFileAttribute(impl_, ff.get(), utils::toStringView(key), nullptr)) {
+void CffiProcessSession::removeAttribute(FlowFile& ff, const std::string_view key) {
+ if (MINIFI_STATUS_SUCCESS != MinifiProcessSessionSetFlowFileAttribute(impl_, ff.get(), utils::minifiStringView(key), nullptr)) {
throw minifi::Exception(minifi::FILE_OPERATION_EXCEPTION, "Failed to remove attribute");
}
}
-std::optional<std::string> ProcessSession::getAttribute(FlowFile& ff, std::string_view key) {
+std::optional<std::string> CffiProcessSession::getAttribute(FlowFile& ff, std::string_view key) {
std::optional<std::string> result;
- MinifiProcessSessionGetFlowFileAttribute(impl_, ff.get(), utils::toStringView(key), [] (void* user_ctx, MinifiStringView value) {
+ MinifiProcessSessionGetFlowFileAttribute(impl_, ff.get(), utils::minifiStringView(key), [] (void* user_ctx, MinifiStringView value) {
*static_cast<std::optional<std::string>*>(user_ctx) = std::string{value.data, value.length};
}, &result);
return result;
}
-std::map<std::string, std::string> ProcessSession::getAttributes(FlowFile& ff) {
+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) {
static_cast<std::map<std::string, std::string>*>(user_ctx)->insert({std::string{key.data, key.length}, std::string{value.data, value.length}});
@@ -145,6 +151,18 @@ std::map<std::string, std::string> ProcessSession::getAttributes(FlowFile& ff) {
return result;
}
+std::string CffiProcessSession::getFlowFileId(const FlowFile& ff) const {
+ std::string result;
+ MinifiProcessSessionGetFlowFileId(impl_, ff.get(), [](void* user_ctx, const MinifiStringView 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());
+}
+
void ProcessSession::writeBuffer(FlowFile& flow_file, std::span<const char> buffer) {
writeBuffer(flow_file, as_bytes(buffer));
}
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 d91a67f60..4f5f2f422 100644
--- a/extension-framework/cpp-extension-lib/src/core/logging/Logger.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/logging/Logger.cpp
@@ -49,19 +49,19 @@ minifi::core::logging::LOG_LEVEL toLogLevel(MinifiLogLevel level) {
} // namespace
-void Logger::set_max_log_size(int size) {
+void CffiLogger::set_max_log_size(const int size) {
MinifiLoggerSetMaxLogSize(impl_, size);
}
-void Logger::log_string(minifi::core::logging::LOG_LEVEL level, std::string str) {
+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()});
}
-bool Logger::should_log(minifi::core::logging::LOG_LEVEL level) {
+bool CffiLogger::should_log(const minifi::core::logging::LOG_LEVEL level) {
return MinifiLoggerShouldLog(impl_, toCLogLevel(level));
}
-[[nodiscard]] minifi::core::logging::LOG_LEVEL Logger::level() const {
+[[nodiscard]] minifi::core::logging::LOG_LEVEL CffiLogger::level() const {
return toLogLevel(MinifiLoggerLevel(impl_));
}
diff --git a/libminifi/test/integration/extension-verification-test/CApiExtension.cpp b/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
index 546a22905..13bec7e0c 100644
--- a/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
+++ b/libminifi/test/integration/extension-verification-test/CApiExtension.cpp
@@ -44,8 +44,8 @@ CEXTENSIONAPI const uint32_t MinifiApiVersion = MINIFI_TEST_API_VERSION;
CEXTENSIONAPI void MinifiInitExtension(MinifiExtensionContext* extension_context) {
MinifiExtensionDefinition extension_definition{
- .name = minifi::api::utils::toStringView(MAKESTRING(EXTENSION_NAME)),
- .version = minifi::api::utils::toStringView(MAKESTRING(EXTENSION_VERSION)),
+ .name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
+ .version = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_VERSION)),
.deinit = nullptr,
.user_data = nullptr
};