Skip to content

MINIFICPP-2769 Move Kafka Extension to stable C API #2175

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
martinzink wants to merge 1 commit into
base: MINIFICPP-2798
Choose a base branch
from
Open

MINIFICPP-2769 Move Kafka Extension to stable C API #2175

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -748,7 +748,7 @@ set(EXTENSIONS_ENABLED_BY_DEFAULT (
minifi-grafana-loki
minifi-archive-extensions
minifi-mqtt-extensions
minifi-rdkafka-extensions
minifi-kafka
minifi-pdh
minifi-prometheus
minifi-rocksdb-repos
Expand Down
22 changes: 21 additions & 1 deletion core-framework/common/include/core/PropertyDefinitionBuilder.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,21 @@ namespace org::apache::nifi::minifi::core {
namespace detail {
template<typename... Types>
inline constexpr auto TypeNames = std::array<std::string_view, sizeof...(Types)>{core::className<Types>()...};
}

template <size_t N>
struct StringLiteral {
char value[N];
constexpr StringLiteral(const char (&str)[N]) { // NOLINT(runtime/explicit)
for (size_t i = 0; i < N; ++i) {
value[i] = str[i];
}
}
};

// A variable template that creates permanent static memory for the span to point to
template <StringLiteral str>
inline constexpr auto StaticAllowedType = std::array<std::string_view, 1>{std::string_view{str.value, sizeof(str.value) - 1}};
} // namespace detail

template<size_t NumAllowedValues = 0>
struct PropertyDefinitionBuilder {
Expand Down Expand Up @@ -81,6 +95,12 @@ struct PropertyDefinitionBuilder {
return *this;
}

template <detail::StringLiteral TypeName>
constexpr PropertyDefinitionBuilder<NumAllowedValues> withAllowedType() {
property.allowed_types = detail::StaticAllowedType<TypeName>;
return *this;
}

constexpr PropertyDefinitionBuilder<NumAllowedValues> withValidator(const PropertyValidator& property_validator) {
property.validator = gsl::make_not_null(&property_validator);
return *this;
Expand Down
4 changes: 2 additions & 2 deletions extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ class ProcessContext {
[[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;
[[nodiscard]] virtual std::expected<utils::net::SslData, std::error_code> getSslData(const minifi::core::PropertyReference& prop) const = 0;
};

class CffiProcessContext : public ProcessContext {
Expand All @@ -58,7 +58,7 @@ class CffiProcessContext : public ProcessContext {
[[nodiscard]] std::map<std::string, std::string> getDynamicProperties(const FlowFile* flow_file) const override;
[[nodiscard]] bool hasNonEmptyProperty(std::string_view name) const override;

[[nodiscard]] std::expected<utils::net::SslData, std::error_code> getSslData(std::string_view name) const override;
[[nodiscard]] std::expected<utils::net::SslData, std::error_code> getSslData(const minifi::core::PropertyReference& prop) const override;

private:
[[nodiscard]] std::expected<std::string, std::error_code> getProperty(std::string_view name, const FlowFile* flow_file) const;
Expand Down
6 changes: 3 additions & 3 deletions extension-framework/cpp-extension-lib/mocklib/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,6 @@

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)
add_library(mock-minifi STATIC ${SRC_FILES})
target_link_libraries(mock-minifi PUBLIC minifi-cpp-extension-lib minifi-core-framework-common minifi-c-api)
target_include_directories(mock-minifi PUBLIC include)
2 changes: 1 addition & 1 deletion extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@ class MockProcessContext : public api::core::ProcessContext {
[[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;
[[nodiscard]] std::expected<api::utils::net::SslData, std::error_code> getSslData(const minifi::core::PropertyReference& prop) const override;

std::map<std::string, std::string, std::less<>> properties_;

Expand Down
2 changes: 1 addition & 1 deletion extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,7 +44,7 @@ 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 {
std::expected<api::utils::net::SslData, std::error_code> MockProcessContext::getSslData(const minifi::core::PropertyReference&) const {
return api::utils::net::SslData{};
}
} // namespace org::apache::nifi::minifi::mock
4 changes: 3 additions & 1 deletion extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-c.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

#include "minifi-c.h"

extern "C" {
MinifiExtension* MINIFI_REGISTER_EXTENSION_FN(MinifiExtensionContext*, const MinifiExtensionDefinition*) {
throw std::runtime_error("Not implemented");
}
Expand Down Expand Up @@ -51,7 +52,7 @@ void MinifiProcessContextGetDynamicProperties(MinifiProcessContext*, MinifiFlowF
throw std::runtime_error("Not implemented");
}

MinifiStatus MinifiProcessContextGetSslData(MinifiProcessContext*, MinifiStringView,
MinifiStatus MinifiProcessContextGetSslDataFromProperty(MinifiProcessContext*, MinifiStringView,
void (*)(void* user_ctx, const MinifiSslData* ssl_data), void*) {
throw std::runtime_error("Not implemented");
}
Expand Down Expand Up @@ -133,3 +134,4 @@ MinifiStatus MinifiControllerServiceContextGetProperty(MinifiControllerServiceCo
void (*)(void* user_ctx, MinifiStringView property_value), void*) {
throw std::runtime_error("Not implemented");
}
} // extern "C"
4 changes: 2 additions & 2 deletions extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,10 +69,10 @@ std::map<std::string, std::string> CffiProcessContext::getDynamicProperties(cons
return result;
}

std::expected<utils::net::SslData, std::error_code> CffiProcessContext::getSslData(const std::string_view name) const {
std::expected<utils::net::SslData, std::error_code> CffiProcessContext::getSslData(const minifi::core::PropertyReference& prop) const {
auto ssl_data = utils::net::SslData{};

if (const auto status = MinifiProcessContextGetSslData(impl_, utils::minifiStringView(name), [](void* data, const MinifiSslData* minifi_ssl_data) {
if (const auto status = MinifiProcessContextGetSslDataFromProperty(impl_, utils::minifiStringView(prop.name), [](void* data, const MinifiSslData* minifi_ssl_data) {
auto* my_ssl_data = static_cast<utils::net::SslData*>(data);
my_ssl_data->ca_loc = utils::toString(minifi_ssl_data->ca_certificate_file);
my_ssl_data->cert_loc = utils::toString(minifi_ssl_data->certificate_file);
Expand Down
10 changes: 5 additions & 5 deletions extensions/kafka/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,13 +27,13 @@ include(${CMAKE_SOURCE_DIR}/extensions/ExtensionHeader.txt)

file(GLOB SOURCES "*.cpp")

add_minifi_library(minifi-rdkafka-extensions SHARED ${SOURCES})
add_minifi_library(minifi-kafka SHARED ${SOURCES})

target_link_libraries(minifi-rdkafka-extensions ${LIBMINIFI} Threads::Threads)
target_link_libraries(minifi-kafka minifi-cpp-extension-lib Threads::Threads)

get_target_property(RDKAFKA_INCLUDE_DIRS rdkafka INCLUDE_DIRECTORIES)
target_include_directories(minifi-rdkafka-extensions SYSTEM PUBLIC ${RDKAFKA_INCLUDE_DIRS})
target_link_libraries(minifi-rdkafka-extensions rdkafka)
target_include_directories(minifi-kafka SYSTEM PUBLIC ${RDKAFKA_INCLUDE_DIRS})
target_link_libraries(minifi-kafka rdkafka)

register_extension(minifi-rdkafka-extensions "RDKAFKA EXTENSIONS" RDKAFKA-EXTENSIONS "This Enables librdkafka functionality including PublishKafka" "extensions/kafka/tests")
register_c_api_extension(minifi-kafka "RDKAFKA EXTENSIONS" RDKAFKA-EXTENSIONS "This Enables librdkafka functionality including PublishKafka" "extensions/kafka/tests")

Loading
Loading