MINIFICPP-2764 Controller Service CPP API

Extensions.md

@@ -48,8 +48,8 @@ extern "C" void MinifiInitExtension(MinifiExtensionContext* extension_context) {
     .user_data = nullptr
   };
   auto* extension = MinifiRegisterExtension(extension_context, &extension_definition);
-  minifi::api::core::useProcessorClassDescription<minifi::extensions::llamacpp::processors::RunLlamaCppInference>([&] (const MinifiProcessorClassDefinition& description) {
-    MinifiRegisterProcessor(extension, &description);
+  minifi::api::core::useProcessorClassDefinition<minifi::extensions::llamacpp::processors::RunLlamaCppInference>([&] (const MinifiProcessorClassDefinition& definition) {
+    MinifiRegisterProcessor(extension, &definition);
   });
 }
 ```

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

@@ -0,0 +1,41 @@
+/**
+ * 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 <string>
+#include <expected>
+
+#include "minifi-c.h"
+#include "minifi-cpp/core/PropertyDefinition.h"
+
+namespace org::apache::nifi::minifi::api::core {
+
+class ControllerServiceContext {
+ public:
+  explicit ControllerServiceContext(MinifiControllerServiceContext* impl) : impl_(impl) {}
+
+  [[nodiscard]] std::expected<std::string, std::error_code> getProperty(std::string_view name) const;
+  [[nodiscard]] std::expected<std::string, std::error_code> getProperty(const minifi::core::PropertyReference& property_reference) const {
+    return getProperty(property_reference.name);
+  }
+
+ private:
+  MinifiControllerServiceContext* impl_;
+};
+
+}  // namespace org::apache::nifi::minifi::api::core

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

@@ -0,0 +1,63 @@
+/**
+ * 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 <memory>
+#include <string>
+
+#include "ControllerServiceContext.h"
+#include "minifi-cpp/core/ControllerServiceMetadata.h"
+#include "minifi-cpp/utils/Id.h"
+#include "utils/SmallString.h"
+
+namespace org::apache::nifi::minifi::api {
+
+class Connection;
+
+namespace core {
+
+/// Helper class that makes implementing the stable API easier
+/// It is used in conjunction with useControllerServiceClassDefinition
+class ControllerServiceImpl {
+ public:
+  explicit ControllerServiceImpl(minifi::core::ControllerServiceMetadata metadata);
+
+  ControllerServiceImpl(const ControllerServiceImpl&) = delete;
+  ControllerServiceImpl(ControllerServiceImpl&&) = delete;
+  ControllerServiceImpl& operator=(const ControllerServiceImpl&) = delete;
+  ControllerServiceImpl& operator=(ControllerServiceImpl&&) = delete;
+
+  virtual ~ControllerServiceImpl();
+
+  MinifiStatus enable(ControllerServiceContext&);
+  void disable();
+
+  [[nodiscard]] std::string getName() const;
+  [[nodiscard]] minifi::utils::Identifier getUUID() const;
+  [[nodiscard]] minifi::utils::SmallString<36> getUUIDStr() const;
+
+ protected:
+  virtual MinifiStatus enableImpl(api::core::ControllerServiceContext&) = 0;
+  virtual void disableImpl() {}
+
+  minifi::core::ControllerServiceMetadata metadata_;
+
+  std::shared_ptr<minifi::core::logging::Logger> logger_;
+};
+
+}  // namespace core
+}  // namespace org::apache::nifi::minifi::api

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

@@ -34,8 +34,9 @@ class ProcessContext {
   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;
 
-  bool hasNonEmptyProperty(std::string_view name) const;
+  [[nodiscard]] bool hasNonEmptyProperty(std::string_view name) const;
 
  private:
   MinifiProcessContext* impl_;

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

@@ -17,9 +17,7 @@
 #pragma once
 
 #include <atomic>
-#include <condition_variable>
 #include <memory>
-#include <mutex>
 #include <string>
 #include <vector>
 
@@ -53,7 +51,7 @@ class ProcessorImpl {
 
   virtual ~ProcessorImpl();
 
-  void setTriggerWhenEmpty(bool trigger_when_empty) {
+  void setTriggerWhenEmpty(const bool trigger_when_empty) {
     trigger_when_empty_ = trigger_when_empty;
   }
 
@@ -89,9 +87,6 @@ class ProcessorImpl {
   std::atomic<bool> trigger_when_empty_;
 
   std::shared_ptr<minifi::core::logging::Logger> logger_;
-
- private:
-  mutable std::mutex mutex_;
 };
 
 }  // namespace core

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

@@ -26,19 +26,21 @@
 #define WIN32_LEAN_AND_MEAN 1
 #endif
 
-#include "minifi-c.h"
-#include "core/ClassName.h"
-#include "api/utils/minifi-c-utils.h"
+#include "ControllerServiceContext.h"
+#include "FlowFile.h"
 #include "ProcessContext.h"
 #include "ProcessSession.h"
-#include "FlowFile.h"
-#include "minifi-cpp/core/ProcessorMetadata.h"
+#include "api/utils/minifi-c-utils.h"
+#include "core/ClassName.h"
 #include "logging/Logger.h"
+#include "minifi-c.h"
+#include "minifi-cpp/core/ControllerServiceMetadata.h"
+#include "minifi-cpp/core/ProcessorMetadata.h"
 
 namespace org::apache::nifi::minifi::api::core {
 
 template<typename Class, typename Fn>
-void useProcessorClassDescription(Fn&& fn) {
+void useProcessorClassDefinition(Fn&& fn) {
   std::vector<std::vector<MinifiStringView>> string_vector_cache;
 
   const auto full_name = minifi::core::className<Class>();
@@ -76,7 +78,7 @@ void useProcessorClassDescription(Fn&& fn) {
     attribute_relationships_cache.push_back(std::move(rel_cache));
   }
 
-  MinifiProcessorClassDefinition description{
+  MinifiProcessorClassDefinition definition{
     .full_name = utils::toStringView(full_name),
     .description = utils::toStringView(Class::Description),
     .class_properties_count = gsl::narrow<uint32_t>(class_properties.size()),
@@ -94,11 +96,12 @@ void useProcessorClassDescription(Fn&& fn) {
 
     .callbacks = MinifiProcessorCallbacks{
       .create = [] (MinifiProcessorMetadata metadata) -> MINIFI_OWNED void* {
-        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)
-        }};
+        try {
+          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)}};
+        } catch (...) { return nullptr; }
       },
       .destroy = [] (MINIFI_OWNED void* self) -> void {
         delete static_cast<Class*>(self);
@@ -127,7 +130,9 @@ void useProcessorClassDescription(Fn&& fn) {
         }
       },
       .onUnSchedule = [] (void* self) -> void {
-        static_cast<Class*>(self)->onUnSchedule();
+        try {
+          static_cast<Class*>(self)->onUnSchedule();
+        } catch (...) {}
       },
       .calculateMetrics = [] (void* self) -> MINIFI_OWNED MinifiPublishedMetrics* {
         auto metrics = static_cast<Class*>(self)->calculateMetrics();
@@ -142,7 +147,46 @@ void useProcessorClassDescription(Fn&& fn) {
     }
   };
 
-  fn(description);
+  fn(definition);
+}
+
+template<typename Class, typename Fn>
+void useControllerServiceClassDefinition(Fn&& fn) {
+  std::vector<std::vector<MinifiStringView>> string_vector_cache;
+
+  const auto full_name = minifi::core::className<Class>();
+
+  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),
+      .class_properties_count = gsl::narrow<uint32_t>(class_properties.size()),
+      .class_properties_ptr = class_properties.data(),
+
+      .callbacks = MinifiControllerServiceCallbacks{
+          .create = [](MinifiControllerServiceMetadata metadata) -> MINIFI_OWNED void* {
+            try {
+              return new Class{minifi::core::ControllerServiceMetadata{
+                  .uuid = minifi::utils::Identifier::parse(std::string{metadata.uuid.data, metadata.uuid.length}).value(),
+                  .name = std::string{metadata.name.data, metadata.name.length},
+                  .logger = std::make_shared<logging::Logger>(metadata.logger)}};
+            } catch (...) { return nullptr; }
+          },
+          .destroy = [](MINIFI_OWNED void* self) -> void { delete static_cast<Class*>(self); },
+          .enable = [](void* self, MinifiControllerServiceContext* context) -> MinifiStatus {
+            ControllerServiceContext context_wrapper(context);
+            try {
+              return static_cast<Class*>(self)->enable(context_wrapper);
+            } catch (...) { return MINIFI_STATUS_UNKNOWN_ERROR; }
+          },
+          .disable = [](void* self) -> void {
+            try {
+              static_cast<Class*>(self)->disable();
+            } catch (...) {}
+          },
+      }};
+
+  fn(definition);
 }
 
 }  // namespace org::apache::nifi::minifi::api::core

extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h

@@ -167,4 +167,28 @@ std::optional<T> parseOptionalEnumProperty(const core::ProcessContext& context,
   return result.value();
 }
 
+template<typename ControllerServiceType>
+ControllerServiceType* parseOptionalControllerService(const core::ProcessContext& context, const minifi::core::PropertyReference& prop) {
+  const auto controller_service_name = context.getProperty(prop.name);
+  if (!controller_service_name || controller_service_name->empty()) {
+    return nullptr;
+  }
+
+  auto service = context.getControllerService(*controller_service_name, minifi::core::className<ControllerServiceType>());
+  if (!service) {
+    return nullptr;
+  }
+
+  return reinterpret_cast<ControllerServiceType*>(*service);
+}
+
+template<typename ControllerServiceType>
+gsl::not_null<ControllerServiceType*> parseControllerService(const core::ProcessContext& context, const minifi::core::PropertyReference& prop) {
+  auto controller_service = parseOptionalControllerService<ControllerServiceType>(context, prop);
+  if (!controller_service) {
+    throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("Required controller service property '{}' is missing", prop.name));
+  }
+  return gsl::make_not_null(controller_service);
+}
+
 }  // namespace org::apache::nifi::minifi::api::utils

extension-framework/cpp-extension-lib/libtest/CProcessorTestUtils.h

@@ -1,5 +1,5 @@
 /**
-*
+ *
  * 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,22 +17,40 @@
  */
 #pragma once
 
+#include "api/core/Resource.h"
 #include "core/Processor.h"
+#include "core/controller/ControllerService.h"
+#include "minifi-cpp/core/ControllerServiceMetadata.h"
 #include "minifi-cpp/core/ProcessorMetadata.h"
-#include "api/core/Resource.h"
+#include "utils/CControllerService.h"
 #include "utils/CProcessor.h"
+#include "minifi-cpp/agent/agent_docs.h"
 
 namespace org::apache::nifi::minifi::test::utils {
 
-template<typename T, typename ...Args>
-std::unique_ptr<minifi::core::Processor> make_custom_c_processor(minifi::core::ProcessorMetadata metadata, Args&&... args) {  // NOLINT(cppcoreguidelines-missing-std-forward)
+template<typename T, typename... Args>
+std::unique_ptr<minifi::core::Processor> make_custom_c_processor(minifi::core::ProcessorMetadata metadata,
+    Args&&... args) {  // NOLINT(cppcoreguidelines-missing-std-forward)
   std::unique_ptr<minifi::core::ProcessorApi> processor_impl;
-  minifi::api::core::useProcessorClassDescription<T>([&] (const MinifiProcessorClassDefinition& description) {
-    minifi::utils::useCProcessorClassDescription(description, [&] (const auto&, auto c_description) {
+  minifi::api::core::useProcessorClassDefinition<T>([&](const MinifiProcessorClassDefinition& definition) {
+    minifi::utils::useCProcessorClassDescription(definition, [&](const auto&, auto c_description) {
       processor_impl = std::make_unique<minifi::utils::CProcessor>(std::move(c_description), metadata, new T(metadata, std::forward<Args>(args)...));
     });
   });
   return std::make_unique<minifi::core::Processor>(metadata.name, metadata.uuid, std::move(processor_impl));
 }
 
+template<typename T, typename... Args>
+std::shared_ptr<minifi::core::controller::ControllerService> make_custom_c_controller_service(minifi::core::ControllerServiceMetadata metadata,
+    Args&&... args) {  // NOLINT(cppcoreguidelines-missing-std-forward)
+  std::unique_ptr<minifi::core::controller::ControllerServiceApi> controller_service_impl;
+  minifi::api::core::useControllerServiceClassDefinition<T>([&](const MinifiControllerServiceClassDefinition& definition) {
+    minifi::utils::useCControllerServiceClassDescription(definition, [&](const auto&, auto c_description) {
+      controller_service_impl = std::make_unique<minifi::utils::CControllerService>(std::move(c_description),
+          metadata,
+          new T(metadata, std::forward<Args>(args)...));
+    });
+  });
+  return std::make_shared<minifi::core::controller::ControllerService>(metadata.name, metadata.uuid, std::move(controller_service_impl));
+}
 }  // namespace org::apache::nifi::minifi::test::utils

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

@@ -0,0 +1,40 @@
+/**
+ * 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 "api/core/ControllerServiceContext.h"
+#include "api/utils/minifi-c-utils.h"
+
+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),
+    [] (void* data, const MinifiStringView result) {
+      (*static_cast<std::optional<std::string>*>(data)) = std::string(result.data, result.length);
+    }, &value);
+
+  if (status != MINIFI_STATUS_SUCCESS) {
+    return std::unexpected{utils::make_error_code(status)};
+  }
+
+  if (!value) {
+    return std::unexpected{utils::make_error_code(MINIFI_STATUS_UNKNOWN_ERROR)};
+  }
+  return value.value();
+}
+
+}  // namespace org::apache::nifi::minifi::api::core