This is an automated email from the ASF dual-hosted git repository.
szaszm 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 7988c8b0e MINIFICPP-2765 Move GCP Extension to stable C API (#2153)
7988c8b0e is described below
commit 7988c8b0e3afcb7a191c364524ae2ea3b958c291
Author: Martin Zink <[email protected]>
AuthorDate: Tue Jun 16 14:29:03 2026 +0200
MINIFICPP-2765 Move GCP Extension to stable C API (#2153)
---
.../src/minifi_behave/steps/checking_steps.py | 2 +-
.../include/api/core/ProcessContext.h | 15 +-
.../include/api/utils/ProcessorConfigUtils.h | 6 +-
.../include/api/utils/{Ssl.h => Proxy.h} | 40 +++---
.../cpp-extension-lib/include/api/utils/Ssl.h | 4 -
.../mocklib/include/MockProcessContext.h | 6 +-
.../mocklib/src/MockProcessContext.cpp | 10 +-
.../mocklib/src/mock-minifi-c.cpp | 7 +-
.../cpp-extension-lib/src/core/ProcessContext.cpp | 47 +++++-
extensions/gcp/CMakeLists.txt | 6 +-
extensions/gcp/ExtensionInitializer.cpp | 46 ++++++
extensions/gcp/GCPAttributes.h | 49 +++----
.../GCPCredentialsControllerService.cpp | 34 ++---
.../GCPCredentialsControllerService.h | 27 ++--
extensions/gcp/processors/DeleteGCSObject.cpp | 52 +++----
extensions/gcp/processors/DeleteGCSObject.h | 14 +-
extensions/gcp/processors/FetchGCSObject.cpp | 78 +++++-----
extensions/gcp/processors/FetchGCSObject.h | 16 +--
extensions/gcp/processors/GCSProcessor.cpp | 32 ++---
extensions/gcp/processors/GCSProcessor.h | 22 ++-
extensions/gcp/processors/ListGCSBucket.cpp | 39 +++--
extensions/gcp/processors/ListGCSBucket.h | 13 +-
extensions/gcp/processors/PutGCSObject.cpp | 71 +++++----
extensions/gcp/processors/PutGCSObject.h | 12 +-
extensions/gcp/tests/CMakeLists.txt | 2 +-
extensions/gcp/tests/DeleteGCSObjectTests.cpp | 82 ++++++-----
extensions/gcp/tests/FetchGCSObjectTests.cpp | 83 ++++++-----
.../tests/GCPCredentialsControllerServiceTests.cpp | 59 +++++---
extensions/gcp/tests/ListGCSBucketTests.cpp | 86 +++++------
extensions/gcp/tests/PutGCSObjectTests.cpp | 159 +++++++++++----------
extensions/kafka/KafkaProcessorBase.cpp | 4 +-
extensions/kafka/rdkafka_utils.h | 1 +
libminifi/src/minifi-c.cpp | 49 ++++++-
minifi-api/include/minifi-c/minifi-c.h | 25 +++-
minifi-api/minifi-c-api.def | 3 +-
35 files changed, 665 insertions(+), 536 deletions(-)
diff --git a/behave_framework/src/minifi_behave/steps/checking_steps.py
b/behave_framework/src/minifi_behave/steps/checking_steps.py
index 5f52ee740..418e875c9 100644
--- a/behave_framework/src/minifi_behave/steps/checking_steps.py
+++ b/behave_framework/src/minifi_behave/steps/checking_steps.py
@@ -119,7 +119,7 @@ def verify_minifi_logs_match_regex(context, regex,
duration):
@step('no errors were generated on the http-proxy regarding "{url}"')
def verify_no_errors_on_http_proxy(context: MinifiTestContext, url: str):
http_proxy_container = next(container for container in
context.containers.values() if isinstance(container, HttpProxy))
- assert http_proxy_container.check_http_proxy_access(url) or
http_proxy_container.log_app_output()
+ assert http_proxy_container.check_http_proxy_access(url) or
log_due_to_failure(context)
@then('in the "{container}" container no files are placed in the "{directory}"
directory in {duration} of running time')
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 87209577c..26a55a44a 100644
--- a/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
+++ b/extension-framework/cpp-extension-lib/include/api/core/ProcessContext.h
@@ -17,10 +17,11 @@
#pragma once
-#include <string>
#include <expected>
+#include <string>
#include "api/core/FlowFile.h"
+#include "api/utils/Proxy.h"
#include "api/utils/Ssl.h"
#include "minifi-c.h"
#include "minifi-cpp/core/PropertyDefinition.h"
@@ -39,12 +40,12 @@ class ProcessContext {
[[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 std::expected<MinifiControllerService*,
std::error_code> getControllerService(const minifi::core::PropertyReference&
prop) 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(const minifi::core::PropertyReference& prop) const = 0;
+ [[nodiscard]] virtual std::expected<std::optional<utils::net::SslData>,
std::error_code> getSslData(const minifi::core::PropertyReference& prop) const
= 0;
+ [[nodiscard]] virtual std::expected<std::optional<utils::ProxyData>,
std::error_code> getProxyData(const minifi::core::PropertyReference& prop)
const = 0;
};
class CffiProcessContext : public ProcessContext {
@@ -53,12 +54,12 @@ class CffiProcessContext : public ProcessContext {
[[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::expected<MinifiControllerService*, std::error_code>
getControllerService(const minifi::core::PropertyReference& prop) 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]] std::expected<utils::net::SslData, std::error_code>
getSslData(const minifi::core::PropertyReference& prop) const override;
+ [[nodiscard]] std::expected<std::optional<utils::net::SslData>,
std::error_code> getSslData(const minifi::core::PropertyReference& prop) const
override;
+ [[nodiscard]] std::expected<std::optional<utils::ProxyData>,
std::error_code> getProxyData(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;
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 db44e1cd5..2ec9571ed 100644
---
a/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
+++
b/extension-framework/cpp-extension-lib/include/api/utils/ProcessorConfigUtils.h
@@ -169,12 +169,8 @@ 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, nullptr);
- if (!controller_service_name || controller_service_name->empty()) {
- return nullptr;
- }
+ auto service = context.getControllerService(prop);
- auto service = context.getControllerService(*controller_service_name,
minifi::core::className<ControllerServiceType>());
if (!service) {
return nullptr;
}
diff --git a/extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
b/extension-framework/cpp-extension-lib/include/api/utils/Proxy.h
similarity index 58%
copy from extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
copy to extension-framework/cpp-extension-lib/include/api/utils/Proxy.h
index b68922477..6add9ef4e 100644
--- a/extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
+++ b/extension-framework/cpp-extension-lib/include/api/utils/Proxy.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
@@ -16,35 +16,27 @@
*/
#pragma once
-#include <string>
-#include <memory>
-#include <optional>
#include <filesystem>
+#include <optional>
+#include <string>
-#include "utils/Enum.h"
-
-namespace org::apache::nifi::minifi::api::utils::net {
+namespace org::apache::nifi::minifi::api::utils {
-enum class ClientAuthOption {
- NONE,
- WANT,
- REQUIRED
+enum class ProxyType {
+ DIRECT,
+ HTTP
};
-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();
- }
+struct BasicAuthCredentials {
+ std::string username;
+ std::string password;
};
-struct SslServerOptions {
- SslData cert_data;
- ClientAuthOption client_auth_option;
+struct ProxyData {
+ ProxyType proxy_type;
+ std::string host;
+ uint16_t port;
+ std::optional<BasicAuthCredentials> proxy_credentials;
};
-} // namespace org::apache::nifi::minifi::api::utils::net
+} // namespace org::apache::nifi::minifi::api::utils
diff --git a/extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
b/extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
index b68922477..c34965abb 100644
--- a/extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
+++ b/extension-framework/cpp-extension-lib/include/api/utils/Ssl.h
@@ -17,12 +17,8 @@
#pragma once
#include <string>
-#include <memory>
-#include <optional>
#include <filesystem>
-#include "utils/Enum.h"
-
namespace org::apache::nifi::minifi::api::utils::net {
enum class ClientAuthOption {
diff --git
a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
index 76a13c8d8..cddcc9761 100644
--- a/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
+++ b/extension-framework/cpp-extension-lib/mocklib/include/MockProcessContext.h
@@ -29,12 +29,12 @@ class MockProcessContext : public api::core::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::expected<MinifiControllerService*, std::error_code>
getControllerService(const minifi::core::PropertyReference& prop) 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(const minifi::core::PropertyReference& prop) const override;
+ [[nodiscard]] std::expected<std::optional<api::utils::net::SslData>,
std::error_code> getSslData(const minifi::core::PropertyReference& prop) const
override;
+ [[nodiscard]] std::expected<std::optional<api::utils::ProxyData>,
std::error_code> getProxyData(const minifi::core::PropertyReference& prop)
const override;
std::map<std::string, std::string, std::less<>> properties_;
diff --git
a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp
b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp
index afd5b9b2f..2b571afcf 100644
--- a/extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp
+++ b/extension-framework/cpp-extension-lib/mocklib/src/MockProcessContext.cpp
@@ -32,7 +32,7 @@ std::expected<std::string, std::error_code>
MockProcessContext::getProperty(cons
return
std::unexpected{make_error_code(core::PropertyErrorCode::PropertyNotSet)};
}
-std::expected<MinifiControllerService*, std::error_code>
MockProcessContext::getControllerService(std::string_view, std::string_view)
const {
+std::expected<MinifiControllerService*, std::error_code>
MockProcessContext::getControllerService(const
minifi::core::PropertyReference&) const {
return nullptr;
}
@@ -44,7 +44,11 @@ bool MockProcessContext::hasNonEmptyProperty(const
std::string_view name) const
return properties_.contains(name);
}
-std::expected<api::utils::net::SslData, std::error_code>
MockProcessContext::getSslData(const minifi::core::PropertyReference&) const {
- return api::utils::net::SslData{};
+std::expected<std::optional<api::utils::net::SslData>, std::error_code>
MockProcessContext::getSslData(const minifi::core::PropertyReference&) const {
+ return std::nullopt;
+}
+
+[[nodiscard]] std::expected<std::optional<api::utils::ProxyData>,
std::error_code> MockProcessContext::getProxyData(const
minifi::core::PropertyReference&) const {
+ return std::nullopt;
}
} // 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
index a50ca712b..873b6527a 100644
--- a/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-c.cpp
+++ b/extension-framework/cpp-extension-lib/mocklib/src/mock-minifi-c.cpp
@@ -44,7 +44,7 @@ MinifiBool
MinifiProcessContextHasNonEmptyProperty(MinifiProcessContext*, Minifi
throw std::runtime_error("Not implemented");
}
-MinifiStatus MinifiProcessContextGetControllerService(MinifiProcessContext*,
MinifiStringView, MinifiStringView, MinifiControllerService**) {
+MinifiStatus
MinifiProcessContextGetControllerServiceFromProperty(MinifiProcessContext*,
MinifiStringView, MinifiStringView, MinifiControllerService**) {
throw std::runtime_error("Not implemented");
}
void MinifiProcessContextGetDynamicProperties(MinifiProcessContext*,
MinifiFlowFile*,
@@ -134,4 +134,9 @@ MinifiStatus
MinifiControllerServiceContextGetProperty(MinifiControllerServiceCo
void (*)(void* user_ctx, MinifiStringView property_value), void*) {
throw std::runtime_error("Not implemented");
}
+
+MinifiStatus
MinifiProcessContextGetProxyDataFromProperty(MinifiProcessContext*,
MinifiStringView,
+ void (*)(void* user_ctx, const MinifiProxyData* proxy_data), void*) {
+ throw std::runtime_error("Not implemented");
+}
} // extern "C"
diff --git a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
index 90c333f0d..e2f3a65c2 100644
--- a/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
+++ b/extension-framework/cpp-extension-lib/src/core/ProcessContext.cpp
@@ -44,14 +44,17 @@ bool
CffiProcessContext::hasNonEmptyProperty(std::string_view name) const {
return MinifiProcessContextHasNonEmptyProperty(impl_,
utils::minifiStringView(name));
}
-std::expected<MinifiControllerService*, std::error_code>
CffiProcessContext::getControllerService(const std::string_view name,
- const std::string_view type) const {
+std::expected<MinifiControllerService*, std::error_code>
CffiProcessContext::getControllerService(const minifi::core::PropertyReference&
prop) const {
MinifiControllerService* controller_service = nullptr;
- if (const MinifiStatus status =
MinifiProcessContextGetControllerService(impl_,
- utils::minifiStringView(name),
- utils::minifiStringView(type),
+ gsl_Assert(prop.allowed_types.size() == 1);
+ const MinifiStatus status =
MinifiProcessContextGetControllerServiceFromProperty(impl_,
+ utils::minifiStringView(prop.name),
+ utils::minifiStringView(prop.allowed_types[0]),
&controller_service);
- status != MINIFI_STATUS_SUCCESS) {
+ if (status == MINIFI_STATUS_PROPERTY_NOT_SET) {
+ return nullptr;
+ }
+ if (status != MINIFI_STATUS_SUCCESS) {
return std::unexpected{utils::make_error_code(status)};
}
return controller_service;
@@ -69,7 +72,10 @@ std::map<std::string, std::string>
CffiProcessContext::getDynamicProperties(cons
return result;
}
-std::expected<utils::net::SslData, std::error_code>
CffiProcessContext::getSslData(const minifi::core::PropertyReference& prop)
const {
+std::expected<std::optional<utils::net::SslData>, std::error_code>
CffiProcessContext::getSslData(const minifi::core::PropertyReference& prop)
const {
+ const auto controller_name = getProperty(prop, nullptr);
+ if (!controller_name) { return std::nullopt; }
+
auto ssl_data = utils::net::SslData{};
if (const auto status = MinifiProcessContextGetSslDataFromProperty(impl_,
utils::minifiStringView(prop.name), [](void* data, const MinifiSslData*
minifi_ssl_data) {
@@ -86,4 +92,31 @@ std::expected<utils::net::SslData, std::error_code>
CffiProcessContext::getSslDa
return ssl_data;
}
+std::expected<std::optional<utils::ProxyData>, std::error_code>
CffiProcessContext::getProxyData(const minifi::core::PropertyReference& prop)
const {
+ auto proxy_data = utils::ProxyData{};
+ const auto status = MinifiProcessContextGetProxyDataFromProperty(
+ impl_,
+ utils::minifiStringView(prop.name),
+ [](void* data, const MinifiProxyData* minifi_proxy_data) {
+ auto* proxy = static_cast<utils::ProxyData*>(data);
+ proxy->host = utils::toString(minifi_proxy_data->hostname);
+ proxy->port = minifi_proxy_data->port;
+ if (minifi_proxy_data->password && minifi_proxy_data->username) {
+ proxy->proxy_credentials = utils::BasicAuthCredentials{.username =
utils::toString(*minifi_proxy_data->username),
+ .password = utils::toString(*minifi_proxy_data->password)};
+ } else {
+ proxy->proxy_credentials = std::nullopt;
+ }
+ if (minifi_proxy_data->proxy_type == MINIFI_PROXY_TYPE_HTTP) {
+ proxy->proxy_type = utils::ProxyType::HTTP;
+ } else {
+ proxy->proxy_type = utils::ProxyType::DIRECT;
+ }
+ },
+ &proxy_data);
+ if (status == MINIFI_STATUS_PROPERTY_NOT_SET) { return std::nullopt; }
+ if (status == MINIFI_STATUS_SUCCESS) { return proxy_data; }
+ return std::unexpected{utils::make_error_code(status)};
+}
+
} // namespace org::apache::nifi::minifi::api::core
diff --git a/extensions/gcp/CMakeLists.txt b/extensions/gcp/CMakeLists.txt
index 654209d08..2fc42260c 100644
--- a/extensions/gcp/CMakeLists.txt
+++ b/extensions/gcp/CMakeLists.txt
@@ -30,8 +30,8 @@ add_minifi_library(minifi-gcp SHARED ${SOURCES})
if (NOT WIN32)
target_compile_options(minifi-gcp PRIVATE
-Wno-error=deprecated-declarations) # Suppress deprecation warnings for
std::rel_ops usage
endif()
-target_link_libraries(minifi-gcp ${LIBMINIFI} google-cloud-cpp::storage)
-target_include_directories(minifi-gcp SYSTEM PUBLIC
${google-cloud-cpp_INCLUDE_DIRS})
+target_link_libraries(minifi-gcp minifi-cpp-extension-lib
google-cloud-cpp::storage)
-register_extension(minifi-gcp "GCP EXTENSIONS" GCP-EXTENSIONS "This enables
Google Cloud Platform support" "extensions/gcp/tests")
+target_include_directories(minifi-gcp SYSTEM PUBLIC
${google-cloud-cpp_INCLUDE_DIRS})
+register_c_api_extension(minifi-gcp "GCP EXTENSIONS" GCP-EXTENSIONS "This
enables Google Cloud Platform support" "extensions/gcp/tests")
diff --git a/extensions/gcp/ExtensionInitializer.cpp
b/extensions/gcp/ExtensionInitializer.cpp
new file mode 100644
index 000000000..89b00e9ff
--- /dev/null
+++ b/extensions/gcp/ExtensionInitializer.cpp
@@ -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.
+ */
+
+#include
"../../extension-framework/cpp-extension-lib/include/api/core/Resource.h"
+#include "api/core/Resource.h"
+#include "api/utils/minifi-c-utils.h"
+#include "processors/DeleteGCSObject.h"
+#include "processors/FetchGCSObject.h"
+#include "processors/ListGCSBucket.h"
+#include "processors/PutGCSObject.h"
+
+#define MKSOC(x) #x
+#define MAKESTRING(x) MKSOC(x) // NOLINT(cppcoreguidelines-macro-usage)
+
+namespace minifi = org::apache::nifi::minifi;
+
+CEXTENSIONAPI const uint32_t MinifiApiVersion = MINIFI_API_VERSION;
+
+CEXTENSIONAPI void MinifiInitExtension(MinifiExtensionContext*
extension_context) {
+ const MinifiExtensionDefinition extension_definition{
+ .name = minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_NAME)),
+ .version =
minifi::api::utils::minifiStringView(MAKESTRING(EXTENSION_VERSION)),
+ .deinit = nullptr,
+ .user_data = nullptr
+ };
+ auto* extension = MinifiRegisterExtension(extension_context,
&extension_definition);
+
minifi::api::core::registerProcessors<minifi::extensions::gcp::DeleteGCSObject,
+ minifi::extensions::gcp::FetchGCSObject,
+ minifi::extensions::gcp::ListGCSBucket,
+ minifi::extensions::gcp::PutGCSObject>(extension);
+
minifi::api::core::registerControllerServices<minifi::extensions::gcp::GCPCredentialsControllerService>(extension);
+}
diff --git a/extensions/gcp/GCPAttributes.h b/extensions/gcp/GCPAttributes.h
index 5657e7200..0ec973ea6 100644
--- a/extensions/gcp/GCPAttributes.h
+++ b/extensions/gcp/GCPAttributes.h
@@ -19,8 +19,9 @@
#include <string_view>
+#include "api/core/FlowFile.h"
+#include "api/core/ProcessSession.h"
#include "google/cloud/storage/object_metadata.h"
-#include "minifi-cpp/core/FlowFile.h"
namespace org::apache::nifi::minifi::extensions::gcp {
@@ -50,32 +51,32 @@ constexpr std::string_view GCS_SELF_LINK_ATTR =
"gcs.self.link";
constexpr std::string_view GCS_ENCRYPTION_ALGORITHM_ATTR =
"gcs.encryption.algorithm";
constexpr std::string_view GCS_ENCRYPTION_SHA256_ATTR =
"gcs.encryption.sha256";
-inline void setAttributesFromObjectMetadata(core::FlowFile& flow_file, const
::google::cloud::storage::ObjectMetadata& object_metadata) {
- flow_file.setAttribute(GCS_BUCKET_ATTR, object_metadata.bucket());
- flow_file.setAttribute(GCS_OBJECT_NAME_ATTR, object_metadata.name());
- flow_file.setAttribute(GCS_SIZE_ATTR,
std::to_string(object_metadata.size()));
- flow_file.setAttribute(GCS_CRC32C_ATTR, object_metadata.crc32c());
- flow_file.setAttribute(GCS_MD5_ATTR, object_metadata.md5_hash());
- flow_file.setAttribute(GCS_CONTENT_ENCODING_ATTR,
object_metadata.content_encoding());
- flow_file.setAttribute(GCS_CONTENT_LANGUAGE_ATTR,
object_metadata.content_language());
- flow_file.setAttribute(GCS_CONTENT_DISPOSITION_ATTR,
object_metadata.content_disposition());
- flow_file.setAttribute(GCS_CREATE_TIME_ATTR,
std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.time_created().time_since_epoch()).count()));
- flow_file.setAttribute(GCS_UPDATE_TIME_ATTR,
std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.updated().time_since_epoch()).count()));
- flow_file.setAttribute(GCS_DELETE_TIME_ATTR,
std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.time_deleted().time_since_epoch()).count()));
- flow_file.setAttribute(GCS_MEDIA_LINK_ATTR, object_metadata.media_link());
- flow_file.setAttribute(GCS_SELF_LINK_ATTR, object_metadata.self_link());
- flow_file.setAttribute(GCS_ETAG_ATTR, object_metadata.etag());
- flow_file.setAttribute(GCS_GENERATED_ID, object_metadata.id());
- flow_file.setAttribute(GCS_META_GENERATION,
std::to_string(object_metadata.metageneration()));
- flow_file.setAttribute(GCS_GENERATION,
std::to_string(object_metadata.generation()));
- flow_file.setAttribute(GCS_STORAGE_CLASS, object_metadata.storage_class());
+inline void setAttributesFromObjectMetadata(api::core::FlowFile& flow_file,
const ::google::cloud::storage::ObjectMetadata& object_metadata,
api::core::ProcessSession& session) {
+ session.setAttribute(flow_file, GCS_BUCKET_ATTR, object_metadata.bucket());
+ session.setAttribute(flow_file, GCS_OBJECT_NAME_ATTR,
object_metadata.name());
+ session.setAttribute(flow_file, GCS_SIZE_ATTR,
std::to_string(object_metadata.size()));
+ session.setAttribute(flow_file, GCS_CRC32C_ATTR, object_metadata.crc32c());
+ session.setAttribute(flow_file, GCS_MD5_ATTR, object_metadata.md5_hash());
+ session.setAttribute(flow_file, GCS_CONTENT_ENCODING_ATTR,
object_metadata.content_encoding());
+ session.setAttribute(flow_file, GCS_CONTENT_LANGUAGE_ATTR,
object_metadata.content_language());
+ session.setAttribute(flow_file, GCS_CONTENT_DISPOSITION_ATTR,
object_metadata.content_disposition());
+ session.setAttribute(flow_file, GCS_CREATE_TIME_ATTR,
std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.time_created().time_since_epoch()).count()));
+ session.setAttribute(flow_file, GCS_UPDATE_TIME_ATTR,
std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.updated().time_since_epoch()).count()));
+ session.setAttribute(flow_file, GCS_DELETE_TIME_ATTR,
std::to_string(std::chrono::duration_cast<std::chrono::milliseconds>(object_metadata.time_deleted().time_since_epoch()).count()));
+ session.setAttribute(flow_file, GCS_MEDIA_LINK_ATTR,
object_metadata.media_link());
+ session.setAttribute(flow_file, GCS_SELF_LINK_ATTR,
object_metadata.self_link());
+ session.setAttribute(flow_file, GCS_ETAG_ATTR, object_metadata.etag());
+ session.setAttribute(flow_file, GCS_GENERATED_ID, object_metadata.id());
+ session.setAttribute(flow_file, GCS_META_GENERATION,
std::to_string(object_metadata.metageneration()));
+ session.setAttribute(flow_file, GCS_GENERATION,
std::to_string(object_metadata.generation()));
+ session.setAttribute(flow_file, GCS_STORAGE_CLASS,
object_metadata.storage_class());
if (object_metadata.has_customer_encryption()) {
- flow_file.setAttribute(GCS_ENCRYPTION_ALGORITHM_ATTR,
object_metadata.customer_encryption().encryption_algorithm);
- flow_file.setAttribute(GCS_ENCRYPTION_SHA256_ATTR,
object_metadata.customer_encryption().key_sha256);
+ session.setAttribute(flow_file, GCS_ENCRYPTION_ALGORITHM_ATTR,
object_metadata.customer_encryption().encryption_algorithm);
+ session.setAttribute(flow_file, GCS_ENCRYPTION_SHA256_ATTR,
object_metadata.customer_encryption().key_sha256);
}
if (object_metadata.has_owner()) {
- flow_file.setAttribute(GCS_OWNER_ENTITY_ATTR,
object_metadata.owner().entity);
- flow_file.setAttribute(GCS_OWNER_ENTITY_ID_ATTR,
object_metadata.owner().entity_id);
+ session.setAttribute(flow_file, GCS_OWNER_ENTITY_ATTR,
object_metadata.owner().entity);
+ session.setAttribute(flow_file, GCS_OWNER_ENTITY_ID_ATTR,
object_metadata.owner().entity_id);
}
}
diff --git
a/extensions/gcp/controllerservices/GCPCredentialsControllerService.cpp
b/extensions/gcp/controllerservices/GCPCredentialsControllerService.cpp
index 9e93c2942..d84bbeb4b 100644
--- a/extensions/gcp/controllerservices/GCPCredentialsControllerService.cpp
+++ b/extensions/gcp/controllerservices/GCPCredentialsControllerService.cpp
@@ -18,34 +18,36 @@
#include "GCPCredentialsControllerService.h"
-#include "core/Resource.h"
#include "google/cloud/storage/client.h"
-#include "utils/ProcessorConfigUtils.h"
-#include "utils/file/FileUtils.h"
namespace org::apache::nifi::minifi::extensions::gcp {
-void GCPCredentialsControllerService::initialize() {
- setSupportedProperties(Properties);
+namespace {
+// TODO(MINIFICPP-2763) use utils::file::get_content instead
+std::string get_content(const std::filesystem::path& file_name) {
+ std::ifstream file(file_name, std::ifstream::binary);
+ std::string content((std::istreambuf_iterator<char>(file)),
std::istreambuf_iterator<char>());
+ return content;
+}
}
-std::shared_ptr<google::cloud::Credentials>
GCPCredentialsControllerService::createCredentialsFromJsonPath() const {
- const auto json_path = getProperty(JsonFilePath.name);
+std::shared_ptr<google::cloud::Credentials>
GCPCredentialsControllerService::createCredentialsFromJsonPath(api::core::ControllerServiceContext&
ctx) const {
+ const auto json_path = ctx.getProperty(JsonFilePath.name);
if (!json_path) {
logger_->log_error("Missing or invalid {}", JsonFilePath.name);
return nullptr;
}
- if (!utils::file::exists(*json_path)) {
+ if (std::error_code ec; !std::filesystem::exists(*json_path, ec) || ec) {
logger_->log_error("JSON file for GCP credentials '{}' does not exist",
*json_path);
return nullptr;
}
- return
google::cloud::MakeServiceAccountCredentials(utils::file::get_content(*json_path));
+ return google::cloud::MakeServiceAccountCredentials(get_content(*json_path));
}
-std::shared_ptr<google::cloud::Credentials>
GCPCredentialsControllerService::createCredentialsFromJsonContents() const {
- auto json_contents = getProperty(JsonContents.name);
+std::shared_ptr<google::cloud::Credentials>
GCPCredentialsControllerService::createCredentialsFromJsonContents(api::core::ControllerServiceContext&
ctx) const {
+ auto json_contents = ctx.getProperty(JsonContents.name);
if (!json_contents) {
logger_->log_error("Missing or invalid {}", JsonContents.name);
return nullptr;
@@ -54,9 +56,9 @@ std::shared_ptr<google::cloud::Credentials>
GCPCredentialsControllerService::cre
return google::cloud::MakeServiceAccountCredentials(*json_contents);
}
-void GCPCredentialsControllerService::onEnable() {
+MinifiStatus
GCPCredentialsControllerService::enableImpl(api::core::ControllerServiceContext&
ctx) {
std::optional<CredentialsLocation> credentials_location;
- if (const auto value = getProperty(CredentialsLoc.name)) {
+ if (const auto value = ctx.getProperty(CredentialsLoc.name)) {
credentials_location = magic_enum::enum_cast<CredentialsLocation>(*value);
}
if (!credentials_location) {
@@ -68,15 +70,15 @@ void GCPCredentialsControllerService::onEnable() {
} else if (*credentials_location ==
CredentialsLocation::USE_COMPUTE_ENGINE_CREDENTIALS) {
credentials_ = google::cloud::MakeComputeEngineCredentials();
} else if (*credentials_location == CredentialsLocation::USE_JSON_FILE) {
- credentials_ = createCredentialsFromJsonPath();
+ credentials_ = createCredentialsFromJsonPath(ctx);
} else if (*credentials_location == CredentialsLocation::USE_JSON_CONTENTS) {
- credentials_ = createCredentialsFromJsonContents();
+ credentials_ = createCredentialsFromJsonContents(ctx);
} else if (*credentials_location ==
CredentialsLocation::USE_ANONYMOUS_CREDENTIALS) {
credentials_ = google::cloud::MakeInsecureCredentials();
}
if (!credentials_)
logger_->log_error("Couldn't create valid credentials");
+ return MINIFI_STATUS_SUCCESS;
}
-REGISTER_RESOURCE(GCPCredentialsControllerService, ControllerService);
} // namespace org::apache::nifi::minifi::extensions::gcp
diff --git
a/extensions/gcp/controllerservices/GCPCredentialsControllerService.h
b/extensions/gcp/controllerservices/GCPCredentialsControllerService.h
index 0eb8fbf8a..b53971ae2 100644
--- a/extensions/gcp/controllerservices/GCPCredentialsControllerService.h
+++ b/extensions/gcp/controllerservices/GCPCredentialsControllerService.h
@@ -18,17 +18,15 @@
#pragma once
#include <filesystem>
-#include <string>
#include <memory>
+#include <string>
-#include "core/controller/ControllerServiceBase.h"
-#include "minifi-cpp/core/logging/Logger.h"
-#include "core/logging/LoggerFactory.h"
-#include "minifi-cpp/core/PropertyDefinition.h"
+#include "api/core/ControllerServiceImpl.h"
+#include "api/utils/Export.h"
#include "core/PropertyDefinitionBuilder.h"
-#include "utils/Enum.h"
-
#include "google/cloud/credentials.h"
+#include "minifi-cpp/core/PropertyDefinition.h"
+#include "utils/Enum.h"
namespace org::apache::nifi::minifi::extensions::gcp {
enum class CredentialsLocation {
@@ -63,7 +61,7 @@ constexpr customize_t
enum_name<CredentialsLocation>(CredentialsLocation value)
namespace org::apache::nifi::minifi::extensions::gcp {
-class GCPCredentialsControllerService : public
core::controller::ControllerServiceBase, public
core::controller::ControllerServiceHandle {
+class GCPCredentialsControllerService : public
api::core::ControllerServiceImpl {
public:
EXTENSIONAPI static constexpr const char* Description = "Manages the
credentials for Google Cloud Platform. This allows for multiple Google Cloud
Platform related processors "
"to reference this single controller service so that Google Cloud
Platform credentials can be managed and controlled in a central location.";
@@ -91,21 +89,16 @@ class GCPCredentialsControllerService : public
core::controller::ControllerServi
EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
- ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_CONTROLLER_SERVICES
-
- using ControllerServiceBase::ControllerServiceBase;
-
- void initialize() override;
- void onEnable() override;
+ using ControllerServiceImpl::ControllerServiceImpl;
- [[nodiscard]] ControllerServiceHandle* getControllerServiceHandle() override
{return this;}
+ MinifiStatus enableImpl(api::core::ControllerServiceContext& ctx) override;
[[nodiscard]] const auto& getCredentials() const { return credentials_; }
protected:
- [[nodiscard]] std::shared_ptr<google::cloud::Credentials>
createCredentialsFromJsonPath() const;
- [[nodiscard]] std::shared_ptr<google::cloud::Credentials>
createCredentialsFromJsonContents() const;
+ [[nodiscard]] std::shared_ptr<google::cloud::Credentials>
createCredentialsFromJsonPath(api::core::ControllerServiceContext& ctx) const;
+ [[nodiscard]] std::shared_ptr<google::cloud::Credentials>
createCredentialsFromJsonContents(api::core::ControllerServiceContext& ctx)
const;
std::shared_ptr<google::cloud::Credentials> credentials_;
diff --git a/extensions/gcp/processors/DeleteGCSObject.cpp
b/extensions/gcp/processors/DeleteGCSObject.cpp
index 4ad4b4327..35206f17e 100644
--- a/extensions/gcp/processors/DeleteGCSObject.cpp
+++ b/extensions/gcp/processors/DeleteGCSObject.cpp
@@ -17,69 +17,61 @@
#include "DeleteGCSObject.h"
-#include "utils/ProcessorConfigUtils.h"
-
#include "../GCPAttributes.h"
-#include "minifi-cpp/core/FlowFile.h"
-#include "minifi-cpp/core/ProcessContext.h"
-#include "core/ProcessSession.h"
-#include "core/Resource.h"
+#include "api/core/ProcessContext.h"
+#include "api/core/ProcessSession.h"
+#include "api/core/Resource.h"
+#include "api/utils/ProcessorConfigUtils.h"
namespace gcs = ::google::cloud::storage;
namespace org::apache::nifi::minifi::extensions::gcp {
-void DeleteGCSObject::initialize() {
- setSupportedProperties(Properties);
- setSupportedRelationships(Relationships);
-}
-void DeleteGCSObject::onTrigger(core::ProcessContext& context,
core::ProcessSession& session) {
+MinifiStatus DeleteGCSObject::onTriggerImpl(api::core::ProcessContext&
context, api::core::ProcessSession& session) {
gsl_Expects(gcp_credentials_);
auto flow_file = session.get();
if (!flow_file) {
- context.yield();
- return;
+ return MINIFI_STATUS_PROCESSOR_YIELD;
}
- auto bucket = context.getProperty(Bucket, flow_file.get());
+ auto bucket = api::utils::parseOptionalProperty(context, Bucket, &flow_file);
if (!bucket || bucket->empty()) {
logger_->log_error("Missing bucket name");
- session.transfer(flow_file, Failure);
- return;
+ session.transfer(std::move(flow_file), Failure);
+ return MINIFI_STATUS_SUCCESS;
}
- auto object_name = context.getProperty(Key, flow_file.get());
+ auto object_name = api::utils::parseOptionalProperty(context, Key,
&flow_file);
if (!object_name || object_name->empty()) {
logger_->log_error("Missing object name");
- session.transfer(flow_file, Failure);
- return;
+ session.transfer(std::move(flow_file), Failure);
+ return MINIFI_STATUS_SUCCESS;
}
gcs::Generation generation;
- if (const auto object_generation_str =
context.getProperty(ObjectGeneration, flow_file.get()); object_generation_str
&& !object_generation_str->empty()) {
+ if (auto object_generation_str = api::utils::parseOptionalProperty(context,
ObjectGeneration, &flow_file); object_generation_str &&
!object_generation_str->empty()) {
if (const auto geni64 =
parsing::parseIntegral<int64_t>(*object_generation_str)) {
generation = gcs::Generation{*geni64};
} else {
logger_->log_error("Invalid generation: {}", *object_generation_str);
- session.transfer(flow_file, Failure);
- return;
+ session.transfer(std::move(flow_file), Failure);
+ return MINIFI_STATUS_SUCCESS;
}
}
auto status = getClient().DeleteObject(*bucket, *object_name, generation,
gcs::IfGenerationNotMatch(0));
if (!status.ok()) {
- flow_file->setAttribute(GCS_STATUS_MESSAGE, status.message());
- flow_file->setAttribute(GCS_ERROR_REASON, status.error_info().reason());
- flow_file->setAttribute(GCS_ERROR_DOMAIN, status.error_info().domain());
+ session.setAttribute(flow_file, GCS_STATUS_MESSAGE, status.message());
+ session.setAttribute(flow_file, GCS_ERROR_REASON,
status.error_info().reason());
+ session.setAttribute(flow_file, GCS_ERROR_DOMAIN,
status.error_info().domain());
logger_->log_error("Failed to delete {} object from {} bucket on Google
Cloud Storage {} {}", *object_name, *bucket, status.message(),
status.error_info().reason());
- session.transfer(flow_file, Failure);
- return;
+ session.transfer(std::move(flow_file), Failure);
+ return MINIFI_STATUS_SUCCESS;
}
- session.transfer(flow_file, Success);
+ session.transfer(std::move(flow_file), Success);
+ return MINIFI_STATUS_SUCCESS;
}
-REGISTER_RESOURCE(DeleteGCSObject, Processor);
-
} // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/DeleteGCSObject.h
b/extensions/gcp/processors/DeleteGCSObject.h
index a5d022847..48364cd45 100644
--- a/extensions/gcp/processors/DeleteGCSObject.h
+++ b/extensions/gcp/processors/DeleteGCSObject.h
@@ -17,24 +17,20 @@
#pragma once
-#include <memory>
-#include <string>
-#include <utility>
-
#include "../GCPAttributes.h"
#include "GCSProcessor.h"
-#include "core/logging/LoggerFactory.h"
#include "minifi-cpp/core/OutputAttributeDefinition.h"
#include "minifi-cpp/core/PropertyDefinition.h"
#include "core/PropertyDefinitionBuilder.h"
#include "utils/ArrayUtils.h"
+#include "minifi-cpp/core/Annotation.h"
+
namespace org::apache::nifi::minifi::extensions::gcp {
class DeleteGCSObject : public GCSProcessor {
public:
using GCSProcessor::GCSProcessor;
- ~DeleteGCSObject() override = default;
EXTENSIONAPI static constexpr const char* Description = "Deletes an object
from a Google Cloud Bucket.";
@@ -79,10 +75,8 @@ class DeleteGCSObject : public GCSProcessor {
EXTENSIONAPI static constexpr core::annotation::Input InputRequirement =
core::annotation::Input::INPUT_REQUIRED;
EXTENSIONAPI static constexpr bool IsSingleThreaded = false;
- ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS
-
- void initialize() override;
- void onTrigger(core::ProcessContext& context, core::ProcessSession& session)
override;
+ protected:
+ MinifiStatus onTriggerImpl(api::core::ProcessContext& context,
api::core::ProcessSession& session) override;
};
} // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/FetchGCSObject.cpp
b/extensions/gcp/processors/FetchGCSObject.cpp
index e43b80415..93f500806 100644
--- a/extensions/gcp/processors/FetchGCSObject.cpp
+++ b/extensions/gcp/processors/FetchGCSObject.cpp
@@ -19,11 +19,9 @@
#include <utility>
-#include "core/Resource.h"
-#include "minifi-cpp/core/FlowFile.h"
-#include "minifi-cpp/core/ProcessContext.h"
-#include "core/ProcessSession.h"
#include "../GCPAttributes.h"
+#include "api/utils/ProcessorConfigUtils.h"
+#include "minifi-cpp/io/OutputStream.h"
namespace gcs = ::google::cloud::storage;
@@ -82,80 +80,78 @@ class FetchFromGCSCallback {
};
} // namespace
-
-void FetchGCSObject::initialize() {
- setSupportedProperties(Properties);
- setSupportedRelationships(Relationships);
-}
-
-void FetchGCSObject::onSchedule(core::ProcessContext& context,
core::ProcessSessionFactory& session_factory) {
- GCSProcessor::onSchedule(context, session_factory);
- if (auto encryption_key = context.getProperty(EncryptionKey)) {
+MinifiStatus FetchGCSObject::onScheduleImpl(api::core::ProcessContext&
context) {
+ const auto status = GCSProcessor::onScheduleImpl(context);
+ if (MINIFI_STATUS_SUCCESS != status) { return status; }
+ if (auto encryption_key = context.getProperty(EncryptionKey, nullptr)) {
try {
encryption_key_ = gcs::EncryptionKey::FromBase64Key(*encryption_key);
} catch (const google::cloud::RuntimeStatusError&) {
- throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION,
"Could not decode the base64-encoded encryption key from property " +
std::string(EncryptionKey.name)); }
+ logger_->log_error("Could not decode the base64-encoded encryption key
from property {}", std::string(EncryptionKey.name));
+ return MINIFI_STATUS_UNKNOWN_ERROR;
+ }
}
+ return MINIFI_STATUS_SUCCESS;
}
-void FetchGCSObject::onTrigger(core::ProcessContext& context,
core::ProcessSession& session) {
+MinifiStatus FetchGCSObject::onTriggerImpl(api::core::ProcessContext& context,
api::core::ProcessSession& session) {
gsl_Expects(gcp_credentials_);
auto flow_file = session.get();
if (!flow_file) {
- context.yield();
- return;
+ return MINIFI_STATUS_PROCESSOR_YIELD;
}
- auto bucket = context.getProperty(Bucket, flow_file.get());
+ auto bucket = api::utils::parseOptionalProperty(context, Bucket, &flow_file);
if (!bucket || bucket->empty()) {
logger_->log_error("Missing bucket name");
- session.transfer(flow_file, Failure);
- return;
+ session.transfer(std::move(flow_file), Failure);
+ return MINIFI_STATUS_SUCCESS;
}
- auto object_name = context.getProperty(Key, flow_file.get());
+ auto object_name = api::utils::parseOptionalProperty(context, Key,
&flow_file);
if (!object_name || object_name->empty()) {
logger_->log_error("Missing object name");
- session.transfer(flow_file, Failure);
- return;
+ session.transfer(std::move(flow_file), Failure);
+ return MINIFI_STATUS_SUCCESS;
}
gcs::Client client = getClient();
FetchFromGCSCallback callback(client, *bucket, *object_name);
callback.setEncryptionKey(encryption_key_);
- if (const auto object_generation_str =
context.getProperty(ObjectGeneration, flow_file.get()); object_generation_str
&& !object_generation_str->empty()) {
+ if (const auto object_generation_str =
api::utils::parseOptionalProperty(context, ObjectGeneration, &flow_file);
object_generation_str && !object_generation_str->empty()) {
if (const auto geni64 =
parsing::parseIntegral<int64_t>(*object_generation_str)) {
gcs::Generation generation = gcs::Generation{*geni64};
callback.setGeneration(generation);
} else {
logger_->log_error("Invalid generation: {}", *object_generation_str);
- session.transfer(flow_file, Failure);
- return;
+ session.transfer(std::move(flow_file), Failure);
+ return MINIFI_STATUS_SUCCESS;
}
}
-
session.write(flow_file, std::ref(callback));
if (!callback.getStatus().ok()) {
- flow_file->setAttribute(GCS_STATUS_MESSAGE,
callback.getStatus().message());
- flow_file->setAttribute(GCS_ERROR_REASON,
callback.getStatus().error_info().reason());
- flow_file->setAttribute(GCS_ERROR_DOMAIN,
callback.getStatus().error_info().domain());
+ session.setAttribute(flow_file, GCS_STATUS_MESSAGE,
callback.getStatus().message());
+ session.setAttribute(flow_file, GCS_ERROR_REASON,
callback.getStatus().error_info().reason());
+ session.setAttribute(flow_file, GCS_ERROR_DOMAIN,
callback.getStatus().error_info().domain());
logger_->log_error("Failed to fetch from Google Cloud Storage {} {}",
callback.getStatus().message(), callback.getStatus().error_info().reason());
- session.transfer(flow_file, Failure);
- return;
+ session.transfer(std::move(flow_file), Failure);
+ return MINIFI_STATUS_SUCCESS;
}
- if (auto generation = callback.getGeneration())
- flow_file->setAttribute(GCS_GENERATION, std::to_string(*generation));
- if (auto meta_generation = callback.getMetaGeneration())
- flow_file->setAttribute(GCS_META_GENERATION,
std::to_string(*meta_generation));
- if (auto storage_class = callback.getStorageClass())
- flow_file->setAttribute(GCS_STORAGE_CLASS, *storage_class);
- session.transfer(flow_file, Success);
+ if (auto generation = callback.getGeneration()) {
+ session.setAttribute(flow_file, GCS_GENERATION,
std::to_string(*generation));
+ }
+ if (auto meta_generation = callback.getMetaGeneration()) {
+ session.setAttribute(flow_file, GCS_META_GENERATION,
std::to_string(*meta_generation));
+ }
+ if (auto storage_class = callback.getStorageClass()) {
+ session.setAttribute(flow_file, GCS_STORAGE_CLASS, *storage_class);
+ }
+ session.transfer(std::move(flow_file), Success);
+ return MINIFI_STATUS_SUCCESS;
}
-REGISTER_RESOURCE(FetchGCSObject, Processor);
-
} // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/FetchGCSObject.h
b/extensions/gcp/processors/FetchGCSObject.h
index 694ad1f5e..51d72dab9 100644
--- a/extensions/gcp/processors/FetchGCSObject.h
+++ b/extensions/gcp/processors/FetchGCSObject.h
@@ -17,24 +17,20 @@
#pragma once
-#include <memory>
-#include <string>
-#include <utility>
-
#include "../GCPAttributes.h"
#include "GCSProcessor.h"
#include "minifi-cpp/core/PropertyDefinition.h"
#include "minifi-cpp/core/RelationshipDefinition.h"
#include "google/cloud/storage/well_known_headers.h"
-#include "core/logging/LoggerFactory.h"
#include "utils/ArrayUtils.h"
+#include "minifi-cpp/core/Annotation.h"
+
namespace org::apache::nifi::minifi::extensions::gcp {
class FetchGCSObject : public GCSProcessor {
public:
using GCSProcessor::GCSProcessor;
- ~FetchGCSObject() override = default;
EXTENSIONAPI static constexpr const char* Description = "Fetches a file from
a Google Cloud Bucket. Designed to be used in tandem with ListGCSBucket.";
@@ -80,11 +76,9 @@ class FetchGCSObject : public GCSProcessor {
EXTENSIONAPI static constexpr core::annotation::Input InputRequirement =
core::annotation::Input::INPUT_REQUIRED;
EXTENSIONAPI static constexpr bool IsSingleThreaded = false;
- ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS
-
- void initialize() override;
- void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&
session_factory) override;
- void onTrigger(core::ProcessContext& context, core::ProcessSession& session)
override;
+ protected:
+ MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
+ MinifiStatus onTriggerImpl(api::core::ProcessContext& context,
api::core::ProcessSession& session) override;
private:
google::cloud::storage::EncryptionKey encryption_key_;
diff --git a/extensions/gcp/processors/GCSProcessor.cpp
b/extensions/gcp/processors/GCSProcessor.cpp
index 274b3f93c..bf9342af8 100644
--- a/extensions/gcp/processors/GCSProcessor.cpp
+++ b/extensions/gcp/processors/GCSProcessor.cpp
@@ -17,45 +17,42 @@
#include "GCSProcessor.h"
-#include "utils/ProcessorConfigUtils.h"
-
#include "../controllerservices/GCPCredentialsControllerService.h"
-#include "minifi-cpp/core/ProcessContext.h"
-#include "core/ProcessSession.h"
+#include "api/utils/ProcessorConfigUtils.h"
namespace gcs = ::google::cloud::storage;
namespace org::apache::nifi::minifi::extensions::gcp {
-std::shared_ptr<google::cloud::Credentials>
GCSProcessor::getCredentials(core::ProcessContext& context) const {
- auto gcp_credentials_controller_service =
utils::parseOptionalControllerService<GCPCredentialsControllerService>(context,
GCSProcessor::GCPCredentials, getUUID());
- if (gcp_credentials_controller_service) {
+std::shared_ptr<google::cloud::Credentials> GCSProcessor::getCredentials(const
api::core::ProcessContext& context) {
+ if (const auto gcp_credentials_controller_service =
api::utils::parseOptionalControllerService<GCPCredentialsControllerService>(context,
GCPCredentials)) {
return gcp_credentials_controller_service->getCredentials();
}
return nullptr;
}
-void GCSProcessor::onSchedule(core::ProcessContext& context,
core::ProcessSessionFactory&) {
- if (auto number_of_retries = utils::parseOptionalU64Property(context,
NumberOfRetries)) {
+MinifiStatus GCSProcessor::onScheduleImpl(api::core::ProcessContext& context) {
+ if (const auto number_of_retries =
api::utils::parseOptionalU64Property(context, NumberOfRetries)) {
retry_policy_ =
std::make_shared<google::cloud::storage::LimitedErrorCountRetryPolicy>(gsl::narrow<int>(*number_of_retries));
}
gcp_credentials_ = getCredentials(context);
if (!gcp_credentials_) {
- throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION,
"Missing GCP Credentials");
+ logger_->log_error("Couldnt find valid credentials");
+ return MINIFI_STATUS_UNKNOWN_ERROR;
}
- endpoint_url_ = context.getProperty(EndpointOverrideURL) |
utils::toOptional();
+ endpoint_url_ = context.getProperty(EndpointOverrideURL, nullptr) |
utils::toOptional();
if (endpoint_url_)
logger_->log_debug("Endpoint overwritten: {}", *endpoint_url_);
- auto proxy_controller_service =
minifi::utils::parseOptionalControllerService<minifi::controllers::ProxyConfigurationServiceInterface>(context,
ProxyConfigurationService, getUUID());
- if (proxy_controller_service && proxy_controller_service->getProxyType() !=
minifi::controllers::ProxyType::DIRECT) {
+ const auto proxy_data = context.getProxyData(ProxyConfigurationService) |
utils::orThrow("Couldnt query ProxyConfigurationService");
+ if (proxy_data && proxy_data->proxy_type != api::utils::ProxyType::DIRECT) {
logger_->log_debug("Proxy configuration is set for GCS processor");
proxy_ = google::cloud::ProxyConfig{};
-
proxy_->set_scheme(minifi::utils::string::startsWith(proxy_controller_service->getHost(),
"https") ? "https" : "http");
- auto proxy_host = proxy_controller_service->getHost();
+ proxy_->set_scheme(minifi::utils::string::startsWith(proxy_data->host,
"https") ? "https" : "http");
+ auto proxy_host = proxy_data->host;
constexpr std::string_view https_prefix = "https://";;
constexpr std::string_view http_prefix = "http://";;
if (minifi::utils::string::startsWith(proxy_host, https_prefix)) {
@@ -64,12 +61,13 @@ void GCSProcessor::onSchedule(core::ProcessContext&
context, core::ProcessSessio
proxy_host = proxy_host.substr(http_prefix.size());
}
proxy_->set_hostname(proxy_host);
- proxy_->set_port(std::to_string(proxy_controller_service->getPort()));
- if (auto proxy_credentials =
proxy_controller_service->getProxyCredentials()) {
+ proxy_->set_port(std::to_string(proxy_data->port));
+ if (auto proxy_credentials = proxy_data->proxy_credentials) {
proxy_->set_username(proxy_credentials->username);
proxy_->set_password(proxy_credentials->password);
}
}
+ return MINIFI_STATUS_SUCCESS;
}
gcs::Client GCSProcessor::getClient() const {
diff --git a/extensions/gcp/processors/GCSProcessor.h
b/extensions/gcp/processors/GCSProcessor.h
index af6a92ace..a05dde421 100644
--- a/extensions/gcp/processors/GCSProcessor.h
+++ b/extensions/gcp/processors/GCSProcessor.h
@@ -16,24 +16,21 @@
*/
#pragma once
-#include <string>
#include <memory>
-#include <utility>
#include <optional>
+#include <string>
#include "../controllerservices/GCPCredentialsControllerService.h"
-#include "minifi-cpp/core/logging/Logger.h"
-#include "core/ProcessorImpl.h"
-#include "minifi-cpp/core/PropertyDefinition.h"
+#include "api/core/ProcessorImpl.h"
#include "core/PropertyDefinitionBuilder.h"
-#include "minifi-cpp/core/PropertyValidator.h"
#include "google/cloud/credentials.h"
#include "google/cloud/storage/client.h"
#include "google/cloud/storage/retry_policy.h"
-#include "minifi-cpp/controllers/ProxyConfigurationServiceInterface.h"
+#include "minifi-cpp/core/PropertyDefinition.h"
+#include "minifi-cpp/core/PropertyValidator.h"
namespace org::apache::nifi::minifi::extensions::gcp {
-class GCSProcessor : public core::ProcessorImpl {
+class GCSProcessor : public api::core::ProcessorImpl {
public:
using ProcessorImpl::ProcessorImpl;
@@ -56,7 +53,7 @@ class GCSProcessor : public core::ProcessorImpl {
.build();
EXTENSIONAPI static constexpr auto ProxyConfigurationService =
core::PropertyDefinitionBuilder<>::createProperty("Proxy Configuration Service")
.withDescription("Specifies the Proxy Configuration Controller Service
to proxy network requests.")
-
.withAllowedTypes<minifi::controllers::ProxyConfigurationServiceInterface>()
+ .withAllowedType<MINIFI_PROXY_CONFIGURATION_SERVICE_PROPERTY_TYPE>()
.build();
EXTENSIONAPI static constexpr auto Properties =
std::to_array<core::PropertyReference>({
GCPCredentials,
@@ -65,12 +62,11 @@ class GCSProcessor : public core::ProcessorImpl {
ProxyConfigurationService
});
-
- void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&
session_factory) override;
-
protected:
+ MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
+
virtual google::cloud::storage::Client getClient() const;
- std::shared_ptr<google::cloud::Credentials>
getCredentials(core::ProcessContext& context) const;
+ static std::shared_ptr<google::cloud::Credentials> getCredentials(const
api::core::ProcessContext& context);
std::optional<std::string> endpoint_url_;
std::shared_ptr<google::cloud::Credentials> gcp_credentials_;
diff --git a/extensions/gcp/processors/ListGCSBucket.cpp
b/extensions/gcp/processors/ListGCSBucket.cpp
index 13426e441..ac4cadd40 100644
--- a/extensions/gcp/processors/ListGCSBucket.cpp
+++ b/extensions/gcp/processors/ListGCSBucket.cpp
@@ -17,47 +17,44 @@
#include "ListGCSBucket.h"
-#include "utils/ProcessorConfigUtils.h"
-
#include "../GCPAttributes.h"
-#include "minifi-cpp/core/FlowFile.h"
-#include "minifi-cpp/core/ProcessContext.h"
-#include "core/ProcessSession.h"
-#include "core/Resource.h"
+#include "api/core/ProcessContext.h"
+#include "api/core/ProcessSession.h"
+#include "api/core/Resource.h"
+#include "api/utils/ProcessorConfigUtils.h"
+#include "minifi-cpp/core/SpecialFlowAttribute.h"
namespace gcs = ::google::cloud::storage;
namespace org::apache::nifi::minifi::extensions::gcp {
-void ListGCSBucket::initialize() {
- setSupportedProperties(Properties);
- setSupportedRelationships(Relationships);
-}
-
-void ListGCSBucket::onSchedule(core::ProcessContext& context,
core::ProcessSessionFactory& session_factory) {
- GCSProcessor::onSchedule(context, session_factory);
- bucket_ = utils::parseProperty(context, Bucket);
+MinifiStatus ListGCSBucket::onScheduleImpl(api::core::ProcessContext& context)
{
+ const auto status = GCSProcessor::onScheduleImpl(context);
+ if (status != MinifiStatus::MINIFI_STATUS_SUCCESS) {
+ return status;
+ }
+ bucket_ = api::utils::parseProperty(context, Bucket);
+ return MinifiStatus::MINIFI_STATUS_SUCCESS;
}
-void ListGCSBucket::onTrigger(core::ProcessContext& context,
core::ProcessSession& session) {
+MinifiStatus ListGCSBucket::onTriggerImpl(api::core::ProcessContext& context,
api::core::ProcessSession& session) {
gsl_Expects(gcp_credentials_);
gcs::Client client = getClient();
- auto list_all_versions = utils::parseOptionalBoolProperty(context,
ListAllVersions);
+ auto list_all_versions = api::utils::parseOptionalBoolProperty(context,
ListAllVersions);
gcs::Versions versions = (list_all_versions && *list_all_versions) ?
gcs::Versions(true) : gcs::Versions(false);
auto objects_in_bucket = client.ListObjects(bucket_, versions);
for (const auto& object_in_bucket : objects_in_bucket) {
if (object_in_bucket.ok()) {
auto flow_file = session.create();
- flow_file->updateAttribute(core::SpecialFlowAttribute::FILENAME,
object_in_bucket->name());
- setAttributesFromObjectMetadata(*flow_file, *object_in_bucket);
- session.transfer(flow_file, Success);
+ session.setAttribute(flow_file, core::SpecialFlowAttribute::FILENAME,
object_in_bucket->name());
+ setAttributesFromObjectMetadata(flow_file, *object_in_bucket, session);
+ session.transfer(std::move(flow_file), Success);
} else {
logger_->log_error("Invalid object in bucket {}", bucket_);
}
}
+ return MinifiStatus::MINIFI_STATUS_SUCCESS;
}
-REGISTER_RESOURCE(ListGCSBucket, Processor);
-
} // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/ListGCSBucket.h
b/extensions/gcp/processors/ListGCSBucket.h
index 36c4ac1e0..e654a314d 100644
--- a/extensions/gcp/processors/ListGCSBucket.h
+++ b/extensions/gcp/processors/ListGCSBucket.h
@@ -17,19 +17,17 @@
#pragma once
-#include <memory>
#include <string>
-#include <utility>
#include "../GCPAttributes.h"
#include "GCSProcessor.h"
-#include "core/logging/LoggerFactory.h"
#include "minifi-cpp/core/OutputAttributeDefinition.h"
#include "minifi-cpp/core/PropertyDefinition.h"
#include "core/PropertyDefinitionBuilder.h"
#include "minifi-cpp/core/PropertyValidator.h"
#include "minifi-cpp/core/RelationshipDefinition.h"
#include "utils/ArrayUtils.h"
+#include "minifi-cpp/core/Annotation.h"
namespace org::apache::nifi::minifi::extensions::gcp {
@@ -44,7 +42,6 @@ inline constexpr auto FILENAME_OUTPUT_ATTRIBUTE_DESCRIPTION =
utils::array_to_st
class ListGCSBucket : public GCSProcessor {
public:
using GCSProcessor::GCSProcessor;
- ~ListGCSBucket() override = default;
EXTENSIONAPI static constexpr const char* Description = "Retrieves a listing
of objects from an GCS bucket. "
"For each object that is listed, creates a FlowFile that represents the
object so that it can be fetched in conjunction with FetchGCSObject.";
@@ -120,11 +117,9 @@ class ListGCSBucket : public GCSProcessor {
EXTENSIONAPI static constexpr core::annotation::Input InputRequirement =
core::annotation::Input::INPUT_FORBIDDEN;
EXTENSIONAPI static constexpr bool IsSingleThreaded = true;
- ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS
-
- void initialize() override;
- void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&
session_factory) override;
- void onTrigger(core::ProcessContext& context, core::ProcessSession& session)
override;
+ protected:
+ MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
+ MinifiStatus onTriggerImpl(api::core::ProcessContext& context,
api::core::ProcessSession& session) override;
private:
std::string bucket_;
diff --git a/extensions/gcp/processors/PutGCSObject.cpp
b/extensions/gcp/processors/PutGCSObject.cpp
index a05380f32..97178eb9a 100644
--- a/extensions/gcp/processors/PutGCSObject.cpp
+++ b/extensions/gcp/processors/PutGCSObject.cpp
@@ -19,12 +19,12 @@
#include <utility>
-#include "core/Resource.h"
-#include "minifi-cpp/core/FlowFile.h"
-#include "minifi-cpp/core/ProcessContext.h"
-#include "core/ProcessSession.h"
#include "../GCPAttributes.h"
-#include "utils/ProcessorConfigUtils.h"
+#include "api/core/ProcessContext.h"
+#include "api/core/ProcessSession.h"
+#include "api/core/Resource.h"
+#include "api/utils/ProcessorConfigUtils.h"
+#include "minifi-cpp/io/InputStream.h"
namespace gcs = ::google::cloud::storage;
@@ -102,80 +102,79 @@ class UploadToGCSCallback {
} // namespace
-void PutGCSObject::initialize() {
- setSupportedProperties(Properties);
- setSupportedRelationships(Relationships);
-}
-
+MinifiStatus PutGCSObject::onScheduleImpl(api::core::ProcessContext& context) {
+ const auto status = GCSProcessor::onScheduleImpl(context);
+ if (status != MinifiStatus::MINIFI_STATUS_SUCCESS) {
+ return status;
+ }
-void PutGCSObject::onSchedule(core::ProcessContext& context,
core::ProcessSessionFactory& session_factory) {
- GCSProcessor::onSchedule(context, session_factory);
- if (auto encryption_key = context.getProperty(EncryptionKey)) {
+ if (auto encryption_key = context.getProperty(EncryptionKey, nullptr)) {
try {
encryption_key_ = gcs::EncryptionKey::FromBase64Key(*encryption_key);
} catch (const google::cloud::RuntimeStatusError&) {
- throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION,
"Could not decode the base64-encoded encryption key from property " +
std::string(EncryptionKey.name));
+ logger_->log_error("Could not decode the base64-encoded encryption key
from property {}", std::string(EncryptionKey.name));
+ return MINIFI_STATUS_UNKNOWN_ERROR;
}
}
+ return MINIFI_STATUS_SUCCESS;
}
-void PutGCSObject::onTrigger(core::ProcessContext& context,
core::ProcessSession& session) {
+MinifiStatus PutGCSObject::onTriggerImpl(api::core::ProcessContext& context,
api::core::ProcessSession& session) {
gsl_Expects(gcp_credentials_);
auto flow_file = session.get();
if (!flow_file) {
- context.yield();
- return;
+ return MINIFI_STATUS_PROCESSOR_YIELD;
}
- auto bucket = context.getProperty(Bucket, flow_file.get());
+ auto bucket = api::utils::parseOptionalProperty(context, Bucket, &flow_file);
+
if (!bucket || bucket->empty()) {
logger_->log_error("Missing bucket name");
- session.transfer(flow_file, Failure);
- return;
+ session.transfer(std::move(flow_file), Failure);
+ return MINIFI_STATUS_SUCCESS;
}
- auto object_name = context.getProperty(Key, flow_file.get());
+ auto object_name = api::utils::parseOptionalProperty(context, Key,
&flow_file);
if (!object_name || object_name->empty()) {
logger_->log_error("Missing object name");
- session.transfer(flow_file, Failure);
- return;
+ session.transfer(std::move(flow_file), Failure);
+ return MINIFI_STATUS_SUCCESS;
}
gcs::Client client = getClient();
UploadToGCSCallback callback(client, *bucket, *object_name);
- if (auto crc32_checksum = context.getProperty(Crc32cChecksum,
flow_file.get())) {
+ if (auto crc32_checksum = api::utils::parseOptionalProperty(context,
Crc32cChecksum, &flow_file)) {
callback.setCrc32CChecksumValue(*crc32_checksum);
}
- if (auto md5_hash = context.getProperty(MD5Hash, flow_file.get())) {
+ if (auto md5_hash = api::utils::parseOptionalProperty(context, MD5Hash,
&flow_file)) {
callback.setHashValue(*md5_hash);
}
- auto content_type = context.getProperty(ContentType, flow_file.get());
+ auto content_type = api::utils::parseOptionalProperty(context, ContentType,
&flow_file);
if (content_type && !content_type->empty())
callback.setContentType(*content_type);
- if (auto predefined_acl =
utils::parseOptionalEnumProperty<put_gcs_object::PredefinedAcl>(context,
ObjectACL))
+ if (auto predefined_acl =
api::utils::parseOptionalEnumProperty<put_gcs_object::PredefinedAcl>(context,
ObjectACL))
callback.setPredefinedAcl(*predefined_acl);
- callback.setIfGenerationMatch(utils::parseOptionalBoolProperty(context,
OverwriteObject));
+ callback.setIfGenerationMatch(api::utils::parseOptionalBoolProperty(context,
OverwriteObject));
callback.setEncryptionKey(encryption_key_);
session.read(flow_file, std::ref(callback));
auto& result = callback.getResult();
if (!result.ok()) {
- flow_file->setAttribute(GCS_STATUS_MESSAGE, result.status().message());
- flow_file->setAttribute(GCS_ERROR_REASON,
result.status().error_info().reason());
- flow_file->setAttribute(GCS_ERROR_DOMAIN,
result.status().error_info().domain());
+ session.setAttribute(flow_file, GCS_STATUS_MESSAGE,
result.status().message());
+ session.setAttribute(flow_file, GCS_ERROR_REASON,
result.status().error_info().reason());
+ session.setAttribute(flow_file, GCS_ERROR_DOMAIN,
result.status().error_info().domain());
logger_->log_error("Failed to upload to Google Cloud Storage {} {}",
result.status().message(), result.status().error_info().reason());
- session.transfer(flow_file, Failure);
+ session.transfer(std::move(flow_file), Failure);
} else {
- setAttributesFromObjectMetadata(*flow_file, *result);
- session.transfer(flow_file, Success);
+ setAttributesFromObjectMetadata(flow_file, *result, session);
+ session.transfer(std::move(flow_file), Success);
}
+ return MINIFI_STATUS_SUCCESS;
}
-REGISTER_RESOURCE(PutGCSObject, Processor);
-
} // namespace org::apache::nifi::minifi::extensions::gcp
diff --git a/extensions/gcp/processors/PutGCSObject.h
b/extensions/gcp/processors/PutGCSObject.h
index 58d7675fa..73d8c8218 100644
--- a/extensions/gcp/processors/PutGCSObject.h
+++ b/extensions/gcp/processors/PutGCSObject.h
@@ -19,17 +19,16 @@
#include <memory>
#include <string>
-#include <utility>
#include "../GCPAttributes.h"
#include "GCSProcessor.h"
#include "minifi-cpp/core/PropertyDefinition.h"
#include "minifi-cpp/core/PropertyValidator.h"
#include "minifi-cpp/core/RelationshipDefinition.h"
-#include "core/logging/LoggerFactory.h"
#include "utils/ArrayUtils.h"
#include "utils/Enum.h"
#include "google/cloud/storage/well_known_headers.h"
+#include "minifi-cpp/core/Annotation.h"
namespace org::apache::nifi::minifi::extensions::gcp::put_gcs_object {
enum class PredefinedAcl {
@@ -73,7 +72,6 @@ namespace org::apache::nifi::minifi::extensions::gcp {
class PutGCSObject : public GCSProcessor {
public:
using GCSProcessor::GCSProcessor;
- ~PutGCSObject() override = default;
EXTENSIONAPI static constexpr const char* Description = "Puts flow files to
a Google Cloud Storage Bucket.";
@@ -191,11 +189,9 @@ class PutGCSObject : public GCSProcessor {
EXTENSIONAPI static constexpr core::annotation::Input InputRequirement =
core::annotation::Input::INPUT_REQUIRED;
EXTENSIONAPI static constexpr bool IsSingleThreaded = false;
- ADD_COMMON_VIRTUAL_FUNCTIONS_FOR_PROCESSORS
-
- void initialize() override;
- void onSchedule(core::ProcessContext& context, core::ProcessSessionFactory&
session_factory) override;
- void onTrigger(core::ProcessContext& context, core::ProcessSession& session)
override;
+ protected:
+ MinifiStatus onScheduleImpl(api::core::ProcessContext& context) override;
+ MinifiStatus onTriggerImpl(api::core::ProcessContext& context,
api::core::ProcessSession& session) override;
private:
google::cloud::storage::EncryptionKey encryption_key_;
diff --git a/extensions/gcp/tests/CMakeLists.txt
b/extensions/gcp/tests/CMakeLists.txt
index 0acc5241b..89fe308e9 100644
--- a/extensions/gcp/tests/CMakeLists.txt
+++ b/extensions/gcp/tests/CMakeLists.txt
@@ -36,8 +36,8 @@ FOREACH(testfile ${GCS_TESTS})
createTests("${testfilename}")
target_link_libraries(${testfilename} minifi-gcp)
- target_link_libraries(${testfilename} minifi-standard-processors)
target_link_libraries(${testfilename} gtest_main gmock)
+ target_link_libraries(${testfilename} libminifi-c-unittest)
gtest_add_tests(TARGET "${testfilename}")
ENDFOREACH()
diff --git a/extensions/gcp/tests/DeleteGCSObjectTests.cpp
b/extensions/gcp/tests/DeleteGCSObjectTests.cpp
index bde7e010f..aee457a60 100644
--- a/extensions/gcp/tests/DeleteGCSObjectTests.cpp
+++ b/extensions/gcp/tests/DeleteGCSObjectTests.cpp
@@ -14,60 +14,65 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "../processors/DeleteGCSObject.h"
#include "../controllerservices/GCPCredentialsControllerService.h"
+#include "../processors/DeleteGCSObject.h"
+#include "CProcessorTestUtils.h"
#include "GCPAttributes.h"
#include "core/Resource.h"
-#include "unit/SingleProcessorTestController.h"
-#include "google/cloud/storage/testing/mock_client.h"
#include "google/cloud/storage/internal/object_metadata_parser.h"
#include "google/cloud/storage/testing/canonical_errors.h"
+#include "google/cloud/storage/testing/mock_client.h"
#include "unit/ProcessorUtils.h"
+#include "unit/SingleProcessorTestController.h"
namespace gcs = ::google::cloud::storage;
-namespace minifi_gcp = org::apache::nifi::minifi::extensions::gcp;
+namespace minifi_gcp = minifi::extensions::gcp;
-using DeleteGCSObject =
org::apache::nifi::minifi::extensions::gcp::DeleteGCSObject;
-using GCPCredentialsControllerService =
org::apache::nifi::minifi::extensions::gcp::GCPCredentialsControllerService;
+using DeleteGCSObject = minifi::extensions::gcp::DeleteGCSObject;
+using GCPCredentialsControllerService =
minifi::extensions::gcp::GCPCredentialsControllerService;
using DeleteObjectRequest = gcs::internal::DeleteObjectRequest;
-using ::google::cloud::storage::testing::canonical_errors::TransientError;
using ::google::cloud::storage::testing::canonical_errors::PermanentError;
+using ::google::cloud::storage::testing::canonical_errors::TransientError;
namespace {
class DeleteGCSObjectMocked : public DeleteGCSObject {
- using
org::apache::nifi::minifi::extensions::gcp::DeleteGCSObject::DeleteGCSObject;
+ using DeleteGCSObject::DeleteGCSObject;
+
public:
- static constexpr const char* Description = "DeleteGCSObjectMocked";
+ DeleteGCSObjectMocked(minifi::core::ProcessorMetadata metadata,
std::shared_ptr<gcs::testing::MockClient> mock_client)
+ : DeleteGCSObject(std::move(metadata)),
+ mock_client_(std::move(mock_client)) {}
- gcs::Client getClient() const override {
- return gcs::testing::UndecoratedClientFromMock(mock_client_);
- }
- std::shared_ptr<gcs::testing::MockClient> mock_client_ =
std::make_shared<gcs::testing::MockClient>();
+ protected:
+ gcs::Client getClient() const override { return
gcs::testing::UndecoratedClientFromMock(mock_client_); }
+ std::shared_ptr<gcs::testing::MockClient> mock_client_;
};
-REGISTER_RESOURCE(DeleteGCSObjectMocked, Processor);
} // namespace
class DeleteGCSObjectTests : public ::testing::Test {
- public:
+ protected:
void SetUp() override {
- delete_gcs_object_ = test_controller_.getProcessor();
- gcp_credentials_node_ =
test_controller_.plan->addController("GCPCredentialsControllerService",
"gcp_credentials_controller_service");
- test_controller_.plan->setProperty(gcp_credentials_node_,
-
GCPCredentialsControllerService::CredentialsLoc,
-
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS));
- test_controller_.plan->setProperty(delete_gcs_object_,
- DeleteGCSObject::GCPCredentials,
- "gcp_credentials_controller_service");
+ const auto gcp_credential_controller_service =
+
minifi::test::utils::make_custom_c_controller_service<GCPCredentialsControllerService>(core::ControllerServiceMetadata{utils::Identifier{},
+ "GCPCredentialsControllerService",
+
logging::LoggerFactory<GCPCredentialsControllerService>::getLogger()});
+ gcp_credentials_node_ =
test_controller_.plan->addController("gcp_credentials_controller_service",
gcp_credential_controller_service);
+
EXPECT_TRUE(gcp_credential_controller_service->setProperty(GCPCredentialsControllerService::CredentialsLoc.name,
+
std::string(magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS))));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(DeleteGCSObject::GCPCredentials.name,
"gcp_credentials_controller_service"));
}
- org::apache::nifi::minifi::test::SingleProcessorTestController
test_controller_{minifi::test::utils::make_processor<DeleteGCSObjectMocked>("DeleteGCSObjectMocked")};
- std::shared_ptr<minifi::core::controller::ControllerServiceNode>
gcp_credentials_node_;
- TypedProcessorWrapper<DeleteGCSObjectMocked> delete_gcs_object_ = nullptr;
+ public:
+ std::shared_ptr<gcs::testing::MockClient> mock_client_ =
std::make_shared<gcs::testing::MockClient>();
+ minifi::test::SingleProcessorTestController
test_controller_{minifi::test::utils::make_custom_c_processor<DeleteGCSObjectMocked>(
+ core::ProcessorMetadata{utils::Identifier{}, "DeleteGCSObjectMocked",
logging::LoggerFactory<DeleteGCSObjectMocked>::getLogger()},
+ mock_client_)};
+ std::shared_ptr<core::controller::ControllerServiceNode>
gcp_credentials_node_;
};
TEST_F(DeleteGCSObjectTests, MissingBucket) {
- EXPECT_CALL(*delete_gcs_object_.get().mock_client_,
CreateResumableUpload).Times(0);
- EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_,
DeleteGCSObject::Bucket, ""));
+ EXPECT_CALL(*mock_client_, CreateResumableUpload).Times(0);
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(DeleteGCSObject::Bucket.name,
""));
const auto& result = test_controller_.trigger("hello world");
EXPECT_EQ(0, result.at(DeleteGCSObject::Success).size());
ASSERT_EQ(1, result.at(DeleteGCSObject::Failure).size());
@@ -77,9 +82,9 @@ TEST_F(DeleteGCSObjectTests, MissingBucket) {
}
TEST_F(DeleteGCSObjectTests, ServerGivesPermaError) {
- EXPECT_CALL(*delete_gcs_object_.get().mock_client_, DeleteObject)
+ EXPECT_CALL(*mock_client_, DeleteObject)
.WillOnce(testing::Return(PermanentError()));
- EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_,
DeleteGCSObject::Bucket, "bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(DeleteGCSObject::Bucket.name,
"bucket-from-property"));
const auto& result = test_controller_.trigger("hello world");
EXPECT_EQ(0, result.at(DeleteGCSObject::Success).size());
ASSERT_EQ(1, result.at(DeleteGCSObject::Failure).size());
@@ -89,9 +94,9 @@ TEST_F(DeleteGCSObjectTests, ServerGivesPermaError) {
}
TEST_F(DeleteGCSObjectTests, ServerGivesTransientErrors) {
- EXPECT_CALL(*delete_gcs_object_.get().mock_client_,
DeleteObject).WillOnce(testing::Return(TransientError()));
- EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_,
DeleteGCSObject::NumberOfRetries, "1"));
- EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_,
DeleteGCSObject::Bucket, "bucket-from-property"));
+ EXPECT_CALL(*mock_client_,
DeleteObject).WillOnce(testing::Return(TransientError()));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(DeleteGCSObject::NumberOfRetries.name,
"1"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(DeleteGCSObject::Bucket.name,
"bucket-from-property"));
const auto& result = test_controller_.trigger("hello world",
{{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"}});
EXPECT_EQ(0, result.at(DeleteGCSObject::Success).size());
ASSERT_EQ(1, result.at(DeleteGCSObject::Failure).size());
@@ -100,9 +105,8 @@ TEST_F(DeleteGCSObjectTests, ServerGivesTransientErrors) {
EXPECT_EQ("hello world",
test_controller_.plan->getContent(result.at(DeleteGCSObject::Failure)[0]));
}
-
TEST_F(DeleteGCSObjectTests, HandlingSuccessfullDeletion) {
- EXPECT_CALL(*delete_gcs_object_.get().mock_client_, DeleteObject)
+ EXPECT_CALL(*mock_client_, DeleteObject)
.WillOnce([](DeleteObjectRequest const& request) {
EXPECT_EQ("bucket-from-attribute", request.bucket_name());
EXPECT_TRUE(request.HasOption<gcs::Generation>());
@@ -110,7 +114,7 @@ TEST_F(DeleteGCSObjectTests, HandlingSuccessfullDeletion) {
EXPECT_EQ(23, request.GetOption<gcs::Generation>().value());
return google::cloud::make_status_or(gcs::internal::EmptyResponse{});
});
- EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_,
DeleteGCSObject::ObjectGeneration, "${gcs.generation}"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(DeleteGCSObject::ObjectGeneration.name,
"${gcs.generation}"));
const auto& result = test_controller_.trigger("hello world",
{{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"},
{std::string(minifi_gcp::GCS_GENERATION), "23"}});
ASSERT_EQ(1, result.at(DeleteGCSObject::Success).size());
EXPECT_EQ(0, result.at(DeleteGCSObject::Failure).size());
@@ -118,13 +122,13 @@ TEST_F(DeleteGCSObjectTests, HandlingSuccessfullDeletion)
{
}
TEST_F(DeleteGCSObjectTests, EmptyGeneration) {
- EXPECT_CALL(*delete_gcs_object_.get().mock_client_, DeleteObject)
+ EXPECT_CALL(*mock_client_, DeleteObject)
.WillOnce([](DeleteObjectRequest const& request) {
EXPECT_EQ("bucket-from-attribute", request.bucket_name());
EXPECT_FALSE(request.HasOption<gcs::Generation>());
return google::cloud::make_status_or(gcs::internal::EmptyResponse{});
});
- EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_,
DeleteGCSObject::ObjectGeneration, "${gcs.generation}"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(DeleteGCSObject::ObjectGeneration.name,
"${gcs.generation}"));
const auto& result = test_controller_.trigger("hello world",
{{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"}});
ASSERT_EQ(1, result.at(DeleteGCSObject::Success).size());
EXPECT_EQ(0, result.at(DeleteGCSObject::Failure).size());
@@ -132,7 +136,7 @@ TEST_F(DeleteGCSObjectTests, EmptyGeneration) {
}
TEST_F(DeleteGCSObjectTests, InvalidGeneration) {
- EXPECT_TRUE(test_controller_.plan->setProperty(delete_gcs_object_,
DeleteGCSObject::ObjectGeneration, "${gcs.generation}"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(DeleteGCSObject::ObjectGeneration.name,
"${gcs.generation}"));
const auto& result = test_controller_.trigger("hello world",
{{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"},
{std::string(minifi_gcp::GCS_GENERATION), "23 banana"}});
ASSERT_EQ(0, result.at(DeleteGCSObject::Success).size());
EXPECT_EQ(1, result.at(DeleteGCSObject::Failure).size());
diff --git a/extensions/gcp/tests/FetchGCSObjectTests.cpp
b/extensions/gcp/tests/FetchGCSObjectTests.cpp
index ea87c7e63..764d57ffe 100644
--- a/extensions/gcp/tests/FetchGCSObjectTests.cpp
+++ b/extensions/gcp/tests/FetchGCSObjectTests.cpp
@@ -14,56 +14,63 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "../processors/FetchGCSObject.h"
+#include <memory>
+
#include "../controllerservices/GCPCredentialsControllerService.h"
+#include "../processors/FetchGCSObject.h"
+#include "CProcessorTestUtils.h"
#include "GCPAttributes.h"
#include "core/Resource.h"
-#include "unit/SingleProcessorTestController.h"
-#include "google/cloud/storage/testing/mock_client.h"
#include "google/cloud/storage/internal/object_metadata_parser.h"
#include "google/cloud/storage/testing/canonical_errors.h"
+#include "google/cloud/storage/testing/mock_client.h"
#include "unit/ProcessorUtils.h"
+#include "unit/SingleProcessorTestController.h"
namespace gcs = ::google::cloud::storage;
-namespace minifi_gcp = org::apache::nifi::minifi::extensions::gcp;
+namespace minifi_gcp = minifi::extensions::gcp;
-using FetchGCSObject =
org::apache::nifi::minifi::extensions::gcp::FetchGCSObject;
-using GCPCredentialsControllerService =
org::apache::nifi::minifi::extensions::gcp::GCPCredentialsControllerService;
+using FetchGCSObject = minifi::extensions::gcp::FetchGCSObject;
+using GCPCredentialsControllerService =
minifi::extensions::gcp::GCPCredentialsControllerService;
namespace {
class FetchGCSObjectMocked : public FetchGCSObject {
- using
org::apache::nifi::minifi::extensions::gcp::FetchGCSObject::FetchGCSObject;
+ using FetchGCSObject::FetchGCSObject;
+
public:
- static constexpr const char* Description = "FetchGCSObjectMocked";
+ FetchGCSObjectMocked(minifi::core::ProcessorMetadata metadata,
std::shared_ptr<gcs::testing::MockClient> mock_client)
+ : FetchGCSObject(std::move(metadata)),
+ mock_client_(std::move(mock_client)) {}
- gcs::Client getClient() const override {
- return gcs::testing::UndecoratedClientFromMock(mock_client_);
- }
- std::shared_ptr<gcs::testing::MockClient> mock_client_ =
std::make_shared<gcs::testing::MockClient>();
+ protected:
+ gcs::Client getClient() const override { return
gcs::testing::UndecoratedClientFromMock(mock_client_); }
+ std::shared_ptr<gcs::testing::MockClient> mock_client_;
};
-REGISTER_RESOURCE(FetchGCSObjectMocked, Processor);
} // namespace
class FetchGCSObjectTests : public ::testing::Test {
- public:
+ protected:
void SetUp() override {
- fetch_gcs_object_ = test_controller_.getProcessor();
- gcp_credentials_node_ =
test_controller_.plan->addController("GCPCredentialsControllerService",
"gcp_credentials_controller_service");
- test_controller_.plan->setProperty(gcp_credentials_node_,
-
GCPCredentialsControllerService::CredentialsLoc,
-
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS));
- test_controller_.plan->setProperty(fetch_gcs_object_,
- FetchGCSObject::GCPCredentials,
- "gcp_credentials_controller_service");
+ const auto gcp_credential_controller_service =
+
minifi::test::utils::make_custom_c_controller_service<GCPCredentialsControllerService>(core::ControllerServiceMetadata{utils::Identifier{},
+ "GCPCredentialsControllerService",
+
logging::LoggerFactory<GCPCredentialsControllerService>::getLogger()});
+ gcp_credentials_node_ =
test_controller_.plan->addController("gcp_credentials_controller_service",
gcp_credential_controller_service);
+
EXPECT_TRUE(gcp_credential_controller_service->setProperty(GCPCredentialsControllerService::CredentialsLoc.name,
+
std::string(magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS))));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(FetchGCSObjectMocked::GCPCredentials.name,
"gcp_credentials_controller_service"));
}
- TypedProcessorWrapper<FetchGCSObjectMocked> fetch_gcs_object_;
- org::apache::nifi::minifi::test::SingleProcessorTestController
test_controller_{minifi::test::utils::make_processor<FetchGCSObjectMocked>("FetchGCSObjectMocked")};
- std::shared_ptr<minifi::core::controller::ControllerServiceNode>
gcp_credentials_node_;
+
+ public:
+ std::shared_ptr<gcs::testing::MockClient> mock_client_ =
std::make_shared<gcs::testing::MockClient>();
+ minifi::test::SingleProcessorTestController
test_controller_{minifi::test::utils::make_custom_c_processor<FetchGCSObjectMocked>(
+ core::ProcessorMetadata{utils::Identifier{}, "FetchGCSObjectMocked",
logging::LoggerFactory<FetchGCSObjectMocked>::getLogger()}, mock_client_)};
+ std::shared_ptr<minifi::core::controller::ControllerServiceNode>
gcp_credentials_node_;
};
TEST_F(FetchGCSObjectTests, MissingBucket) {
- EXPECT_CALL(*fetch_gcs_object_.get().mock_client_,
CreateResumableUpload).Times(0);
- EXPECT_TRUE(test_controller_.plan->setProperty(fetch_gcs_object_,
FetchGCSObject::Bucket, ""));
+ EXPECT_CALL(*mock_client_, CreateResumableUpload).Times(0);
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(FetchGCSObject::Bucket.name,
""));
const auto& result = test_controller_.trigger("hello world");
EXPECT_EQ(0, result.at(FetchGCSObject::Success).size());
ASSERT_EQ(1, result.at(FetchGCSObject::Failure).size());
@@ -73,7 +80,7 @@ TEST_F(FetchGCSObjectTests, MissingBucket) {
}
TEST_F(FetchGCSObjectTests, ServerError) {
- EXPECT_CALL(*fetch_gcs_object_.get().mock_client_, ReadObject)
+ EXPECT_CALL(*mock_client_, ReadObject)
.WillOnce([](gcs::internal::ReadObjectRangeRequest const& request) {
EXPECT_EQ(request.bucket_name(), "bucket-from-property") << request;
auto mock_source =
std::make_unique<gcs::testing::MockObjectReadSource>();
@@ -88,7 +95,7 @@ TEST_F(FetchGCSObjectTests, ServerError) {
std::unique_ptr<gcs::internal::ObjectReadSource> object_read_source =
std::move(mock_source);
return google::cloud::make_status_or(std::move(object_read_source));
});
- EXPECT_TRUE(test_controller_.plan->setProperty(fetch_gcs_object_,
FetchGCSObject::Bucket, "bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(FetchGCSObject::Bucket.name,
"bucket-from-property"));
const auto& result = test_controller_.trigger("hello world",
{{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"}});
EXPECT_EQ(0, result.at(FetchGCSObject::Success).size());
ASSERT_EQ(1, result.at(FetchGCSObject::Failure).size());
@@ -97,7 +104,7 @@ TEST_F(FetchGCSObjectTests, ServerError) {
}
TEST_F(FetchGCSObjectTests, HappyPath) {
- std::string const text = "stored text";
+ constexpr std::string_view text = "stored text";
std::size_t offset = 0;
// Simulate a Read() call in the MockObjectReadSource object created below
auto simulate_read = [&text, &offset](void* buf, std::size_t n) {
@@ -107,13 +114,13 @@ TEST_F(FetchGCSObjectTests, HappyPath) {
return gcs::internal::ReadSourceResult{
l, gcs::internal::HttpResponse{200, {}, {}}};
};
- EXPECT_CALL(*fetch_gcs_object_.get().mock_client_, ReadObject)
+ EXPECT_CALL(*mock_client_, ReadObject)
.WillOnce([&](gcs::internal::ReadObjectRangeRequest const& request) {
EXPECT_EQ(request.bucket_name(), "bucket-from-attribute") << request;
EXPECT_TRUE(request.HasOption<gcs::Generation>());
EXPECT_TRUE(request.GetOption<gcs::Generation>().has_value());
EXPECT_EQ(23, request.GetOption<gcs::Generation>().value());
- std::unique_ptr<gcs::testing::MockObjectReadSource> mock_source(new
gcs::testing::MockObjectReadSource);
+ auto mock_source =
std::make_unique<gcs::testing::MockObjectReadSource>();
::testing::InSequence seq;
EXPECT_CALL(*mock_source,
IsOpen()).WillRepeatedly(testing::Return(true));
EXPECT_CALL(*mock_source, Read).WillOnce(simulate_read);
@@ -123,7 +130,7 @@ TEST_F(FetchGCSObjectTests, HappyPath) {
std::unique_ptr<gcs::internal::ObjectReadSource>(
std::move(mock_source)));
});
- EXPECT_TRUE(test_controller_.plan->setProperty(fetch_gcs_object_,
FetchGCSObject::ObjectGeneration, "${gcs.generation}"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(FetchGCSObject::ObjectGeneration.name,
"${gcs.generation}"));
const auto& result = test_controller_.trigger("hello world",
{{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"},
{std::string(minifi_gcp::GCS_GENERATION), "23"}});
ASSERT_EQ(1, result.at(FetchGCSObject::Success).size());
EXPECT_EQ(0, result.at(FetchGCSObject::Failure).size());
@@ -131,7 +138,7 @@ TEST_F(FetchGCSObjectTests, HappyPath) {
}
TEST_F(FetchGCSObjectTests, EmptyGeneration) {
- std::string const text = "stored text";
+ constexpr std::string_view text = "stored text";
std::size_t offset = 0;
// Simulate a Read() call in the MockObjectReadSource object created below
auto simulate_read = [&text, &offset](void* buf, std::size_t n) {
@@ -141,11 +148,11 @@ TEST_F(FetchGCSObjectTests, EmptyGeneration) {
return gcs::internal::ReadSourceResult{
l, gcs::internal::HttpResponse{200, {}, {}}};
};
- EXPECT_CALL(*fetch_gcs_object_.get().mock_client_, ReadObject)
+ EXPECT_CALL(*mock_client_, ReadObject)
.WillOnce([&](gcs::internal::ReadObjectRangeRequest const& request) {
EXPECT_EQ(request.bucket_name(), "bucket-from-attribute") << request;
EXPECT_FALSE(request.HasOption<gcs::Generation>());
- std::unique_ptr<gcs::testing::MockObjectReadSource> mock_source(new
gcs::testing::MockObjectReadSource);
+ auto mock_source =
std::make_unique<gcs::testing::MockObjectReadSource>();
::testing::InSequence seq;
EXPECT_CALL(*mock_source,
IsOpen()).WillRepeatedly(testing::Return(true));
EXPECT_CALL(*mock_source, Read).WillOnce(simulate_read);
@@ -155,7 +162,7 @@ TEST_F(FetchGCSObjectTests, EmptyGeneration) {
std::unique_ptr<gcs::internal::ObjectReadSource>(
std::move(mock_source)));
});
- EXPECT_TRUE(test_controller_.plan->setProperty(fetch_gcs_object_,
FetchGCSObject::ObjectGeneration, "${gcs.generation}"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(FetchGCSObject::ObjectGeneration.name,
"${gcs.generation}"));
const auto& result = test_controller_.trigger("hello world",
{{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"}});
ASSERT_EQ(1, result.at(FetchGCSObject::Success).size());
EXPECT_EQ(0, result.at(FetchGCSObject::Failure).size());
@@ -163,7 +170,7 @@ TEST_F(FetchGCSObjectTests, EmptyGeneration) {
}
TEST_F(FetchGCSObjectTests, InvalidGeneration) {
- EXPECT_TRUE(test_controller_.plan->setProperty(fetch_gcs_object_,
FetchGCSObject::ObjectGeneration, "${gcs.generation}"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(FetchGCSObject::ObjectGeneration.name,
"${gcs.generation}"));
const auto& result = test_controller_.trigger("hello world",
{{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"},
{std::string(minifi_gcp::GCS_GENERATION), "23 banana"}});
ASSERT_EQ(0, result.at(FetchGCSObject::Success).size());
EXPECT_EQ(1, result.at(FetchGCSObject::Failure).size());
diff --git a/extensions/gcp/tests/GCPCredentialsControllerServiceTests.cpp
b/extensions/gcp/tests/GCPCredentialsControllerServiceTests.cpp
index 0334cb4d9..bedbac924 100644
--- a/extensions/gcp/tests/GCPCredentialsControllerServiceTests.cpp
+++ b/extensions/gcp/tests/GCPCredentialsControllerServiceTests.cpp
@@ -16,19 +16,20 @@
*/
#define EXTENSION_LIST "minifi-gcp" // NOLINT(cppcoreguidelines-macro-usage)
-#include "unit/TestBase.h"
-#include "gtest/gtest.h"
#include "../controllerservices/GCPCredentialsControllerService.h"
-#include "core/Resource.h"
+#include "CProcessorTestUtils.h"
#include "core/Processor.h"
+#include "core/Resource.h"
#include "core/controller/ControllerServiceNode.h"
+#include "gtest/gtest.h"
#include "rapidjson/document.h"
#include "rapidjson/stream.h"
#include "rapidjson/writer.h"
#include "unit/DummyProcessor.h"
+#include "unit/TestBase.h"
#include "utils/Environment.h"
-namespace minifi_gcp = org::apache::nifi::minifi::extensions::gcp;
+namespace minifi_gcp = minifi::extensions::gcp;
using GCPCredentialsControllerService =
minifi_gcp::GCPCredentialsControllerService;
namespace {
@@ -70,14 +71,20 @@ std::optional<std::filesystem::path>
create_mock_json_file(const std::filesystem
class GCPCredentialsTests : public ::testing::Test {
protected:
void SetUp() override {
- ASSERT_TRUE(gcp_credentials_node_);
- ASSERT_TRUE(gcp_credentials_);
plan_->addProcessor("DummyProcessor", "dummy_processor");
+ auto gcp_credential_controller_service =
+
minifi::test::utils::make_custom_c_controller_service<GCPCredentialsControllerService>(core::ControllerServiceMetadata{utils::Identifier{},
+ "GCPCredentialsControllerService",
+
logging::LoggerFactory<GCPCredentialsControllerService>::getLogger()});
+ gcp_credentials_node_ =
plan_->addController("gcp_credentials_controller_service",
gcp_credential_controller_service);
+ gcp_credentials_ = static_cast<GCPCredentialsControllerService*>(
+
gcp_credentials_node_->getControllerServiceImplementation<utils::CControllerService>()->getImpl());
}
TestController test_controller_;
std::shared_ptr<TestPlan> plan_ = test_controller_.createPlan();
- std::shared_ptr<minifi::core::controller::ControllerServiceNode>
gcp_credentials_node_ = plan_->addController("GCPCredentialsControllerService",
"gcp_credentials_controller_service");
- std::shared_ptr<GCPCredentialsControllerService> gcp_credentials_ =
gcp_credentials_node_->getControllerServiceImplementation<GCPCredentialsControllerService>();
+
+ std::shared_ptr<minifi::core::controller::ControllerServiceNode>
gcp_credentials_node_;
+ GCPCredentialsControllerService* gcp_credentials_ = nullptr;
};
TEST_F(GCPCredentialsTests, DefaultGCPCredentialsWithEnv) {
@@ -85,13 +92,17 @@ TEST_F(GCPCredentialsTests, DefaultGCPCredentialsWithEnv) {
auto path = create_mock_json_file(temp_directory);
ASSERT_TRUE(path.has_value());
minifi::utils::Environment::setEnvironmentVariable("GOOGLE_APPLICATION_CREDENTIALS",
path->string().c_str());
- plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::CredentialsLoc,
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_DEFAULT_CREDENTIALS));
+ EXPECT_TRUE(plan_->setProperty(gcp_credentials_node_,
+ GCPCredentialsControllerService::CredentialsLoc,
+
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_DEFAULT_CREDENTIALS)));
ASSERT_NO_THROW(test_controller_.runSession(plan_));
EXPECT_NE(nullptr, gcp_credentials_->getCredentials());
}
TEST_F(GCPCredentialsTests, CredentialsFromJsonWithoutProperty) {
- plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::CredentialsLoc,
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_JSON_FILE));
+ EXPECT_TRUE(plan_->setProperty(gcp_credentials_node_,
+ GCPCredentialsControllerService::CredentialsLoc,
+ magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_JSON_FILE)));
ASSERT_NO_THROW(test_controller_.runSession(plan_));
EXPECT_EQ(nullptr, gcp_credentials_->getCredentials());
}
@@ -100,8 +111,10 @@ TEST_F(GCPCredentialsTests,
CredentialsFromJsonWithProperty) {
auto temp_directory = test_controller_.createTempDirectory();
auto path = create_mock_json_file(temp_directory);
ASSERT_TRUE(path.has_value());
- plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::CredentialsLoc,
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_JSON_FILE));
- plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::JsonFilePath, path->string());
+ EXPECT_TRUE(plan_->setProperty(gcp_credentials_node_,
+ GCPCredentialsControllerService::CredentialsLoc,
+ magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_JSON_FILE)));
+ ASSERT_TRUE(plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::JsonFilePath, path->string()));
ASSERT_NO_THROW(test_controller_.runSession(plan_));
EXPECT_NE(nullptr, gcp_credentials_->getCredentials());
}
@@ -110,32 +123,42 @@ TEST_F(GCPCredentialsTests,
CredentialsFromJsonWithInvalidPath) {
auto temp_directory = test_controller_.createTempDirectory();
auto path = create_mock_json_file(temp_directory);
ASSERT_TRUE(path.has_value());
- plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::CredentialsLoc,
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_JSON_FILE));
- plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::JsonFilePath,
"/invalid/path/to/credentials.json");
+ EXPECT_TRUE(plan_->setProperty(gcp_credentials_node_,
+ GCPCredentialsControllerService::CredentialsLoc,
+ magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_JSON_FILE)));
+ ASSERT_TRUE(plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::JsonFilePath,
"/invalid/path/to/credentials.json"));
ASSERT_NO_THROW(test_controller_.runSession(plan_));
EXPECT_EQ(nullptr, gcp_credentials_->getCredentials());
}
TEST_F(GCPCredentialsTests, CredentialsFromComputeEngineVM) {
- plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::CredentialsLoc,
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_COMPUTE_ENGINE_CREDENTIALS));
+ EXPECT_TRUE(plan_->setProperty(gcp_credentials_node_,
+ GCPCredentialsControllerService::CredentialsLoc,
+
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_COMPUTE_ENGINE_CREDENTIALS)));
ASSERT_NO_THROW(test_controller_.runSession(plan_));
EXPECT_NE(nullptr, gcp_credentials_->getCredentials());
}
TEST_F(GCPCredentialsTests, AnonymousCredentials) {
- plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::CredentialsLoc,
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS));
+ EXPECT_TRUE(plan_->setProperty(gcp_credentials_node_,
+ GCPCredentialsControllerService::CredentialsLoc,
+
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS)));
ASSERT_NO_THROW(test_controller_.runSession(plan_));
EXPECT_NE(nullptr, gcp_credentials_->getCredentials());
}
TEST_F(GCPCredentialsTests, CredentialsFromJsonContentsWithoutProperty) {
- plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::CredentialsLoc,
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_JSON_CONTENTS));
+ EXPECT_TRUE(plan_->setProperty(gcp_credentials_node_,
+ GCPCredentialsControllerService::CredentialsLoc,
+
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_JSON_CONTENTS)));
ASSERT_NO_THROW(test_controller_.runSession(plan_));
EXPECT_EQ(nullptr, gcp_credentials_->getCredentials());
}
TEST_F(GCPCredentialsTests, CredentialsFromJsonContentsWithProperty) {
- plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::CredentialsLoc,
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_JSON_CONTENTS));
+ EXPECT_TRUE(plan_->setProperty(gcp_credentials_node_,
+ GCPCredentialsControllerService::CredentialsLoc,
+
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_JSON_CONTENTS)));
plan_->setProperty(gcp_credentials_node_,
GCPCredentialsControllerService::JsonContents, create_mock_service_json());
ASSERT_NO_THROW(test_controller_.runSession(plan_));
EXPECT_NE(nullptr, gcp_credentials_->getCredentials());
diff --git a/extensions/gcp/tests/ListGCSBucketTests.cpp
b/extensions/gcp/tests/ListGCSBucketTests.cpp
index 16b2c8dad..d286fc739 100644
--- a/extensions/gcp/tests/ListGCSBucketTests.cpp
+++ b/extensions/gcp/tests/ListGCSBucketTests.cpp
@@ -14,37 +14,39 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "../processors/ListGCSBucket.h"
#include "../controllerservices/GCPCredentialsControllerService.h"
+#include "../processors/ListGCSBucket.h"
+#include "CProcessorTestUtils.h"
#include "core/Resource.h"
-#include "unit/SingleProcessorTestController.h"
-#include "google/cloud/storage/testing/mock_client.h"
#include "google/cloud/storage/internal/object_metadata_parser.h"
#include "google/cloud/storage/testing/canonical_errors.h"
+#include "google/cloud/storage/testing/mock_client.h"
#include "unit/ProcessorUtils.h"
+#include "unit/SingleProcessorTestController.h"
namespace gcs = ::google::cloud::storage;
-namespace minifi_gcp = org::apache::nifi::minifi::extensions::gcp;
+namespace minifi_gcp = minifi::extensions::gcp;
-using ListGCSBucket =
org::apache::nifi::minifi::extensions::gcp::ListGCSBucket;
+using ListGCSBucket = minifi::extensions::gcp::ListGCSBucket;
using ListObjectsRequest = gcs::internal::ListObjectsRequest;
using ListObjectsResponse = gcs::internal::ListObjectsResponse;
-using GCPCredentialsControllerService =
org::apache::nifi::minifi::extensions::gcp::GCPCredentialsControllerService;
-using ::google::cloud::storage::testing::canonical_errors::TransientError;
+using GCPCredentialsControllerService =
minifi::extensions::gcp::GCPCredentialsControllerService;
using ::google::cloud::storage::testing::canonical_errors::PermanentError;
+using ::google::cloud::storage::testing::canonical_errors::TransientError;
namespace {
class ListGCSBucketMocked : public ListGCSBucket {
- using
org::apache::nifi::minifi::extensions::gcp::ListGCSBucket::ListGCSBucket;
+ using ListGCSBucket::ListGCSBucket;
+
public:
- static constexpr const char* Description = "ListGCSBucketMocked";
+ ListGCSBucketMocked(minifi::core::ProcessorMetadata metadata,
std::shared_ptr<gcs::testing::MockClient> mock_client)
+ : ListGCSBucket(std::move(metadata)),
+ mock_client_(std::move(mock_client)) {}
- gcs::Client getClient() const override {
- return gcs::testing::UndecoratedClientFromMock(mock_client_);
- }
- std::shared_ptr<gcs::testing::MockClient> mock_client_ =
std::make_shared<gcs::testing::MockClient>();
+ protected:
+ gcs::Client getClient() const override { return
gcs::testing::UndecoratedClientFromMock(mock_client_); }
+ std::shared_ptr<gcs::testing::MockClient> mock_client_;
};
-REGISTER_RESOURCE(ListGCSBucketMocked, Processor);
auto CreateObject(int index, int generation = 1) {
std::string id = "object-" + std::to_string(index);
@@ -64,34 +66,34 @@ auto CreateObject(int index, int generation = 1) {
} // namespace
class ListGCSBucketTests : public ::testing::Test {
- public:
+ protected:
void SetUp() override {
- list_gcs_bucket_ = test_controller_.getProcessor();
- gcp_credentials_node_ =
test_controller_.plan->addController("GCPCredentialsControllerService",
"gcp_credentials_controller_service");
- test_controller_.plan->setProperty(gcp_credentials_node_,
-
GCPCredentialsControllerService::CredentialsLoc,
-
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS));
- test_controller_.plan->setProperty(list_gcs_bucket_,
- ListGCSBucket::GCPCredentials,
- "gcp_credentials_controller_service");
+ const auto gcp_credential_controller_service =
+
minifi::test::utils::make_custom_c_controller_service<GCPCredentialsControllerService>(core::ControllerServiceMetadata{utils::Identifier{},
+ "GCPCredentialsControllerService",
+
logging::LoggerFactory<GCPCredentialsControllerService>::getLogger()});
+ gcp_credentials_node_ =
test_controller_.plan->addController("gcp_credentials_controller_service",
gcp_credential_controller_service);
+
EXPECT_TRUE(gcp_credential_controller_service->setProperty(GCPCredentialsControllerService::CredentialsLoc.name,
+
std::string(magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS))));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(ListGCSBucketMocked::GCPCredentials.name,
"gcp_credentials_controller_service"));
}
- org::apache::nifi::minifi::test::SingleProcessorTestController
test_controller_{minifi::test::utils::make_processor<ListGCSBucketMocked>("ListGCSBucketMocked")};
- TypedProcessorWrapper<ListGCSBucketMocked> list_gcs_bucket_;
- std::shared_ptr<minifi::core::controller::ControllerServiceNode>
gcp_credentials_node_;
+
+ public:
+ std::shared_ptr<gcs::testing::MockClient> mock_client_ =
std::make_shared<gcs::testing::MockClient>();
+ minifi::test::SingleProcessorTestController
test_controller_{minifi::test::utils::make_custom_c_processor<ListGCSBucketMocked>(
+ core::ProcessorMetadata{utils::Identifier{}, "ListGCSBucketMocked",
logging::LoggerFactory<ListGCSBucketMocked>::getLogger()}, mock_client_)};
+ std::shared_ptr<minifi::core::controller::ControllerServiceNode>
gcp_credentials_node_;
};
TEST_F(ListGCSBucketTests, MissingBucket) {
- EXPECT_CALL(*list_gcs_bucket_.get().mock_client_,
CreateResumableUpload).Times(0);
+ EXPECT_CALL(*mock_client_, CreateResumableUpload).Times(0);
EXPECT_THROW(test_controller_.trigger(), std::runtime_error);
}
TEST_F(ListGCSBucketTests, ServerGivesPermaError) {
- auto return_permanent_error = [](ListObjectsRequest const&) {
- return google::cloud::StatusOr<ListObjectsResponse>(PermanentError());
- };
- EXPECT_CALL(*list_gcs_bucket_.get().mock_client_, ListObjects)
- .WillOnce(return_permanent_error);
- EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_,
ListGCSBucket::Bucket, "bucket-from-property"));
+ auto return_permanent_error = [](ListObjectsRequest const&) { return
google::cloud::StatusOr<ListObjectsResponse>(PermanentError()); };
+ EXPECT_CALL(*mock_client_, ListObjects).WillOnce(return_permanent_error);
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(ListGCSBucket::Bucket.name,
"bucket-from-property"));
const auto& result = test_controller_.trigger();
EXPECT_EQ(0, result.at(ListGCSBucket::Success).size());
}
@@ -100,15 +102,15 @@ TEST_F(ListGCSBucketTests, ServerGivesTransientErrors) {
auto return_temp_error = [](ListObjectsRequest const&) {
return google::cloud::StatusOr<ListObjectsResponse>(TransientError());
};
- EXPECT_CALL(*list_gcs_bucket_.get().mock_client_,
ListObjects).WillOnce(return_temp_error);
- EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_,
ListGCSBucket::NumberOfRetries, "1"));
- EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_,
ListGCSBucket::Bucket, "bucket-from-property"));
+ EXPECT_CALL(*mock_client_, ListObjects).WillOnce(return_temp_error);
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(ListGCSBucket::NumberOfRetries.name,
"1"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(ListGCSBucket::Bucket.name,
"bucket-from-property"));
const auto& result = test_controller_.trigger();
EXPECT_EQ(0, result.at(ListGCSBucket::Success).size());
}
TEST_F(ListGCSBucketTests, WithoutVersions) {
- EXPECT_CALL(*list_gcs_bucket_.get().mock_client_, ListObjects)
+ EXPECT_CALL(*mock_client_, ListObjects)
.WillOnce([](ListObjectsRequest const& req)
-> google::cloud::StatusOr<ListObjectsResponse> {
EXPECT_EQ("bucket-from-property", req.bucket_name());
@@ -121,14 +123,13 @@ TEST_F(ListGCSBucketTests, WithoutVersions) {
response.items.emplace_back(CreateObject(1, 3));
return response;
});
- EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_,
ListGCSBucket::Bucket, "bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(ListGCSBucket::Bucket.name,
"bucket-from-property"));
const auto& result = test_controller_.trigger();
EXPECT_EQ(3, result.at(ListGCSBucket::Success).size());
}
-
TEST_F(ListGCSBucketTests, WithVersions) {
- EXPECT_CALL(*list_gcs_bucket_.get().mock_client_, ListObjects)
+ EXPECT_CALL(*mock_client_, ListObjects)
.WillOnce([](ListObjectsRequest const& req)
-> google::cloud::StatusOr<ListObjectsResponse> {
EXPECT_EQ("bucket-from-property", req.bucket_name());
@@ -141,9 +142,8 @@ TEST_F(ListGCSBucketTests, WithVersions) {
response.items.emplace_back(CreateObject(3));
return response;
});
- EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_,
ListGCSBucket::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(list_gcs_bucket_,
ListGCSBucket::ListAllVersions, "true"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(ListGCSBucket::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(ListGCSBucket::ListAllVersions.name,
"true"));
const auto& result = test_controller_.trigger();
EXPECT_EQ(3, result.at(ListGCSBucket::Success).size());
}
-
diff --git a/extensions/gcp/tests/PutGCSObjectTests.cpp
b/extensions/gcp/tests/PutGCSObjectTests.cpp
index a8355b50f..01b58104e 100644
--- a/extensions/gcp/tests/PutGCSObjectTests.cpp
+++ b/extensions/gcp/tests/PutGCSObjectTests.cpp
@@ -14,56 +14,59 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-#include "../processors/PutGCSObject.h"
#include "../controllerservices/GCPCredentialsControllerService.h"
+#include "../processors/PutGCSObject.h"
+#include "CProcessorTestUtils.h"
#include "GCPAttributes.h"
#include "core/Resource.h"
-#include "unit/SingleProcessorTestController.h"
-#include "google/cloud/storage/testing/mock_client.h"
#include "google/cloud/storage/internal/object_metadata_parser.h"
-#include "google/cloud/storage/retry_policy.h"
#include "google/cloud/storage/testing/canonical_errors.h"
+#include "google/cloud/storage/testing/mock_client.h"
#include "unit/ProcessorUtils.h"
+#include "unit/SingleProcessorTestController.h"
namespace gcs = ::google::cloud::storage;
-namespace minifi_gcp = org::apache::nifi::minifi::extensions::gcp;
+namespace minifi_gcp = minifi::extensions::gcp;
-using PutGCSObject = org::apache::nifi::minifi::extensions::gcp::PutGCSObject;
-using GCPCredentialsControllerService =
org::apache::nifi::minifi::extensions::gcp::GCPCredentialsControllerService;
+using PutGCSObject = minifi::extensions::gcp::PutGCSObject;
+using GCPCredentialsControllerService =
minifi::extensions::gcp::GCPCredentialsControllerService;
using ResumableUploadRequest = gcs::internal::ResumableUploadRequest;
using QueryResumableUploadResponse =
gcs::internal::QueryResumableUploadResponse;
-using ::google::cloud::storage::testing::canonical_errors::TransientError;
using ::google::cloud::storage::testing::canonical_errors::PermanentError;
+using ::google::cloud::storage::testing::canonical_errors::TransientError;
namespace {
class PutGCSObjectMocked : public PutGCSObject {
- using org::apache::nifi::minifi::extensions::gcp::PutGCSObject::PutGCSObject;
+ using PutGCSObject::PutGCSObject;
+
public:
- static constexpr const char* Description = "PutGCSObjectMocked";
+ PutGCSObjectMocked(minifi::core::ProcessorMetadata metadata,
std::shared_ptr<gcs::testing::MockClient> mock_client)
+ : PutGCSObject(std::move(metadata)),
+ mock_client_(std::move(mock_client)) {}
- gcs::Client getClient() const override {
- return gcs::testing::UndecoratedClientFromMock(mock_client_);
- }
- std::shared_ptr<gcs::testing::MockClient> mock_client_ =
std::make_shared<gcs::testing::MockClient>();
+ protected:
+ gcs::Client getClient() const override { return
gcs::testing::UndecoratedClientFromMock(mock_client_); }
+ std::shared_ptr<gcs::testing::MockClient> mock_client_;
};
-REGISTER_RESOURCE(PutGCSObjectMocked, Processor);
} // namespace
class PutGCSObjectTests : public ::testing::Test {
- public:
+ protected:
void SetUp() override {
- put_gcs_object_ = test_controller_.getProcessor();
- gcp_credentials_node_ =
test_controller_.plan->addController("GCPCredentialsControllerService",
"gcp_credentials_controller_service");
- test_controller_.plan->setProperty(gcp_credentials_node_,
-
GCPCredentialsControllerService::CredentialsLoc,
-
magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS));
- test_controller_.plan->setProperty(put_gcs_object_,
- PutGCSObject::GCPCredentials,
- "gcp_credentials_controller_service");
+ const auto gcp_credential_controller_service =
+
minifi::test::utils::make_custom_c_controller_service<GCPCredentialsControllerService>(core::ControllerServiceMetadata{utils::Identifier{},
+ "GCPCredentialsControllerService",
+
logging::LoggerFactory<GCPCredentialsControllerService>::getLogger()});
+ gcp_credentials_node_ =
test_controller_.plan->addController("gcp_credentials_controller_service",
gcp_credential_controller_service);
+
EXPECT_TRUE(gcp_credential_controller_service->setProperty(GCPCredentialsControllerService::CredentialsLoc.name,
+
std::string(magic_enum::enum_name(minifi_gcp::CredentialsLocation::USE_ANONYMOUS_CREDENTIALS))));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObjectMocked::GCPCredentials.name,
"gcp_credentials_controller_service"));
}
- TypedProcessorWrapper<PutGCSObjectMocked> put_gcs_object_;
- org::apache::nifi::minifi::test::SingleProcessorTestController
test_controller_{minifi::test::utils::make_processor<PutGCSObjectMocked>("PutGCSObjectMocked")};
- std::shared_ptr<minifi::core::controller::ControllerServiceNode>
gcp_credentials_node_;
+
+ std::shared_ptr<gcs::testing::MockClient> mock_client_ =
std::make_shared<gcs::testing::MockClient>();
+ minifi::test::SingleProcessorTestController
test_controller_{minifi::test::utils::make_custom_c_processor<PutGCSObjectMocked>(
+ core::ProcessorMetadata{utils::Identifier{}, "PutGCSObjectMocked",
logging::LoggerFactory<PutGCSObjectMocked>::getLogger()}, mock_client_)};
+ std::shared_ptr<minifi::core::controller::ControllerServiceNode>
gcp_credentials_node_;
static auto return_upload_done(const ResumableUploadRequest& request) {
using ObjectMetadataParser = gcs::internal::ObjectMetadataParser;
@@ -75,13 +78,14 @@ class PutGCSObjectTests : public ::testing::Test {
metadata_json["customerEncryption"]["encryptionAlgorithm"] = "AES256";
metadata_json["customerEncryption"]["keySha256"] =
"zkeXIcAB56dkHp0z1023TQZ+mzm+fZ5JRVgmAQ3bEVE=";
}
- return
testing::Return(google::cloud::make_status_or(QueryResumableUploadResponse{absl::nullopt,
*ObjectMetadataParser::FromJson(metadata_json)}));
+ return
testing::Return(google::cloud::make_status_or(QueryResumableUploadResponse{absl::nullopt,
+ *ObjectMetadataParser::FromJson(metadata_json)}));
}
};
TEST_F(PutGCSObjectTests, MissingBucket) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
CreateResumableUpload).Times(0);
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, ""));
+ EXPECT_CALL(*mock_client_, CreateResumableUpload).Times(0);
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
""));
const auto& result = test_controller_.trigger("hello world");
EXPECT_EQ(0, result.at(PutGCSObject::Success).size());
ASSERT_EQ(1, result.at(PutGCSObject::Failure).size());
@@ -91,13 +95,13 @@ TEST_F(PutGCSObjectTests, MissingBucket) {
}
TEST_F(PutGCSObjectTests, BucketFromAttribute) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_, CreateResumableUpload)
+ EXPECT_CALL(*mock_client_, CreateResumableUpload)
.WillOnce([this](const ResumableUploadRequest& request) {
EXPECT_EQ("bucket-from-attribute", request.bucket_name());
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
+ EXPECT_CALL(*mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
return
gcs::internal::CreateResumableUploadResponse{"test-only-upload-id"};
});
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "${gcs.bucket}"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"${gcs.bucket}"));
const auto& result = test_controller_.trigger("hello world",
{{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"}});
ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
@@ -105,10 +109,10 @@ TEST_F(PutGCSObjectTests, BucketFromAttribute) {
}
TEST_F(PutGCSObjectTests, ServerGivesTransientErrors) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
CreateResumableUpload).WillOnce(testing::Return(TransientError()));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::NumberOfRetries, "2"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Key, "object-name-from-property"));
+ EXPECT_CALL(*mock_client_,
CreateResumableUpload).WillOnce(testing::Return(TransientError()));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::NumberOfRetries.name,
"2"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Key.name,
"object-name-from-property"));
const auto& result = test_controller_.trigger("hello world");
EXPECT_EQ(0, result.at(PutGCSObject::Success).size());
ASSERT_EQ(1, result.at(PutGCSObject::Failure).size());
@@ -118,9 +122,9 @@ TEST_F(PutGCSObjectTests, ServerGivesTransientErrors) {
}
TEST_F(PutGCSObjectTests, ServerGivesPermaError) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
CreateResumableUpload).WillOnce(testing::Return(PermanentError()));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Key, "object-name-from-property"));
+ EXPECT_CALL(*mock_client_,
CreateResumableUpload).WillOnce(testing::Return(PermanentError()));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Key.name,
"object-name-from-property"));
const auto& result = test_controller_.trigger("hello world");
EXPECT_EQ(0, result.at(PutGCSObject::Success).size());
ASSERT_EQ(1, result.at(PutGCSObject::Failure).size());
@@ -130,51 +134,51 @@ TEST_F(PutGCSObjectTests, ServerGivesPermaError) {
}
TEST_F(PutGCSObjectTests, NonRequiredPropertiesAreMissing) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_, CreateResumableUpload)
+ EXPECT_CALL(*mock_client_, CreateResumableUpload)
.WillOnce([this](const ResumableUploadRequest& request) {
EXPECT_FALSE(request.HasOption<gcs::MD5HashValue>());
EXPECT_FALSE(request.HasOption<gcs::Crc32cChecksumValue>());
EXPECT_FALSE(request.HasOption<gcs::PredefinedAcl>());
EXPECT_FALSE(request.HasOption<gcs::IfGenerationMatch>());
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
+ EXPECT_CALL(*mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
return
gcs::internal::CreateResumableUploadResponse{"test-only-upload-id"};
});
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Key, "object-name-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Key.name,
"object-name-from-property"));
const auto& result = test_controller_.trigger("hello world");
EXPECT_EQ(1, result.at(PutGCSObject::Success).size());
EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
}
TEST_F(PutGCSObjectTests, Crc32cMD5LocationTest) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_, CreateResumableUpload)
+ EXPECT_CALL(*mock_client_, CreateResumableUpload)
.WillOnce([this](const ResumableUploadRequest& request) {
EXPECT_TRUE(request.HasOption<gcs::Crc32cChecksumValue>());
EXPECT_EQ("yZRlqg==",
request.GetOption<gcs::Crc32cChecksumValue>().value());
EXPECT_TRUE(request.HasOption<gcs::MD5HashValue>());
EXPECT_EQ("XrY7u+Ae7tCTyyK7j1rNww==",
request.GetOption<gcs::MD5HashValue>().value());
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
+ EXPECT_CALL(*mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
return
gcs::internal::CreateResumableUploadResponse{"test-only-upload-id"};
});
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::MD5Hash, "${md5}"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Crc32cChecksum, "${crc32c}"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Key, "object-name-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::MD5Hash.name,
"${md5}"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Crc32cChecksum.name,
"${crc32c}"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Key.name,
"object-name-from-property"));
const auto& result = test_controller_.trigger("hello world", {{"crc32c",
"yZRlqg=="}, {"md5", "XrY7u+Ae7tCTyyK7j1rNww=="}});
EXPECT_EQ(1, result.at(PutGCSObject::Success).size());
EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
}
TEST_F(PutGCSObjectTests, DontOverwriteTest) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_, CreateResumableUpload)
+ EXPECT_CALL(*mock_client_, CreateResumableUpload)
.WillOnce([this](const ResumableUploadRequest& request) {
EXPECT_TRUE(request.HasOption<gcs::IfGenerationMatch>());
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
+ EXPECT_CALL(*mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
return
gcs::internal::CreateResumableUploadResponse{"test-only-upload-id"};
});
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::OverwriteObject, "false"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Key, "object-name-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::OverwriteObject.name,
"false"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Key.name,
"object-name-from-property"));
const auto& result = test_controller_.trigger("hello world", {{"crc32c",
"yZRlqg=="}, {"md5", "XrY7u+Ae7tCTyyK7j1rNww=="}});
ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
@@ -182,15 +186,15 @@ TEST_F(PutGCSObjectTests, DontOverwriteTest) {
}
TEST_F(PutGCSObjectTests, ValidServerSideEncryptionTest) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_, CreateResumableUpload)
+ EXPECT_CALL(*mock_client_, CreateResumableUpload)
.WillOnce([this](const ResumableUploadRequest& request) {
EXPECT_TRUE(request.HasOption<gcs::EncryptionKey>());
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
+ EXPECT_CALL(*mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
return
gcs::internal::CreateResumableUploadResponse{"test-only-upload-id"};
});
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::EncryptionKey, "ZW5jcnlwdGlvbl9rZXk="));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Key, "object-name-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::EncryptionKey.name,
"ZW5jcnlwdGlvbl9rZXk="));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Key.name,
"object-name-from-property"));
const auto& result = test_controller_.trigger("hello world");
ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
@@ -200,22 +204,22 @@ TEST_F(PutGCSObjectTests, ValidServerSideEncryptionTest) {
}
TEST_F(PutGCSObjectTests, InvalidServerSideEncryptionTest) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
CreateResumableUpload).Times(0);
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::EncryptionKey, "not_base64_key"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Key, "object-name-from-property"));
+ EXPECT_CALL(*mock_client_, CreateResumableUpload).Times(0);
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::EncryptionKey.name,
"not_base64_key"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Key.name,
"object-name-from-property"));
EXPECT_THROW(test_controller_.trigger("hello world"), minifi::Exception);
}
TEST_F(PutGCSObjectTests, NoContentType) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_, CreateResumableUpload)
+ EXPECT_CALL(*mock_client_, CreateResumableUpload)
.WillOnce([this](const ResumableUploadRequest& request) {
EXPECT_FALSE(request.HasOption<gcs::ContentType>());
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
+ EXPECT_CALL(*mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
return
gcs::internal::CreateResumableUploadResponse{"test-only-upload-id"};
});
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Key, "object-name-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Key.name,
"object-name-from-property"));
const auto& result = test_controller_.trigger("hello world");
ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
@@ -223,15 +227,15 @@ TEST_F(PutGCSObjectTests, NoContentType) {
}
TEST_F(PutGCSObjectTests, ContentTypeFromAttribute) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_, CreateResumableUpload)
+ EXPECT_CALL(*mock_client_, CreateResumableUpload)
.WillOnce([this](const ResumableUploadRequest& request) {
EXPECT_TRUE(request.HasOption<gcs::ContentType>());
EXPECT_EQ("text/attribute",
request.GetOption<gcs::ContentType>().value());
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
+ EXPECT_CALL(*mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
return
gcs::internal::CreateResumableUploadResponse{"test-only-upload-id"};
});
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Key, "object-name-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Key.name,
"object-name-from-property"));
const auto& result = test_controller_.trigger("hello world", {{"mime.type",
"text/attribute"}});
ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
@@ -239,16 +243,17 @@ TEST_F(PutGCSObjectTests, ContentTypeFromAttribute) {
}
TEST_F(PutGCSObjectTests, ObjectACLTest) {
- EXPECT_CALL(*put_gcs_object_.get().mock_client_, CreateResumableUpload)
+ EXPECT_CALL(*mock_client_, CreateResumableUpload)
.WillOnce([this](const ResumableUploadRequest& request) {
EXPECT_TRUE(request.HasOption<gcs::PredefinedAcl>());
EXPECT_EQ(gcs::PredefinedAcl::AuthenticatedRead().value(),
request.GetOption<gcs::PredefinedAcl>().value());
- EXPECT_CALL(*put_gcs_object_.get().mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
+ EXPECT_CALL(*mock_client_,
UploadChunk).WillOnce(return_upload_done(request));
return
gcs::internal::CreateResumableUploadResponse{"test-only-upload-id"};
});
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Bucket, "bucket-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::Key, "object-name-from-property"));
- EXPECT_TRUE(test_controller_.plan->setProperty(put_gcs_object_,
PutGCSObject::ObjectACL,
magic_enum::enum_name(minifi_gcp::put_gcs_object::PredefinedAcl::AUTHENTICATED_READ)));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Bucket.name,
"bucket-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::Key.name,
"object-name-from-property"));
+
EXPECT_TRUE(test_controller_.getProcessor()->setProperty(PutGCSObject::ObjectACL.name,
+
std::string(magic_enum::enum_name(minifi_gcp::put_gcs_object::PredefinedAcl::AUTHENTICATED_READ))));
const auto& result = test_controller_.trigger("hello world");
ASSERT_EQ(1, result.at(PutGCSObject::Success).size());
EXPECT_EQ(0, result.at(PutGCSObject::Failure).size());
diff --git a/extensions/kafka/KafkaProcessorBase.cpp
b/extensions/kafka/KafkaProcessorBase.cpp
index 02233edff..e4e224015 100644
--- a/extensions/kafka/KafkaProcessorBase.cpp
+++ b/extensions/kafka/KafkaProcessorBase.cpp
@@ -25,7 +25,7 @@
KafkaProcessorBase::KafkaProcessorBase(core::ProcessorMetadata metadata) : Proce
}
std::optional<api::utils::net::SslData>
KafkaProcessorBase::getSslData(api::core::ProcessContext& context) const {
- return context.getSslData(SSLContextService) | utils::toOptional();
+ return context.getSslData(SSLContextService).value_or(std::nullopt);
}
void
KafkaProcessorBase::setKafkaAuthenticationParameters(api::core::ProcessContext&
context, gsl::not_null<rd_kafka_conf_t*> config) { //
NOLINT(performance-unnecessary-value-param)
@@ -33,7 +33,7 @@ void
KafkaProcessorBase::setKafkaAuthenticationParameters(api::core::ProcessCont
utils::setKafkaConfigurationField(*config, "security.protocol",
std::string{magic_enum::enum_name(security_protocol_)});
logger_->log_debug("Kafka security.protocol [{}]",
magic_enum::enum_name(security_protocol_));
if (security_protocol_ == kafka::SecurityProtocolOption::ssl ||
security_protocol_ == kafka::SecurityProtocolOption::sasl_ssl) {
- if (auto ssl_data = context.getSslData(SSLContextService) |
utils::toOptional()) {
+ if (auto ssl_data = getSslData(context)) {
if (ssl_data->ca_loc.empty() && ssl_data->cert_loc.empty() &&
ssl_data->key_loc.empty() && ssl_data->key_pw.empty()) {
logger_->log_warn("Security protocol is set to {}, but no valid
security parameters are set in the properties or in the SSL Context Service.",
magic_enum::enum_name(security_protocol_));
diff --git a/extensions/kafka/rdkafka_utils.h b/extensions/kafka/rdkafka_utils.h
index 96d0d79fe..91f59deb8 100644
--- a/extensions/kafka/rdkafka_utils.h
+++ b/extensions/kafka/rdkafka_utils.h
@@ -25,6 +25,7 @@
#include "rdkafka.h"
#include "api/utils/Ssl.h"
#include "minifi-cpp/core/logging/Logger.h"
+#include "magic_enum/magic_enum.hpp"
namespace org::apache::nifi::minifi::utils {
diff --git a/libminifi/src/minifi-c.cpp b/libminifi/src/minifi-c.cpp
index 4cb3da553..220ec9513 100644
--- a/libminifi/src/minifi-c.cpp
+++ b/libminifi/src/minifi-c.cpp
@@ -25,6 +25,7 @@
#include "core/extension/ExtensionManager.h"
#include "minifi-cpp/Exception.h"
#include "minifi-cpp/controllers/SSLContextServiceInterface.h"
+#include "minifi-cpp/controllers/ProxyConfigurationServiceInterface.h"
#include "minifi-cpp/core/Annotation.h"
#include "minifi-cpp/core/ClassLoader.h"
#include "minifi-cpp/core/ProcessContext.h"
@@ -180,6 +181,16 @@ class CControllerServiceFactory : public
minifi::core::controller::ControllerSer
minifi::utils::CControllerServiceClassDescription class_description_;
};
+MinifiProxyType minifiProxyType(const minifi::controllers::ProxyType&
proxy_type) {
+ switch (proxy_type) {
+ case minifi::controllers::ProxyType::DIRECT:
+ return MinifiProxyType::MINIFI_PROXY_TYPE_DIRECT;
+ case minifi::controllers::ProxyType::HTTP:
+ return MinifiProxyType::MINIFI_PROXY_TYPE_HTTP;
+ }
+ std::unreachable();
+}
+
} // namespace
namespace org::apache::nifi::minifi::utils {
@@ -578,9 +589,9 @@ MinifiStatus
MinifiControllerServiceContextGetProperty(MinifiControllerServiceCo
}
}
-MinifiStatus MinifiProcessContextGetControllerService(
+MinifiStatus MinifiProcessContextGetControllerServiceFromProperty(
MinifiProcessContext* process_context,
- const MinifiStringView controller_service_name,
+ const MinifiStringView property_name,
const MinifiStringView controller_service_type,
MinifiControllerService** controller_service_out) {
if (!controller_service_out) {
@@ -589,8 +600,10 @@ MinifiStatus MinifiProcessContextGetControllerService(
gsl_Assert(process_context != MINIFI_NULL);
const auto context =
reinterpret_cast<minifi::core::ProcessContext*>(process_context);
- const auto name_str = std::string{toStringView(controller_service_name)};
- const auto service_shared_ptr = context->getControllerService(name_str,
context->getProcessorInfo().getUUID());
+ const auto property_name_str = std::string{toStringView(property_name)};
+ const auto name_str = context->getProperty(property_name_str, nullptr);
+ if (!name_str) { return MINIFI_STATUS_PROPERTY_NOT_SET; }
+ const auto service_shared_ptr = context->getControllerService(*name_str,
context->getProcessorInfo().getUUID());
if (!service_shared_ptr) {
return MINIFI_STATUS_VALIDATION_FAILED;
}
@@ -645,4 +658,32 @@ MinifiStatus
MinifiProcessContextGetSslDataFromProperty(MinifiProcessContext* pr
}
+MinifiStatus
MinifiProcessContextGetProxyDataFromProperty(MinifiProcessContext*
process_context, MinifiStringView property_name,
+ void (*cb)(void* user_ctx, const MinifiProxyData* proxy_data), void*
user_ctx) {
+ gsl_Assert(process_context != MINIFI_NULL);
+ const auto context =
reinterpret_cast<minifi::core::ProcessContext*>(process_context);
+ const auto property_name_str = std::string{toStringView(property_name)};
+ const auto name_str = context->getProperty(property_name_str, nullptr);
+ if (!name_str) { return MINIFI_STATUS_PROPERTY_NOT_SET; }
+ const auto service_shared_ptr = context->getControllerService(*name_str,
context->getProcessorInfo().getUUID());
+ if (!service_shared_ptr) { return MINIFI_STATUS_VALIDATION_FAILED; }
+ if (const auto proxy_service =
dynamic_cast<minifi::controllers::ProxyConfigurationServiceInterface*>(service_shared_ptr.get()))
{
+ const std::string hostname = proxy_service->getHost();
+ const auto basic_auth_data = proxy_service->getProxyCredentials();
+ MinifiStringView username_holder = basic_auth_data ?
minifiStringView(basic_auth_data->username) : MinifiStringView{};
+ MinifiStringView password_holder = basic_auth_data ?
minifiStringView(basic_auth_data->password) : MinifiStringView{};
+
+ MinifiProxyData proxy_data{
+ .hostname = minifiStringView(hostname),
+ .port = proxy_service->getPort(),
+ .username = basic_auth_data ? &username_holder : nullptr,
+ .password = basic_auth_data ? &password_holder : nullptr,
+ .proxy_type = minifiProxyType(proxy_service->getProxyType()),
+ };
+ cb(user_ctx, &proxy_data);
+ return MINIFI_STATUS_SUCCESS;
+ }
+ return MINIFI_STATUS_VALIDATION_FAILED;
+}
+
} // extern "C"
diff --git a/minifi-api/include/minifi-c/minifi-c.h
b/minifi-api/include/minifi-c/minifi-c.h
index b885cd614..8294dd91f 100644
--- a/minifi-api/include/minifi-c/minifi-c.h
+++ b/minifi-api/include/minifi-c/minifi-c.h
@@ -45,6 +45,11 @@ extern "C" {
/// use MINIFI_SSL_CONTEXT_SERVICE_PROPERTY_TYPE in the type field of the
property definition (MinifiPropertyDefinition::type)
#define MINIFI_SSL_CONTEXT_SERVICE_PROPERTY_TYPE
"org.apache.nifi.minifi.controllers.SSLContextServiceInterface"
+/// To declare a processor property that expects an ProxyConfigurationService,
+/// use MINIFI_PROXY_CONFIGURATION_SERVICE_PROPERTY_TYPE in the type field of
the property definition (MinifiPropertyDefinition::type)
+#define MINIFI_PROXY_CONFIGURATION_SERVICE_PROPERTY_TYPE
"org.apache.nifi.minifi.controllers.ProxyConfigurationServiceInterface"
+
+
enum : uint32_t {
MINIFI_API_VERSION = 3
};
@@ -233,8 +238,8 @@ MinifiStatus
MinifiProcessContextGetProperty(MinifiProcessContext* context, Mini
void(*cb)(void* user_ctx,
MinifiStringView property_value), void* user_ctx);
MinifiBool MinifiProcessContextHasNonEmptyProperty(MinifiProcessContext*
context, MinifiStringView property_name);
-MinifiStatus MinifiProcessContextGetControllerService(
- MinifiProcessContext* process_context, MinifiStringView
controller_service_name, MinifiStringView controller_service_type,
MinifiControllerService** controller_service_out);
+MinifiStatus MinifiProcessContextGetControllerServiceFromProperty(
+ MinifiProcessContext* process_context, MinifiStringView property_name,
MinifiStringView controller_service_type, MinifiControllerService**
controller_service_out);
void MinifiProcessContextGetDynamicProperties(MinifiProcessContext* context,
MinifiFlowFile* minifi_flow_file,
void (*cb)(void* user_ctx, MinifiStringView dynamic_property_name,
MinifiStringView dynamic_property_value), void* user_ctx);
@@ -283,6 +288,22 @@ typedef struct MinifiSslData {
MinifiStatus MinifiProcessContextGetSslDataFromProperty(MinifiProcessContext*
process_context, MinifiStringView property_name,
void (*cb)(void* user_ctx, const MinifiSslData* ssl_data), void* user_ctx);
+typedef enum MinifiProxyType : uint8_t {
+ MINIFI_PROXY_TYPE_DIRECT,
+ MINIFI_PROXY_TYPE_HTTP
+} MinifiProxyType;
+
+typedef struct MinifiProxyData {
+ MinifiStringView hostname;
+ uint16_t port;
+ MinifiStringView* username;
+ MinifiStringView* password;
+ MinifiProxyType proxy_type;
+} MinifiProxyData;
+
+MinifiStatus
MinifiProcessContextGetProxyDataFromProperty(MinifiProcessContext*
process_context, MinifiStringView property_name,
+ void (*cb)(void* user_ctx, const MinifiProxyData* proxy_data), void*
user_ctx);
+
#ifdef __cplusplus
} // extern "C"
#endif // __cplusplus
diff --git a/minifi-api/minifi-c-api.def b/minifi-api/minifi-c-api.def
index 0baa3063c..e74b9b9f1 100644
--- a/minifi-api/minifi-c-api.def
+++ b/minifi-api/minifi-c-api.def
@@ -4,7 +4,7 @@ EXPORTS
MinifiRegisterProcessor
MinifiRegisterControllerService
MinifiPublishedMetricsCreate
- MinifiProcessContextGetControllerService
+ MinifiProcessContextGetControllerServiceFromProperty
MinifiProcessContextGetProperty
MinifiProcessContextHasNonEmptyProperty
MinifiControllerServiceContextGetProperty
@@ -30,3 +30,4 @@ EXPORTS
MinifiProcessSessionGetFlowFileId
MinifiProcessContextGetDynamicProperties
MinifiProcessContextGetSslDataFromProperty
+ MinifiProcessContextGetProxyDataFromProperty
