MINIFICPP-2769 Move Kafka Extension to stable C API

Author: martinzink

CMakeLists.txt

@@ -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

core-framework/common/include/core/PropertyDefinitionBuilder.h

@@ -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 {
@@ -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;

extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h

@@ -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 {
@@ -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;

extension-framework/cpp-extension-lib/mocklib/CMakeLists.txt

@@ -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)

extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h

@@ -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_;
 

extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp

@@ -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

extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-c.cpp

@@ -19,6 +19,7 @@
 
 #include "minifi-c.h"
 
+extern "C" {
 MinifiExtension* MINIFI_REGISTER_EXTENSION_FN(MinifiExtensionContext*, const MinifiExtensionDefinition*) {
   throw std::runtime_error("Not implemented");
 }
@@ -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");
 }
@@ -133,3 +134,4 @@ MinifiStatus MinifiControllerServiceContextGetProperty(MinifiControllerServiceCo
     void (*)(void* user_ctx, MinifiStringView property_value), void*) {
   throw std::runtime_error("Not implemented");
 }
+}  // extern "C"

extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp

@@ -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);

extensions/kafka/CMakeLists.txt

@@ -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")