This is an automated email from the ASF dual-hosted git repository.
martinzink pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
The following commit(s) were added to refs/heads/main by this push:
new a50d59166 MINIFICPP-2798 Mock library for C extensions (#2170)
a50d59166 is described below
commit a50d5916614dec24cf8e3916299914a7a1638f6c
Author: Martin Zink <[email protected]>
AuthorDate: Tue Jun 2 09:01:59 2026 +0200
MINIFICPP-2798 Mock library for C extensions (#2170)
---
.../{ => 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 +-
.../cpp-extension-lib/include/api/core/FlowFile.h | 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 ++---
.../include/api/{core/FlowFile.h => utils/Ssl.h} | 44 ++++---
.../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 | 81 +++++++++++++
.../mocklib/include/MockStreams.h | 98 +++++++++++++++
.../FlowFile.h => mocklib/include/MockUtils.h} | 31 ++---
.../mocklib/src/MockProcessContext.cpp | 50 ++++++++
.../mocklib/src/MockProcessSession.cpp | 87 +++++++++++++
.../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 +-
.../llamacpp/processors/ExtensionInitializer.cpp | 4 +-
.../llamacpp/processors/RunLlamaCppInference.cpp | 2 +-
.../extension-verification-test/CApiExtension.cpp | 4 +-
29 files changed, 790 insertions(+), 167 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/FlowFile.h
b/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h
index 4a288f36d..da9e6e8e1 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/FlowFile.h
@@ -17,8 +17,10 @@
*/
#pragma once
+#include <exception>
#include <memory>
#include <stdexcept>
+
#include "minifi-c.h"
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 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/FlowFile.h
b/extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
similarity index 58%
copy from extension-framework/cpp-extension-lib/include/api/core/FlowFile.h
copy to extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
index 4a288f36d..b68922477 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/FlowFile.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,24 +16,35 @@
*/
#pragma once
+#include <string>
#include <memory>
-#include <stdexcept>
-#include "minifi-c.h"
-
-namespace org::apache::nifi::minifi::api::core {
-
-struct EnsureMovedFromDeleter {
- void operator()(MinifiFlowFile* 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)
- } else {
- throw std::logic_error("Each flowfile should be either transferred or
removed");
- }
- }
+#include <optional>
+#include <filesystem>
+
+#include "utils/Enum.h"
+
+namespace org::apache::nifi::minifi::api::utils::net {
+
+enum class ClientAuthOption {
+ NONE,
+ WANT,
+ REQUIRED
+};
+
+struct SslData {
+ std::filesystem::path ca_loc;
+ std::filesystem::path cert_loc;
+ std::filesystem::path key_loc;
+ std::string key_pw;
+
+ bool isValid() const {
+ return !cert_loc.empty() && !key_loc.empty();
}
};
-using FlowFile = std::unique_ptr<MinifiFlowFile, EnsureMovedFromDeleter>;
+struct SslServerOptions {
+ SslData cert_data;
+ ClientAuthOption client_auth_option;
+};
-} // namespace org::apache::nifi::minifi::api::core
+} // 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..2c54cc663 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..51aeeb020
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessSession.h
@@ -0,0 +1,81 @@
+/**
+ * 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"
+
+struct MinifiFlowFile {
+ explicit MinifiFlowFile(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;
+
+ MinifiFlowFile() = default;
+ MinifiFlowFile(const MinifiFlowFile&) = default;
+ MinifiFlowFile(MinifiFlowFile&&) = default;
+ MinifiFlowFile& operator=(const MinifiFlowFile&) = default;
+ MinifiFlowFile& operator=(MinifiFlowFile&&) = default;
+
+ std::map<std::string, std::string> attributes;
+ std::vector<std::byte> content;
+ bool is_penalized = false;
+ std::string id;
+};
+
+namespace org::apache::nifi::minifi::mock {
+
+class MockProcessSession : public api::core::ProcessSession {
+ public:
+ MockProcessSession() = default;
+ MockProcessSession(const MockProcessSession&) = delete;
+ MockProcessSession& operator=(const MockProcessSession&) = delete;
+ MockProcessSession(MockProcessSession&&) = delete;
+ MockProcessSession& operator=(MockProcessSession&&) = delete;
+
+ 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(std::unique_ptr<MinifiFlowFile> flow_file);
+
+ private:
+ std::vector<std::unique_ptr<MinifiFlowFile>> 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_;
+};
+
+} // 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..b5204c58c
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockStreams.h
@@ -0,0 +1,98 @@
+/**
+ * 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 <cstring>
+#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/FlowFile.h
b/extension-framework/cpp-extension-lib/mocklib/include/MockUtils.h
similarity index 56%
copy from extension-framework/cpp-extension-lib/include/api/core/FlowFile.h
copy to extension-framework/cpp-extension-lib/mocklib/include/MockUtils.h
index 4a288f36d..cae06ca4f 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/FlowFile.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,26 +14,20 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#pragma once
-#include <memory>
-#include <stdexcept>
-#include "minifi-c.h"
+#pragma once
-namespace org::apache::nifi::minifi::api::core {
+#include <string>
-struct EnsureMovedFromDeleter {
- void operator()(MinifiFlowFile* 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)
- } else {
- throw std::logic_error("Each flowfile should be either transferred or
removed");
- }
- }
- }
-};
+#include "MockLogger.h"
+#include "minifi-cpp/core/ProcessorMetadata.h"
-using FlowFile = std::unique_ptr<MinifiFlowFile, EnsureMovedFromDeleter>;
+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
+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..c3edc0d66
--- /dev/null
+++ b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessSession.cpp
@@ -0,0 +1,87 @@
+/**
+ * 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 {
+
+api::core::FlowFile MockProcessSession::create(const api::core::FlowFile*) {
+ return api::core::FlowFile{new MinifiFlowFile};
+}
+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();
+
+ return api::core::FlowFile{ff.release()};
+}
+
+void MockProcessSession::penalize(api::core::FlowFile& ff) {
+ ff->is_penalized = true;
+}
+
+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()));
+}
+
+void MockProcessSession::remove(api::core::FlowFile ff) {
+ removed_flow_files_.push_back(std::unique_ptr<MinifiFlowFile>(ff.release()));
+}
+
+void MockProcessSession::write(api::core::FlowFile& ff, const
io::OutputStreamCallback& callback) {
+ const auto stream = std::make_shared<MockOutputStream>(ff->content);
+ callback(stream);
+}
+
+void MockProcessSession::read(api::core::FlowFile& ff, const
io::InputStreamCallback& callback) {
+ const auto stream = std::make_shared<MockInputStream>(ff->content);
+ callback(stream);
+}
+
+void MockProcessSession::setAttribute(api::core::FlowFile& ff,
std::string_view key, std::string value) {
+ ff->attributes[std::string(key)] = std::move(value);
+}
+
+void MockProcessSession::removeAttribute(api::core::FlowFile& ff,
std::string_view key) {
+ ff->attributes.erase(std::string(key));
+}
+
+std::optional<std::string>
MockProcessSession::getAttribute(api::core::FlowFile& ff, std::string_view key)
{
+ auto& attributes = ff->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 ff->attributes;
+}
+
+std::string MockProcessSession::getFlowFileId(const api::core::FlowFile& ff)
const {
+ return ff->id;
+}
+
+uint64_t MockProcessSession::getFlowFileSize(const api::core::FlowFile& ff)
const {
+ return ff->content.size();
+}
+
+void MockProcessSession::addInputFlowFile(std::unique_ptr<MinifiFlowFile>
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-c.cpp
b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-c.cpp
new file mode 100644
index 000000000..a546f07ad
--- /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*,
MinifiFlowFile*,
+ 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/extensions/llamacpp/processors/ExtensionInitializer.cpp
b/extensions/llamacpp/processors/ExtensionInitializer.cpp
index 85d610a7d..8e06bf0cc 100644
--- a/extensions/llamacpp/processors/ExtensionInitializer.cpp
+++ b/extensions/llamacpp/processors/ExtensionInitializer.cpp
@@ -28,8 +28,8 @@ CEXTENSIONAPI const uint32_t MinifiApiVersion =
MINIFI_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
};
diff --git a/extensions/llamacpp/processors/RunLlamaCppInference.cpp
b/extensions/llamacpp/processors/RunLlamaCppInference.cpp
index ede77cd06..888280cb1 100644
--- a/extensions/llamacpp/processors/RunLlamaCppInference.cpp
+++ b/extensions/llamacpp/processors/RunLlamaCppInference.cpp
@@ -32,7 +32,7 @@ MinifiStatus
RunLlamaCppInference::onScheduleImpl(api::core::ProcessContext& con
model_path_.clear();
model_path_ = api::utils::parseProperty(context, ModelPath);
multimodal_model_path_ = api::utils::parseOptionalProperty(context,
MultiModalModelPath);
- system_prompt_ = context.getProperty(SystemPrompt).value_or("");
+ system_prompt_ = api::utils::parseOptionalProperty(context,
SystemPrompt).value_or("");
output_attribute_ = api::utils::parseOptionalProperty(context,
OutputAttributeName);
LlamaSamplerParams llama_sampler_params;
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
};
