You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by lo...@apache.org on 2023/08/16 17:54:30 UTC
[nifi-minifi-cpp] branch main updated (2cef282cf -> fbd69e7e2)
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a change to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
from 2cef282cf MINIFICPP-2183 fix MacOS CI failure
new deab43e44 MINIFICPP-2129 Refactor threadpool
new 14e2586aa MINIFICPP-2170 Fix system certificate store usage in SSLContextService on Linux
new 4307333d1 MINIFICPP-2116 - Support GCC13
new 6372d67ab MINIFICPP-2175 Increase CMAKE requirements to 3.24
new fbd69e7e2 MINIFICPP-2164 Use a list of types for allowed types in PropertyDefinition
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/ci.yml | 2 +-
.github/workflows/gcc13-compat.yml | 45 ++++++
CMakeLists.txt | 2 +-
README.md | 2 +-
centos.sh | 8 +-
cmake/Abseil.cmake | 4 +-
cmake/BundledAzureSdkCpp.cmake | 16 ++-
cmake/GoogleCloudCpp.cmake | 2 +
cmake/KubernetesClientC.cmake | 3 +-
cmake/curl/dummy/FindCURL.cmake | 1 +
docker/Dockerfile | 2 +-
docker/centos/Dockerfile | 2 +-
docker/requirements.txt | 2 +-
docker/rockylinux/Dockerfile | 4 +-
.../cluster/DockerTestDirectoryBindings.py | 2 +
.../features/MiNiFi_integration_test_driver.py | 24 +++-
docker/test/integration/features/https.feature | 155 +++++++++++++++++++--
docker/test/integration/features/kafka.feature | 8 +-
docker/test/integration/features/steps/steps.py | 43 +++---
.../minifi/controllers/SSLContextService.py | 5 +-
.../test/integration/ssl_utils/SSL_cert_utils.py | 2 +-
extensions/aws/processors/S3Processor.cpp | 6 +-
extensions/azure/storage/BlobStorageClient.h | 2 +-
extensions/azure/storage/DataLakeStorageClient.h | 2 +-
extensions/elasticsearch/PostElasticsearch.h | 9 +-
extensions/gcp/processors/GCSProcessor.h | 4 +-
extensions/http-curl/client/HTTPClient.cpp | 24 ++--
extensions/http-curl/client/HTTPClient.h | 2 +-
extensions/http-curl/processors/InvokeHTTP.cpp | 2 +-
extensions/http-curl/processors/InvokeHTTP.h | 4 +-
extensions/http-curl/tests/HTTPHandlers.h | 2 +-
.../http-curl/tests/VerifyInvokeHTTPGetTest.cpp | 2 -
extensions/librdkafka/KafkaProcessorBase.h | 52 +++----
extensions/rocksdb-repos/FlowFileLoader.cpp | 6 +-
extensions/rocksdb-repos/FlowFileLoader.h | 2 +-
extensions/sftp/client/SFTPClient.cpp | 5 +-
extensions/sftp/processors/ListSFTP.cpp | 4 +-
extensions/splunk/SplunkHECProcessor.h | 40 +++---
extensions/standard-processors/processors/GetTCP.h | 4 +-
.../standard-processors/processors/ListenSyslog.h | 4 +-
.../standard-processors/processors/ListenTCP.h | 4 +-
.../standard-processors/processors/PutFile.h | 2 +-
extensions/standard-processors/processors/PutTCP.h | 98 ++++++-------
.../standard-processors/processors/TailFile.h | 4 +-
.../tests/unit/YamlConfigurationTests.cpp | 8 +-
libminifi/include/CronDrivenSchedulingAgent.h | 2 +-
libminifi/include/EventDrivenSchedulingAgent.h | 2 +-
libminifi/include/FlowController.h | 2 +-
libminifi/include/SchedulingAgent.h | 4 +-
libminifi/include/ThreadedSchedulingAgent.h | 2 +-
libminifi/include/TimerDrivenSchedulingAgent.h | 2 +-
libminifi/include/c2/C2Agent.h | 2 +-
libminifi/include/core/OutputAttributeDefinition.h | 4 +-
libminifi/include/core/Processor.h | 1 +
libminifi/include/core/PropertyDefinition.h | 18 +--
libminifi/include/core/PropertyDefinitionBuilder.h | 41 +++---
libminifi/include/core/Resource.h | 4 +-
libminifi/include/utils/HTTPUtils.h | 4 +-
libminifi/include/utils/Monitors.h | 82 +++--------
libminifi/include/utils/ThreadPool.h | 98 ++++---------
libminifi/src/CronDrivenSchedulingAgent.cpp | 3 +-
libminifi/src/EventDrivenSchedulingAgent.cpp | 2 +-
libminifi/src/FlowController.cpp | 2 +-
libminifi/src/ThreadedSchedulingAgent.cpp | 17 +--
libminifi/src/TimerDrivenSchedulingAgent.cpp | 2 +-
libminifi/src/c2/C2Agent.cpp | 5 +-
libminifi/src/controllers/SSLContextService.cpp | 13 +-
libminifi/src/core/Property.cpp | 6 +-
libminifi/src/utils/HTTPUtils.cpp | 9 +-
libminifi/src/utils/ThreadPool.cpp | 80 ++++-------
libminifi/src/utils/TimeUtil.cpp | 2 +-
libminifi/test/TestBase.cpp | 3 +-
libminifi/test/unit/BackTraceTests.cpp | 97 ++++++-------
libminifi/test/unit/ComponentManifestTests.cpp | 4 +-
libminifi/test/unit/SchedulingAgentTests.cpp | 57 ++++----
libminifi/test/unit/SocketTests.cpp | 46 +++---
libminifi/test/unit/ThreadPoolTests.cpp | 64 +++------
libminifi/test/unit/TypeListTests.cpp | 61 ++++++++
linux.sh | 30 ++++
nanofi/include/cxx/Instance.h | 5 +-
.../azure-sdk-cpp-remove-samples.patch | 23 ---
thirdparty/azure-sdk-cpp/gcc13.patch | 44 ++++++
thirdparty/azure-sdk-cpp/remove-amqp.patch | 15 ++
83 files changed, 826 insertions(+), 658 deletions(-)
create mode 100644 .github/workflows/gcc13-compat.yml
create mode 100644 libminifi/test/unit/TypeListTests.cpp
delete mode 100644 thirdparty/azure-sdk-cpp/azure-sdk-cpp-remove-samples.patch
create mode 100644 thirdparty/azure-sdk-cpp/gcc13.patch
create mode 100644 thirdparty/azure-sdk-cpp/remove-amqp.patch
[nifi-minifi-cpp] 02/05: MINIFICPP-2170 Fix system certificate store usage in SSLContextService on Linux
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 14e2586aac2977f76b49c4c458908008edc9f211
Author: Ferenc Gerlits <fg...@gmail.com>
AuthorDate: Tue Jul 25 11:53:04 2023 +0200
MINIFICPP-2170 Fix system certificate store usage in SSLContextService on Linux
Use utils::getDefaultCAFile() in the SSLContextService to find the system CA
certificate store file, as the compiled-in default value is usually wrong.
If the system CA certificate store file is still not found, you can set the
SSL_CERT_DIR or SSL_CERT_FILE environment variables to tell OpenSSL where
the certificates are.
Also: fix the incorrect handling of InvokeHTTP::DisablePeerVerification.
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #1620
---
docker/requirements.txt | 2 +-
.../cluster/DockerTestDirectoryBindings.py | 2 +
.../features/MiNiFi_integration_test_driver.py | 24 +++-
docker/test/integration/features/https.feature | 155 +++++++++++++++++++--
docker/test/integration/features/kafka.feature | 8 +-
docker/test/integration/features/steps/steps.py | 43 +++---
.../minifi/controllers/SSLContextService.py | 5 +-
.../test/integration/ssl_utils/SSL_cert_utils.py | 2 +-
extensions/aws/processors/S3Processor.cpp | 6 +-
extensions/http-curl/client/HTTPClient.cpp | 24 ++--
extensions/http-curl/client/HTTPClient.h | 2 +-
extensions/http-curl/processors/InvokeHTTP.cpp | 2 +-
.../http-curl/tests/VerifyInvokeHTTPGetTest.cpp | 2 -
libminifi/include/utils/HTTPUtils.h | 4 +-
libminifi/src/controllers/SSLContextService.cpp | 13 +-
libminifi/src/utils/HTTPUtils.cpp | 9 +-
16 files changed, 228 insertions(+), 75 deletions(-)
diff --git a/docker/requirements.txt b/docker/requirements.txt
index 715077678..9272e430d 100644
--- a/docker/requirements.txt
+++ b/docker/requirements.txt
@@ -4,7 +4,7 @@ behavex==2.0.1
docker==5.0.0
kafka-python==2.0.2
confluent-kafka==1.7.0
-PyYAML==5.4.1
+PyYAML==6.0.1
m2crypto==0.38.0
watchdog==2.1.2
pyopenssl==23.0.0
diff --git a/docker/test/integration/cluster/DockerTestDirectoryBindings.py b/docker/test/integration/cluster/DockerTestDirectoryBindings.py
index a27163e69..fec53dfb1 100644
--- a/docker/test/integration/cluster/DockerTestDirectoryBindings.py
+++ b/docker/test/integration/cluster/DockerTestDirectoryBindings.py
@@ -42,6 +42,7 @@ class DockerTestDirectoryBindings:
"input_dir": "/tmp/.nifi-test-input." + self.feature_id,
"output_dir": "/tmp/.nifi-test-output." + self.feature_id,
"resources_dir": "/tmp/.nifi-test-resources." + self.feature_id,
+ "system_certs_dir": "/tmp/.nifi-test-resources." + self.feature_id + "/system_certs_dir",
"minifi_config_dir": "/tmp/.nifi-test-minifi-config-dir." + self.feature_id,
"nifi_config_dir": "/tmp/.nifi-test-nifi-config-dir." + self.feature_id,
"kubernetes_temp_dir": "/tmp/.nifi-test-kubernetes-temp-dir." + self.feature_id,
@@ -87,6 +88,7 @@ class DockerTestDirectoryBindings:
vols[self.data_directories[feature_id]["input_dir"]] = {"bind": "/tmp/input", "mode": "rw"}
vols[self.data_directories[feature_id]["output_dir"]] = {"bind": "/tmp/output", "mode": "rw"}
vols[self.data_directories[feature_id]["resources_dir"]] = {"bind": "/tmp/resources", "mode": "rw"}
+ vols[self.data_directories[feature_id]["system_certs_dir"]] = {"bind": "/usr/local/share/certs", "mode": "rw"}
vols[self.data_directories[feature_id]["minifi_config_dir"]] = {"bind": "/tmp/minifi_config", "mode": "rw"}
vols[self.data_directories[feature_id]["nifi_config_dir"]] = {"bind": "/tmp/nifi_config", "mode": "rw"}
vols[self.data_directories[feature_id]["kubernetes_config_dir"]] = {"bind": "/tmp/kubernetes_config", "mode": "rw"}
diff --git a/docker/test/integration/features/MiNiFi_integration_test_driver.py b/docker/test/integration/features/MiNiFi_integration_test_driver.py
index cc774c1c2..ea0b2370d 100644
--- a/docker/test/integration/features/MiNiFi_integration_test_driver.py
+++ b/docker/test/integration/features/MiNiFi_integration_test_driver.py
@@ -22,7 +22,7 @@ import OpenSSL.crypto
from pydoc import locate
-from ssl_utils.SSL_cert_utils import make_ca, make_cert_without_extended_usage
+from ssl_utils.SSL_cert_utils import make_self_signed_cert, make_cert_without_extended_usage, make_server_cert
from minifi.core.InputPort import InputPort
from cluster.DockerTestCluster import DockerTestCluster
@@ -52,22 +52,38 @@ class MiNiFi_integration_test:
self.docker_directory_bindings = context.directory_bindings
self.cluster.set_directory_bindings(self.docker_directory_bindings.get_directory_bindings(self.feature_id), self.docker_directory_bindings.get_data_directories(self.feature_id))
- self.root_ca_cert, self.root_ca_key = make_ca("root CA")
+ self.root_ca_cert, self.root_ca_key = make_self_signed_cert("root CA")
minifi_client_cert, minifi_client_key = make_cert_without_extended_usage(common_name=f"minifi-cpp-flow-{self.feature_id}",
ca_cert=self.root_ca_cert,
ca_key=self.root_ca_key)
+ minifi_server_cert, minifi_server_key = make_server_cert(common_name=f"server-{self.feature_id}",
+ ca_cert=self.root_ca_cert,
+ ca_key=self.root_ca_key)
+ self_signed_server_cert, self_signed_server_key = make_self_signed_cert(f"server-{self.feature_id}")
+
self.put_test_resource('root_ca.crt',
OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
cert=self.root_ca_cert))
-
+ self.put_test_resource("system_certs_dir/ca-root-nss.crt",
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=self.root_ca_cert))
self.put_test_resource('minifi_client.crt',
OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
cert=minifi_client_cert))
self.put_test_resource('minifi_client.key',
OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
pkey=minifi_client_key))
-
+ self.put_test_resource('minifi_server.crt',
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=minifi_server_cert)
+ + OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
+ pkey=minifi_server_key))
+ self.put_test_resource('self_signed_server.crt',
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=self_signed_server_cert)
+ + OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
+ pkey=self_signed_server_key))
self.put_test_resource('minifi_merged_cert.crt',
OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
cert=minifi_client_cert)
diff --git a/docker/test/integration/features/https.feature b/docker/test/integration/features/https.feature
index f7f3cae4f..f19e339d1 100644
--- a/docker/test/integration/features/https.feature
+++ b/docker/test/integration/features/https.feature
@@ -14,26 +14,155 @@
# limitations under the License.
@CORE
-Feature: Using SSL context service to send data with TLS
- In order to send data via HTTPS
- As a user of MiNiFi
- I need to have access to the SSLContextService
+Feature: Transfer data from and to MiNiFi using HTTPS
Background:
Given the content of "/tmp/output" is monitored
- Scenario: A MiNiFi instance sends data using InvokeHTTP to a receiver using ListenHTTP with TLS
- Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
- And 200 files with the content "test" are present in "/tmp/input"
- And a InvokeHTTP processor with the "Remote URL" property set to "https://secondary-${feature_id}:4430/contentListener"
+
+ Scenario: InvokeHTTP to ListenHTTP without an SSLContextService works (no mutual TLS in this case)
+ Given a GenerateFlowFile processor with the "Data Format" property set to "Text"
+ And the "Unique FlowFiles" property of the GenerateFlowFile processor is set to "false"
+ And the "Custom Text" property of the GenerateFlowFile processor is set to "Lorem ipsum dolor sit amet"
+ And a InvokeHTTP processor with the "Remote URL" property set to "https://server-${feature_id}:4430/contentListener"
And the "HTTP Method" property of the InvokeHTTP processor is set to "POST"
+ And the "success" relationship of the GenerateFlowFile processor is connected to the InvokeHTTP
+
+ And a ListenHTTP processor with the "Listening Port" property set to "4430" in a "server" flow
+ And the "SSL Certificate" property of the ListenHTTP processor is set to "/tmp/resources/minifi_server.crt"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "server" flow
+ And the "success" relationship of the ListenHTTP processor is connected to the PutFile
+
+ When both instances start up
+ Then a flowfile with the content "Lorem ipsum dolor sit amet" is placed in the monitored directory in less than 10s
+
+
+ Scenario: InvokeHTTP to ListenHTTP without an SSLContextService requires a server cert signed by a CA
+ Given a GenerateFlowFile processor with the "Data Format" property set to "Text"
+ And the "Unique FlowFiles" property of the GenerateFlowFile processor is set to "false"
+ And the "Custom Text" property of the GenerateFlowFile processor is set to "consectetur adipiscing elit"
+ And a InvokeHTTP processor with the "Remote URL" property set to "https://server-${feature_id}:4430/contentListener"
+ And the "HTTP Method" property of the InvokeHTTP processor is set to "POST"
+ And the "success" relationship of the GenerateFlowFile processor is connected to the InvokeHTTP
+
+ And a ListenHTTP processor with the "Listening Port" property set to "4430" in a "server" flow
+ And the "SSL Certificate" property of the ListenHTTP processor is set to "/tmp/resources/self_signed_server.crt"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "server" flow
+ And the "success" relationship of the ListenHTTP processor is connected to the PutFile
+
+ When both instances start up
+ Then no files are placed in the monitored directory in 10s of running time
+
- And the "success" relationship of the GetFile processor is connected to the InvokeHTTP
+ Scenario: InvokeHTTP to ListenHTTP without an SSLContextService works without a proper server cert if peer verification is disabled
+ Given a GenerateFlowFile processor with the "Data Format" property set to "Text"
+ And the "Unique FlowFiles" property of the GenerateFlowFile processor is set to "false"
+ And the "Custom Text" property of the GenerateFlowFile processor is set to "sed do eiusmod tempor incididunt"
+ And a InvokeHTTP processor with the "Remote URL" property set to "https://server-${feature_id}:4430/contentListener"
+ And the "HTTP Method" property of the InvokeHTTP processor is set to "POST"
+ And the "Disable Peer Verification" property of the InvokeHTTP processor is set to "true"
+ And the "success" relationship of the GenerateFlowFile processor is connected to the InvokeHTTP
+
+ And a ListenHTTP processor with the "Listening Port" property set to "4430" in a "server" flow
+ And the "SSL Certificate" property of the ListenHTTP processor is set to "/tmp/resources/self_signed_server.crt"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "server" flow
+ And the "success" relationship of the ListenHTTP processor is connected to the PutFile
+
+ When both instances start up
+ Then a flowfile with the content "sed do eiusmod tempor incididunt" is placed in the monitored directory in less than 10s
+
+
+ Scenario: InvokeHTTP to ListenHTTP with mutual TLS, using certificate files
+ Given a GenerateFlowFile processor with the "Data Format" property set to "Text"
+ And the "Unique FlowFiles" property of the GenerateFlowFile processor is set to "false"
+ And the "Custom Text" property of the GenerateFlowFile processor is set to "ut labore et dolore magna aliqua"
+ And a InvokeHTTP processor with the "Remote URL" property set to "https://server-${feature_id}:4430/contentListener"
+ And the "HTTP Method" property of the InvokeHTTP processor is set to "POST"
+ And an ssl context service with a manual CA cert file is set up for InvokeHTTP
+ And the "success" relationship of the GenerateFlowFile processor is connected to the InvokeHTTP
+
+ And a ListenHTTP processor with the "Listening Port" property set to "4430" in a "server" flow
+ And the "SSL Certificate" property of the ListenHTTP processor is set to "/tmp/resources/minifi_server.crt"
+ And the "SSL Certificate Authority" property of the ListenHTTP processor is set to "/usr/local/share/certs/ca-root-nss.crt"
+ And the "SSL Verify Peer" property of the ListenHTTP processor is set to "yes"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "server" flow
+ And the "success" relationship of the ListenHTTP processor is connected to the PutFile
+
+ When both instances start up
+ Then a flowfile with the content "ut labore et dolore magna aliqua" is placed in the monitored directory in less than 10s
+
+
+ Scenario: InvokeHTTP to ListenHTTP without mutual TLS, using the system certificate store
+ Given a GenerateFlowFile processor with the "Data Format" property set to "Text"
+ And the "Unique FlowFiles" property of the GenerateFlowFile processor is set to "false"
+ And the "Custom Text" property of the GenerateFlowFile processor is set to "Ut enim ad minim veniam"
+ And a InvokeHTTP processor with the "Remote URL" property set to "https://server-${feature_id}:4430/contentListener"
+ And the "HTTP Method" property of the InvokeHTTP processor is set to "POST"
+ And an ssl context service using the system CA cert store is set up for InvokeHTTP
+ And the "success" relationship of the GenerateFlowFile processor is connected to the InvokeHTTP
+
+ And a ListenHTTP processor with the "Listening Port" property set to "4430" in a "server" flow
+ And the "SSL Certificate" property of the ListenHTTP processor is set to "/tmp/resources/minifi_server.crt"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "server" flow
+ And the "success" relationship of the ListenHTTP processor is connected to the PutFile
+
+ When both instances start up
+ Then a flowfile with the content "Ut enim ad minim veniam" is placed in the monitored directory in less than 10s
+
+
+ Scenario: InvokeHTTP to ListenHTTP with mutual TLS, using the system certificate store
+ Given a GenerateFlowFile processor with the "Data Format" property set to "Text"
+ And the "Unique FlowFiles" property of the GenerateFlowFile processor is set to "false"
+ And the "Custom Text" property of the GenerateFlowFile processor is set to "quis nostrud exercitation ullamco laboris nisi"
+ And a InvokeHTTP processor with the "Remote URL" property set to "https://server-${feature_id}:4430/contentListener"
+ And the "HTTP Method" property of the InvokeHTTP processor is set to "POST"
+ And an ssl context service using the system CA cert store is set up for InvokeHTTP
+ And the "success" relationship of the GenerateFlowFile processor is connected to the InvokeHTTP
+
+ And a ListenHTTP processor with the "Listening Port" property set to "4430" in a "server" flow
+ And the "SSL Certificate" property of the ListenHTTP processor is set to "/tmp/resources/minifi_server.crt"
+ And the "SSL Certificate Authority" property of the ListenHTTP processor is set to "/usr/local/share/certs/ca-root-nss.crt"
+ And the "SSL Verify Peer" property of the ListenHTTP processor is set to "yes"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "server" flow
+ And the "success" relationship of the ListenHTTP processor is connected to the PutFile
+
+ When both instances start up
+ Then a flowfile with the content "quis nostrud exercitation ullamco laboris nisi" is placed in the monitored directory in less than 10s
+
+
+ Scenario: InvokeHTTP to ListenHTTP without mutual TLS, using the system certificate store, requires a server cert signed by a CA
+ Given a GenerateFlowFile processor with the "Data Format" property set to "Text"
+ And the "Unique FlowFiles" property of the GenerateFlowFile processor is set to "false"
+ And the "Custom Text" property of the GenerateFlowFile processor is set to "ut aliquip ex ea commodo consequat"
+ And a InvokeHTTP processor with the "Remote URL" property set to "https://server-${feature_id}:4430/contentListener"
+ And the "HTTP Method" property of the InvokeHTTP processor is set to "POST"
+ And an ssl context service using the system CA cert store is set up for InvokeHTTP
+ And the "success" relationship of the GenerateFlowFile processor is connected to the InvokeHTTP
+
+ And a ListenHTTP processor with the "Listening Port" property set to "4430" in a "server" flow
+ And the "SSL Certificate" property of the ListenHTTP processor is set to "/tmp/resources/self_signed_server.crt"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "server" flow
+ And the "success" relationship of the ListenHTTP processor is connected to the PutFile
+
+ When both instances start up
+ Then no files are placed in the monitored directory in 10s of running time
+
+
+ Scenario: InvokeHTTP to ListenHTTP with mutual TLS, using the system certificate store, requires a server cert signed by a CA
+ Given a GenerateFlowFile processor with the "Data Format" property set to "Text"
+ And the "Unique FlowFiles" property of the GenerateFlowFile processor is set to "false"
+ And the "Custom Text" property of the GenerateFlowFile processor is set to "Duis aute irure dolor in reprehenderit in voluptate"
+ And a InvokeHTTP processor with the "Remote URL" property set to "https://server-${feature_id}:4430/contentListener"
+ And the "HTTP Method" property of the InvokeHTTP processor is set to "POST"
+ And an ssl context service using the system CA cert store is set up for InvokeHTTP
+ And the "success" relationship of the GenerateFlowFile processor is connected to the InvokeHTTP
- And a ListenHTTP processor with the "Listening Port" property set to "4430" in a "secondary" flow
- And a PutFile processor with the "Directory" property set to "/tmp/output" in the "secondary" flow
+ And a ListenHTTP processor with the "Listening Port" property set to "4430" in a "server" flow
+ And the "SSL Certificate" property of the ListenHTTP processor is set to "/tmp/resources/self_signed_server.crt"
+ And the "SSL Certificate Authority" property of the ListenHTTP processor is set to "/usr/local/share/certs/ca-root-nss.crt"
+ And the "SSL Verify Peer" property of the ListenHTTP processor is set to "yes"
+ And a PutFile processor with the "Directory" property set to "/tmp/output" in the "server" flow
And the "success" relationship of the ListenHTTP processor is connected to the PutFile
- And an ssl context service set up for InvokeHTTP and ListenHTTP
When both instances start up
- Then 200 flowfiles with the content "test" are placed in the monitored directory in less than 60 seconds
+ Then no files are placed in the monitored directory in 10s of running time
diff --git a/docker/test/integration/features/kafka.feature b/docker/test/integration/features/kafka.feature
index 0a48b550d..7fcae658f 100644
--- a/docker/test/integration/features/kafka.feature
+++ b/docker/test/integration/features/kafka.feature
@@ -175,7 +175,7 @@ Feature: Sending data to using Kafka streaming platform using PublishKafka
| PublishKafka | Username | alice |
| PublishKafka | Password | alice-secret |
And a PutFile processor with the "Directory" property set to "/tmp/output"
- And an ssl context service set up for PublishKafka
+ And an ssl context service is set up for PublishKafka
And the "success" relationship of the GetFile processor is connected to the PublishKafka
And the "success" relationship of the PublishKafka processor is connected to the PutFile
@@ -200,7 +200,7 @@ Feature: Sending data to using Kafka streaming platform using PublishKafka
| PublishKafka | Message Timeout | 12 sec |
| PublishKafka | Security Protocol | ssl |
And a PutFile processor with the "Directory" property set to "/tmp/output"
- And an ssl context service set up for PublishKafka
+ And an ssl context service is set up for PublishKafka
And the "success" relationship of the GetFile processor is connected to the PublishKafka
And the "success" relationship of the PublishKafka processor is connected to the PutFile
@@ -414,7 +414,7 @@ Feature: Sending data to using Kafka streaming platform using PublishKafka
| ConsumeKafka | Kafka Brokers | kafka-broker-${feature_id}:9093 |
| ConsumeKafka | Security Protocol | ssl |
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
- And an ssl context service set up for ConsumeKafka
+ And an ssl context service is set up for ConsumeKafka
And the "success" relationship of the ConsumeKafka processor is connected to the PutFile
And a kafka broker is set up in correspondence with the publisher flow
@@ -435,7 +435,7 @@ Feature: Sending data to using Kafka streaming platform using PublishKafka
| ConsumeKafka | Username | alice |
| ConsumeKafka | Password | alice-secret |
And a PutFile processor with the "Directory" property set to "/tmp/output" in the "kafka-consumer-flow" flow
- And an ssl context service set up for ConsumeKafka
+ And an ssl context service is set up for ConsumeKafka
And the "success" relationship of the ConsumeKafka processor is connected to the PutFile
And a kafka broker is set up in correspondence with the publisher flow
diff --git a/docker/test/integration/features/steps/steps.py b/docker/test/integration/features/steps/steps.py
index d4cbfb15a..b52862ffa 100644
--- a/docker/test/integration/features/steps/steps.py
+++ b/docker/test/integration/features/steps/steps.py
@@ -371,34 +371,27 @@ def step_impl(context):
# TLS
-@given("an ssl context service set up for {producer_name} and {consumer_name}")
-def step_impl(context, producer_name, consumer_name):
- minifi_crt_file = '/tmp/resources/minifi_client.crt'
- minifi_key_file = '/tmp/resources/minifi_client.key'
- root_ca_crt_file = '/tmp/resources/root_ca.crt'
- ssl_context_service = SSLContextService(cert=minifi_crt_file, ca_cert=root_ca_crt_file, key=minifi_key_file)
+@given("an ssl context service is set up for {processor_name}")
+@given("an ssl context service with a manual CA cert file is set up for {processor_name}")
+def step_impl(context, processor_name):
+ ssl_context_service = SSLContextService(cert='/tmp/resources/minifi_client.crt',
+ key='/tmp/resources/minifi_client.key',
+ ca_cert='/tmp/resources/root_ca.crt')
- secondary_cert, secondary_key = make_server_cert(context.test.get_container_name_with_postfix("secondary"), context.test.root_ca_cert, context.test.root_ca_key)
- secondary_pem_file = '/tmp/resources/secondary.crt'
- context.test.put_test_resource('secondary.crt', OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM, cert=secondary_cert) + OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM, pkey=secondary_key))
- producer = context.test.get_node_by_name(producer_name)
- producer.controller_services.append(ssl_context_service)
- producer.set_property("SSL Context Service", ssl_context_service.name)
- consumer = context.test.get_node_by_name(consumer_name)
- consumer.set_property("SSL Certificate Authority", root_ca_crt_file)
- consumer.set_property("SSL Certificate", secondary_pem_file)
- consumer.set_property("SSL Verify Peer", "yes")
+ processor = context.test.get_node_by_name(processor_name)
+ processor.controller_services.append(ssl_context_service)
+ processor.set_property('SSL Context Service', ssl_context_service.name)
-@given("an ssl context service set up for {producer_name}")
-def step_impl(context, producer_name):
- minifi_crt_file = '/tmp/resources/minifi_client.crt'
- minifi_key_file = '/tmp/resources/minifi_client.key'
- root_ca_crt_file = '/tmp/resources/root_ca.crt'
- ssl_context_service = SSLContextService(cert=minifi_crt_file, ca_cert=root_ca_crt_file, key=minifi_key_file)
- producer = context.test.get_node_by_name(producer_name)
- producer.controller_services.append(ssl_context_service)
- producer.set_property("SSL Context Service", ssl_context_service.name)
+@given("an ssl context service using the system CA cert store is set up for {processor_name}")
+def step_impl(context, processor_name):
+ ssl_context_service = SSLContextService(cert='/tmp/resources/minifi_client.crt',
+ key='/tmp/resources/minifi_client.key',
+ use_system_cert_store='true')
+
+ processor = context.test.get_node_by_name(processor_name)
+ processor.controller_services.append(ssl_context_service)
+ processor.set_property('SSL Context Service', ssl_context_service.name)
# Kubernetes
diff --git a/docker/test/integration/minifi/controllers/SSLContextService.py b/docker/test/integration/minifi/controllers/SSLContextService.py
index 01ff7b5ac..af16368f3 100644
--- a/docker/test/integration/minifi/controllers/SSLContextService.py
+++ b/docker/test/integration/minifi/controllers/SSLContextService.py
@@ -18,7 +18,7 @@ from ..core.ControllerService import ControllerService
class SSLContextService(ControllerService):
- def __init__(self, name=None, cert=None, key=None, ca_cert=None, passphrase=None):
+ def __init__(self, name=None, cert=None, key=None, ca_cert=None, passphrase=None, use_system_cert_store=None):
super(SSLContextService, self).__init__(name=name)
self.service_class = 'SSLContextService'
@@ -34,3 +34,6 @@ class SSLContextService(ControllerService):
if passphrase is not None:
self.properties['Passphrase'] = passphrase
+
+ if use_system_cert_store is not None:
+ self.properties['Use System Cert Store'] = use_system_cert_store
diff --git a/docker/test/integration/ssl_utils/SSL_cert_utils.py b/docker/test/integration/ssl_utils/SSL_cert_utils.py
index 113071e48..d39cf9646 100644
--- a/docker/test/integration/ssl_utils/SSL_cert_utils.py
+++ b/docker/test/integration/ssl_utils/SSL_cert_utils.py
@@ -75,7 +75,7 @@ def gen_req():
return req, key
-def make_ca(common_name):
+def make_self_signed_cert(common_name):
ca_key = crypto.PKey()
ca_key.generate_key(crypto.TYPE_RSA, 2048)
diff --git a/extensions/aws/processors/S3Processor.cpp b/extensions/aws/processors/S3Processor.cpp
index bf0b9d184..0d6e983e5 100644
--- a/extensions/aws/processors/S3Processor.cpp
+++ b/extensions/aws/processors/S3Processor.cpp
@@ -126,9 +126,9 @@ void S3Processor::onSchedule(const std::shared_ptr<core::ProcessContext>& contex
throw Exception(PROCESS_SCHEDULE_EXCEPTION, "Communications Timeout missing or invalid");
}
- static const auto default_ca_path = minifi::utils::getDefaultCAPath();
- if (default_ca_path) {
- client_config_->caFile = default_ca_path->string();
+ static const auto default_ca_file = minifi::utils::getDefaultCAFile();
+ if (default_ca_file) {
+ client_config_->caFile = *default_ca_file;
}
}
diff --git a/extensions/http-curl/client/HTTPClient.cpp b/extensions/http-curl/client/HTTPClient.cpp
index f375dc772..af131f3d6 100644
--- a/extensions/http-curl/client/HTTPClient.cpp
+++ b/extensions/http-curl/client/HTTPClient.cpp
@@ -44,8 +44,8 @@ HTTPClient::HTTPClient(std::string url, std::shared_ptr<minifi::controllers::SSL
http_session_.reset(curl_easy_init());
}
-HTTPClient::HTTPClient(std::string name, const utils::Identifier& uuid)
- : core::Connectable(std::move(name), uuid) {
+HTTPClient::HTTPClient(std::string_view name, const utils::Identifier& uuid)
+ : core::Connectable(name, uuid) {
http_session_.reset(curl_easy_init());
}
@@ -125,7 +125,11 @@ void HTTPClient::initialize(std::string method, std::string url, std::shared_ptr
}
void HTTPClient::setPeerVerification(bool peer_verification) {
- logger_->log_debug("%s peer verification", peer_verification ? "Enabling" : "Disabling");
+ if (peer_verification) {
+ logger_->log_debug("Enabling peer verification");
+ } else {
+ logger_->log_warn("Disabling peer verification: the authenticity of https servers will not be verified!");
+ }
curl_easy_setopt(http_session_.get(), CURLOPT_SSL_VERIFYPEER, peer_verification);
}
@@ -458,26 +462,22 @@ int HTTPClient::onProgress(void *clientp, curl_off_t /*dltotal*/, curl_off_t dln
void HTTPClient::configure_secure_connection() {
#ifdef OPENSSL_SUPPORT
if (ssl_context_service_) {
- logger_->log_debug("Using certificate file \"%s\"", ssl_context_service_->getCertificateFile().string());
- logger_->log_debug("Using private key file \"%s\"", ssl_context_service_->getPrivateKeyFile().string());
- logger_->log_debug("Using CA certificate file \"%s\"", ssl_context_service_->getCACertificate().string());
-
curl_easy_setopt(http_session_.get(), CURLOPT_SSL_CTX_FUNCTION, &configure_ssl_context);
curl_easy_setopt(http_session_.get(), CURLOPT_SSL_CTX_DATA, static_cast<void *>(ssl_context_service_.get()));
curl_easy_setopt(http_session_.get(), CURLOPT_CAINFO, nullptr);
curl_easy_setopt(http_session_.get(), CURLOPT_CAPATH, nullptr);
} else {
- static const auto default_ca_path = utils::getDefaultCAPath();
+ static const auto default_ca_file = utils::getDefaultCAFile();
- if (default_ca_path)
- logger_->log_debug("Using CA certificate file \"%s\"", default_ca_path->string());
+ if (default_ca_file)
+ logger_->log_debug("Using CA certificate file \"%s\"", std::string(*default_ca_file));
else
logger_->log_error("Could not find valid CA certificate file");
curl_easy_setopt(http_session_.get(), CURLOPT_SSL_CTX_FUNCTION, nullptr);
curl_easy_setopt(http_session_.get(), CURLOPT_SSL_CTX_DATA, nullptr);
- if (default_ca_path)
- curl_easy_setopt(http_session_.get(), CURLOPT_CAINFO, default_ca_path->string().c_str());
+ if (default_ca_file)
+ curl_easy_setopt(http_session_.get(), CURLOPT_CAINFO, std::string(*default_ca_file).c_str());
else
curl_easy_setopt(http_session_.get(), CURLOPT_CAINFO, nullptr);
curl_easy_setopt(http_session_.get(), CURLOPT_CAPATH, nullptr);
diff --git a/extensions/http-curl/client/HTTPClient.h b/extensions/http-curl/client/HTTPClient.h
index 30c10f486..2e98bab67 100644
--- a/extensions/http-curl/client/HTTPClient.h
+++ b/extensions/http-curl/client/HTTPClient.h
@@ -68,7 +68,7 @@ class HTTPClient : public utils::BaseHTTPClient, public core::Connectable {
public:
HTTPClient();
- HTTPClient(std::string name, const utils::Identifier& uuid);
+ HTTPClient(std::string_view name, const utils::Identifier& uuid);
HTTPClient(const HTTPClient&) = delete;
HTTPClient& operator=(const HTTPClient&) = delete;
diff --git a/extensions/http-curl/processors/InvokeHTTP.cpp b/extensions/http-curl/processors/InvokeHTTP.cpp
index b9f1799c4..eab6d0804 100644
--- a/extensions/http-curl/processors/InvokeHTTP.cpp
+++ b/extensions/http-curl/processors/InvokeHTTP.cpp
@@ -72,7 +72,7 @@ void setupClientProxy(extensions::curl::HTTPClient& client, const core::ProcessC
void setupClientPeerVerification(extensions::curl::HTTPClient& client, const core::ProcessContext& context) {
if (auto disable_peer_verification = context.getProperty(InvokeHTTP::DisablePeerVerification) | utils::flatMap(&utils::StringUtils::toBool)) {
- client.setPeerVerification(*disable_peer_verification);
+ client.setPeerVerification(!*disable_peer_verification);
}
}
diff --git a/extensions/http-curl/tests/VerifyInvokeHTTPGetTest.cpp b/extensions/http-curl/tests/VerifyInvokeHTTPGetTest.cpp
index 9b923fba1..c509877d2 100644
--- a/extensions/http-curl/tests/VerifyInvokeHTTPGetTest.cpp
+++ b/extensions/http-curl/tests/VerifyInvokeHTTPGetTest.cpp
@@ -20,8 +20,6 @@
#include "HTTPHandlers.h"
#include "utils/IntegrationTestUtils.h"
-#define CURLOPT_SSL_VERIFYPEER_DISABLE 1
-
class VerifyHTTPGet : public VerifyInvokeHTTP {
public:
void runAssertions() override {
diff --git a/libminifi/include/utils/HTTPUtils.h b/libminifi/include/utils/HTTPUtils.h
index f748b1d54..3aa5bd3df 100644
--- a/libminifi/include/utils/HTTPUtils.h
+++ b/libminifi/include/utils/HTTPUtils.h
@@ -19,8 +19,8 @@
#pragma once
#include <string>
+#include <string_view>
#include <optional>
-#include <filesystem>
#include "io/ClientSocket.h"
#include "utils/RegexUtils.h"
@@ -52,6 +52,6 @@ inline bool parse_http_components(const std::string &url, std::string &port, std
return false;
}
-std::optional<std::filesystem::path> getDefaultCAPath();
+std::optional<std::string_view> getDefaultCAFile();
} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/src/controllers/SSLContextService.cpp b/libminifi/src/controllers/SSLContextService.cpp
index 32a697ead..1631fcaed 100644
--- a/libminifi/src/controllers/SSLContextService.cpp
+++ b/libminifi/src/controllers/SSLContextService.cpp
@@ -36,6 +36,7 @@
#include "core/Resource.h"
#include "io/validation.h"
#include "properties/Configure.h"
+#include "utils/HTTPUtils.h"
#include "utils/tls/CertificateUtils.h"
#include "utils/tls/TLSUtils.h"
#include "utils/tls/DistinguishedName.h"
@@ -313,7 +314,12 @@ bool SSLContextService::addServerCertificatesFromSystemStoreToSSLContext(SSL_CTX
return true;
#else
- SSL_CTX_set_default_verify_paths(ctx);
+ static const auto default_ca_file = utils::getDefaultCAFile();
+ if (default_ca_file) {
+ SSL_CTX_load_verify_file(ctx, std::string(*default_ca_file).c_str());
+ } else {
+ SSL_CTX_set_default_verify_paths(ctx);
+ }
return true;
#endif // WIN32
}
@@ -498,6 +504,11 @@ void SSLContextService::onEnable() {
client_cert_key_usage_ = utils::tls::ExtendedKeyUsage{client_cert_key_usage};
#endif // WIN32
+ logger_->log_debug("Using certificate file \"%s\"", certificate_.string());
+ logger_->log_debug("Using private key file \"%s\"", private_key_.string());
+ logger_->log_debug("Using CA certificate file \"%s\"", ca_certificate_.string());
+ logger_->log_debug("Using the system cert store: %s", use_system_cert_store_ ? "yes" : "no");
+
verifyCertificateExpiration();
}
diff --git a/libminifi/src/utils/HTTPUtils.cpp b/libminifi/src/utils/HTTPUtils.cpp
index 30b84150f..96219e7cd 100644
--- a/libminifi/src/utils/HTTPUtils.cpp
+++ b/libminifi/src/utils/HTTPUtils.cpp
@@ -17,17 +17,18 @@
*/
#include "utils/HTTPUtils.h"
-#include <vector>
+#include <array>
+#include <filesystem>
namespace org::apache::nifi::minifi::utils {
-std::optional<std::filesystem::path> getDefaultCAPath() {
+std::optional<std::string_view> getDefaultCAFile() {
#ifndef WIN32
- const std::vector<std::filesystem::path> possible_ca_paths = {
+ static constexpr std::array<std::string_view, 5> possible_ca_paths = {
+ "/usr/local/share/certs/ca-root-nss.crt",
"/etc/ssl/certs/ca-certificates.crt",
"/etc/pki/tls/certs/ca-bundle.crt",
"/usr/share/ssl/certs/ca-bundle.crt",
- "/usr/local/share/certs/ca-root-nss.crt",
"/etc/ssl/cert.pem"
};
[nifi-minifi-cpp] 05/05: MINIFICPP-2164 Use a list of types for allowed types in PropertyDefinition
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit fbd69e7e211b3f2fbd72d8ce806ac025aa63b6e0
Author: Ferenc Gerlits <fg...@gmail.com>
AuthorDate: Wed Aug 9 14:50:53 2023 +0200
MINIFICPP-2164 Use a list of types for allowed types in PropertyDefinition
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #1626
---
extensions/elasticsearch/PostElasticsearch.h | 9 +-
extensions/gcp/processors/GCSProcessor.h | 4 +-
extensions/http-curl/processors/InvokeHTTP.h | 4 +-
extensions/librdkafka/KafkaProcessorBase.h | 52 ++++++------
extensions/splunk/SplunkHECProcessor.h | 40 ++++-----
extensions/standard-processors/processors/GetTCP.h | 4 +-
.../standard-processors/processors/ListenSyslog.h | 4 +-
.../standard-processors/processors/ListenTCP.h | 4 +-
.../standard-processors/processors/PutFile.h | 2 +-
extensions/standard-processors/processors/PutTCP.h | 98 +++++++++++-----------
.../standard-processors/processors/TailFile.h | 4 +-
.../tests/unit/YamlConfigurationTests.cpp | 8 +-
libminifi/include/core/OutputAttributeDefinition.h | 4 +-
libminifi/include/core/PropertyDefinition.h | 18 ++--
libminifi/include/core/PropertyDefinitionBuilder.h | 41 +++++----
libminifi/src/core/Property.cpp | 6 +-
libminifi/test/unit/ComponentManifestTests.cpp | 4 +-
libminifi/test/unit/TypeListTests.cpp | 61 ++++++++++++++
18 files changed, 217 insertions(+), 150 deletions(-)
diff --git a/extensions/elasticsearch/PostElasticsearch.h b/extensions/elasticsearch/PostElasticsearch.h
index 367e35d21..ec213b982 100644
--- a/extensions/elasticsearch/PostElasticsearch.h
+++ b/extensions/elasticsearch/PostElasticsearch.h
@@ -53,16 +53,16 @@ class PostElasticsearch : public core::Processor {
.withPropertyType(core::StandardPropertyTypes::UNSIGNED_LONG_TYPE)
.withDefaultValue("100")
.build();
- EXTENSIONAPI static constexpr auto ElasticCredentials = core::PropertyDefinitionBuilder<0, 1>::createProperty("Elasticsearch Credentials Provider Service")
+ EXTENSIONAPI static constexpr auto ElasticCredentials = core::PropertyDefinitionBuilder<>::createProperty("Elasticsearch Credentials Provider Service")
.withDescription("The Controller Service used to obtain Elasticsearch credentials.")
.isRequired(true)
- .withAllowedTypes({core::className<ElasticsearchCredentialsControllerService>()})
+ .withAllowedTypes<ElasticsearchCredentialsControllerService>()
.build();
- EXTENSIONAPI static constexpr auto SSLContext = core::PropertyDefinitionBuilder<0, 1>::createProperty("SSL Context Service")
+ EXTENSIONAPI static constexpr auto SSLContext = core::PropertyDefinitionBuilder<>::createProperty("SSL Context Service")
.withDescription("The SSL Context Service used to provide client certificate "
"information for TLS/SSL (https) connections.")
.isRequired(false)
- .withAllowedTypes({core::className<minifi::controllers::SSLContextService>()})
+ .withAllowedTypes<minifi::controllers::SSLContextService>()
.build();
EXTENSIONAPI static constexpr auto Hosts = core::PropertyDefinitionBuilder<>::createProperty("Hosts")
.withDescription("A comma-separated list of HTTP hosts that host Elasticsearch query nodes. Currently only supports a single host.")
@@ -90,7 +90,6 @@ class PostElasticsearch : public core::Processor {
Identifier
};
-
EXTENSIONAPI static constexpr auto Success = core::RelationshipDefinition{"success", "All flowfiles that succeed in being transferred into Elasticsearch go here."};
EXTENSIONAPI static constexpr auto Failure = core::RelationshipDefinition{"failure", "All flowfiles that fail for reasons unrelated to server availability go to this relationship."};
EXTENSIONAPI static constexpr auto Error = core::RelationshipDefinition{"error", "All flowfiles that Elasticsearch responded to with an error go to this relationship."};
diff --git a/extensions/gcp/processors/GCSProcessor.h b/extensions/gcp/processors/GCSProcessor.h
index b54fbe691..8cfbe9c1d 100644
--- a/extensions/gcp/processors/GCSProcessor.h
+++ b/extensions/gcp/processors/GCSProcessor.h
@@ -39,10 +39,10 @@ class GCSProcessor : public core::Processor {
logger_(std::move(logger)) {
}
- EXTENSIONAPI static constexpr auto GCPCredentials = core::PropertyDefinitionBuilder<0, 1>::createProperty("GCP Credentials Provider Service")
+ EXTENSIONAPI static constexpr auto GCPCredentials = core::PropertyDefinitionBuilder<>::createProperty("GCP Credentials Provider Service")
.withDescription("The Controller Service used to obtain Google Cloud Platform credentials. Should be the name of a GCPCredentialsControllerService.")
.isRequired(true)
- .withAllowedTypes({core::className<GCPCredentialsControllerService>()})
+ .withAllowedTypes<GCPCredentialsControllerService>()
.build();
EXTENSIONAPI static constexpr auto NumberOfRetries = core::PropertyDefinitionBuilder<>::createProperty("Number of retries")
.withDescription("How many retry attempts should be made before routing to the failure relationship.")
diff --git a/extensions/http-curl/processors/InvokeHTTP.h b/extensions/http-curl/processors/InvokeHTTP.h
index c9d291a34..99a6a8968 100644
--- a/extensions/http-curl/processors/InvokeHTTP.h
+++ b/extensions/http-curl/processors/InvokeHTTP.h
@@ -103,10 +103,10 @@ class InvokeHTTP : public core::Processor {
EXTENSIONAPI static constexpr auto AttributesToSend = core::PropertyDefinitionBuilder<>::createProperty("Attributes to Send")
.withDescription("Regular expression that defines which attributes to send as HTTP headers in the request. If not defined, no attributes are sent as headers.")
.build();
- EXTENSIONAPI static constexpr auto SSLContext = core::PropertyDefinitionBuilder<0, 1, 0, 1>::createProperty("SSL Context Service")
+ EXTENSIONAPI static constexpr auto SSLContext = core::PropertyDefinitionBuilder<0, 0, 1>::createProperty("SSL Context Service")
.withDescription("The SSL Context Service used to provide client certificate information for TLS/SSL (https) connections.")
.isRequired(false)
- .withAllowedTypes({core::className<minifi::controllers::SSLContextService>()})
+ .withAllowedTypes<minifi::controllers::SSLContextService>()
.withExclusiveOfProperties({{{"Remote URL", "^http:.*$"}}})
.build();
EXTENSIONAPI static constexpr auto ProxyHost = core::PropertyDefinitionBuilder<>::createProperty("Proxy Host")
diff --git a/extensions/librdkafka/KafkaProcessorBase.h b/extensions/librdkafka/KafkaProcessorBase.h
index eecc3f2fc..6e88c6689 100644
--- a/extensions/librdkafka/KafkaProcessorBase.h
+++ b/extensions/librdkafka/KafkaProcessorBase.h
@@ -47,37 +47,37 @@ enum class SASLMechanismOption {
class KafkaProcessorBase : public core::Processor {
public:
- EXTENSIONAPI static constexpr auto SSLContextService = core::PropertyDefinitionBuilder<0, 1>::createProperty("SSL Context Service")
- .withDescription("SSL Context Service Name")
- .withAllowedTypes({core::className<minifi::controllers::SSLContextService>()})
- .build();
+ EXTENSIONAPI static constexpr auto SSLContextService = core::PropertyDefinitionBuilder<>::createProperty("SSL Context Service")
+ .withDescription("SSL Context Service Name")
+ .withAllowedTypes<minifi::controllers::SSLContextService>()
+ .build();
EXTENSIONAPI static constexpr auto SecurityProtocol = core::PropertyDefinitionBuilder<magic_enum::enum_count<kafka::SecurityProtocolOption>()>::createProperty("Security Protocol")
- .withDescription("Protocol used to communicate with brokers. Corresponds to Kafka's 'security.protocol' property.")
- .withDefaultValue(magic_enum::enum_name(kafka::SecurityProtocolOption::plaintext))
- .withAllowedValues(magic_enum::enum_names<kafka::SecurityProtocolOption>())
- .isRequired(true)
- .build();
+ .withDescription("Protocol used to communicate with brokers. Corresponds to Kafka's 'security.protocol' property.")
+ .withDefaultValue(magic_enum::enum_name(kafka::SecurityProtocolOption::plaintext))
+ .withAllowedValues(magic_enum::enum_names<kafka::SecurityProtocolOption>())
+ .isRequired(true)
+ .build();
EXTENSIONAPI static constexpr auto KerberosServiceName = core::PropertyDefinitionBuilder<>::createProperty("Kerberos Service Name")
- .withDescription("Kerberos Service Name")
- .build();
+ .withDescription("Kerberos Service Name")
+ .build();
EXTENSIONAPI static constexpr auto KerberosPrincipal = core::PropertyDefinitionBuilder<>::createProperty("Kerberos Principal")
- .withDescription("Kerberos Principal")
- .build();
+ .withDescription("Kerberos Principal")
+ .build();
EXTENSIONAPI static constexpr auto KerberosKeytabPath = core::PropertyDefinitionBuilder<>::createProperty("Kerberos Keytab Path")
- .withDescription("The path to the location on the local filesystem where the kerberos keytab is located. Read permission on the file is required.")
- .build();
+ .withDescription("The path to the location on the local filesystem where the kerberos keytab is located. Read permission on the file is required.")
+ .build();
EXTENSIONAPI static constexpr auto SASLMechanism = core::PropertyDefinitionBuilder<magic_enum::enum_count<kafka::SASLMechanismOption>()>::createProperty("SASL Mechanism")
- .withDescription("The SASL mechanism to use for authentication. Corresponds to Kafka's 'sasl.mechanism' property.")
- .withDefaultValue(magic_enum::enum_name(kafka::SASLMechanismOption::GSSAPI))
- .withAllowedValues(magic_enum::enum_names<kafka::SASLMechanismOption>())
- .isRequired(true)
- .build();
+ .withDescription("The SASL mechanism to use for authentication. Corresponds to Kafka's 'sasl.mechanism' property.")
+ .withDefaultValue(magic_enum::enum_name(kafka::SASLMechanismOption::GSSAPI))
+ .withAllowedValues(magic_enum::enum_names<kafka::SASLMechanismOption>())
+ .isRequired(true)
+ .build();
EXTENSIONAPI static constexpr auto Username = core::PropertyDefinitionBuilder<>::createProperty("Username")
- .withDescription("The username when the SASL Mechanism is sasl_plaintext")
- .build();
+ .withDescription("The username when the SASL Mechanism is sasl_plaintext")
+ .build();
EXTENSIONAPI static constexpr auto Password = core::PropertyDefinitionBuilder<>::createProperty("Password")
- .withDescription("The password for the given username when the SASL Mechanism is sasl_plaintext")
- .build();
+ .withDescription("The password for the given username when the SASL Mechanism is sasl_plaintext")
+ .build();
EXTENSIONAPI static constexpr auto Properties = std::array<core::PropertyReference, 8>{
SSLContextService,
SecurityProtocol,
@@ -92,14 +92,14 @@ class KafkaProcessorBase : public core::Processor {
KafkaProcessorBase(std::string name, const utils::Identifier& uuid, std::shared_ptr<core::logging::Logger> logger)
: core::Processor(std::move(name), uuid),
- logger_(logger) {
+ logger_(std::move(logger)) {
}
protected:
virtual std::optional<utils::net::SslData> getSslData(core::ProcessContext& context) const;
void setKafkaAuthenticationParameters(core::ProcessContext& context, gsl::not_null<rd_kafka_conf_t*> config);
- kafka::SecurityProtocolOption security_protocol_;
+ kafka::SecurityProtocolOption security_protocol_{};
std::shared_ptr<core::logging::Logger> logger_;
};
diff --git a/extensions/splunk/SplunkHECProcessor.h b/extensions/splunk/SplunkHECProcessor.h
index e33a2bf92..bc885ac13 100644
--- a/extensions/splunk/SplunkHECProcessor.h
+++ b/extensions/splunk/SplunkHECProcessor.h
@@ -36,29 +36,29 @@ namespace org::apache::nifi::minifi::extensions::splunk {
class SplunkHECProcessor : public core::Processor {
public:
EXTENSIONAPI static constexpr auto Hostname = core::PropertyDefinitionBuilder<>::createProperty("Hostname")
- .withDescription("The ip address or hostname of the Splunk server.")
- .isRequired(true)
- .build();
+ .withDescription("The ip address or hostname of the Splunk server.")
+ .isRequired(true)
+ .build();
EXTENSIONAPI static constexpr auto Port = core::PropertyDefinitionBuilder<>::createProperty("Port")
- .withDescription("The HTTP Event Collector HTTP Port Number.")
- .withPropertyType(core::StandardPropertyTypes::PORT_TYPE)
- .withDefaultValue("8088")
- .isRequired(true)
- .build();
+ .withDescription("The HTTP Event Collector HTTP Port Number.")
+ .withPropertyType(core::StandardPropertyTypes::PORT_TYPE)
+ .withDefaultValue("8088")
+ .isRequired(true)
+ .build();
EXTENSIONAPI static constexpr auto Token = core::PropertyDefinitionBuilder<>::createProperty("Token")
- .withDescription("HTTP Event Collector token starting with the string Splunk. For example \'Splunk 1234578-abcd-1234-abcd-1234abcd\'")
- .isRequired(true)
- .build();
+ .withDescription("HTTP Event Collector token starting with the string Splunk. For example \'Splunk 1234578-abcd-1234-abcd-1234abcd\'")
+ .isRequired(true)
+ .build();
EXTENSIONAPI static constexpr auto SplunkRequestChannel = core::PropertyDefinitionBuilder<>::createProperty("Splunk Request Channel")
- .withDescription("Identifier of the used request channel.")
- .isRequired(true)
- .build();
- EXTENSIONAPI static constexpr auto SSLContext = core::PropertyDefinitionBuilder<0, 1, 0, 1>::createProperty("SSL Context Service")
- .withDescription("The SSL Context Service used to provide client certificate information for TLS/SSL (https) connections.")
- .isRequired(false)
- .withExclusiveOfProperties({{{"Hostname", "^http:.*$"}}})
- .withAllowedTypes({core::className<minifi::controllers::SSLContextService>()})
- .build();
+ .withDescription("Identifier of the used request channel.")
+ .isRequired(true)
+ .build();
+ EXTENSIONAPI static constexpr auto SSLContext = core::PropertyDefinitionBuilder<0, 0, 1>::createProperty("SSL Context Service")
+ .withDescription("The SSL Context Service used to provide client certificate information for TLS/SSL (https) connections.")
+ .isRequired(false)
+ .withExclusiveOfProperties({{{"Hostname", "^http:.*$"}}})
+ .withAllowedTypes<minifi::controllers::SSLContextService>()
+ .build();
EXTENSIONAPI static constexpr auto Properties = std::array<core::PropertyReference, 5>{
Hostname,
Port,
diff --git a/extensions/standard-processors/processors/GetTCP.h b/extensions/standard-processors/processors/GetTCP.h
index 5846956be..184da62df 100644
--- a/extensions/standard-processors/processors/GetTCP.h
+++ b/extensions/standard-processors/processors/GetTCP.h
@@ -69,9 +69,9 @@ class GetTCP : public core::Processor {
.withDescription("A comma delimited list of the endpoints to connect to. The format should be <server_address>:<port>.")
.isRequired(true)
.build();
- EXTENSIONAPI static constexpr auto SSLContextService = core::PropertyDefinitionBuilder<0, 1>::createProperty("SSL Context Service")
+ EXTENSIONAPI static constexpr auto SSLContextService = core::PropertyDefinitionBuilder<>::createProperty("SSL Context Service")
.withDescription("SSL Context Service Name")
- .withAllowedTypes({core::className<minifi::controllers::SSLContextService>()})
+ .withAllowedTypes<minifi::controllers::SSLContextService>()
.build();
EXTENSIONAPI static constexpr auto MessageDelimiter = core::PropertyDefinitionBuilder<>::createProperty("Message Delimiter")
.withDescription("Character that denotes the end of the message.")
diff --git a/extensions/standard-processors/processors/ListenSyslog.h b/extensions/standard-processors/processors/ListenSyslog.h
index 3f975cb2c..0f172e35e 100644
--- a/extensions/standard-processors/processors/ListenSyslog.h
+++ b/extensions/standard-processors/processors/ListenSyslog.h
@@ -75,10 +75,10 @@ class ListenSyslog : public NetworkListenerProcessor {
.withPropertyType(core::StandardPropertyTypes::UNSIGNED_LONG_TYPE)
.withDefaultValue("10000")
.build();
- EXTENSIONAPI static constexpr auto SSLContextService = core::PropertyDefinitionBuilder<0, 1>::createProperty("SSL Context Service")
+ EXTENSIONAPI static constexpr auto SSLContextService = core::PropertyDefinitionBuilder<>::createProperty("SSL Context Service")
.withDescription("The Controller Service to use in order to obtain an SSL Context. If this property is set, messages will be received over a secure connection. "
"This Property is only considered if the <Protocol> Property has a value of \"TCP\".")
- .withAllowedTypes({core::className<minifi::controllers::SSLContextService>()})
+ .withAllowedTypes<minifi::controllers::SSLContextService>()
.build();
EXTENSIONAPI static constexpr auto ClientAuth = core::PropertyDefinitionBuilder<magic_enum::enum_count<utils::net::ClientAuthOption>()>::createProperty("Client Auth")
.withDescription("The client authentication policy to use for the SSL Context. Only used if an SSL Context Service is provided.")
diff --git a/extensions/standard-processors/processors/ListenTCP.h b/extensions/standard-processors/processors/ListenTCP.h
index 9f096838a..79d13fa1b 100644
--- a/extensions/standard-processors/processors/ListenTCP.h
+++ b/extensions/standard-processors/processors/ListenTCP.h
@@ -60,9 +60,9 @@ class ListenTCP : public NetworkListenerProcessor {
.withDefaultValue("10000")
.isRequired(true)
.build();
- EXTENSIONAPI static constexpr auto SSLContextService = core::PropertyDefinitionBuilder<0, 1>::createProperty("SSL Context Service")
+ EXTENSIONAPI static constexpr auto SSLContextService = core::PropertyDefinitionBuilder<>::createProperty("SSL Context Service")
.withDescription("The Controller Service to use in order to obtain an SSL Context. If this property is set, messages will be received over a secure connection.")
- .withAllowedTypes({core::className<minifi::controllers::SSLContextService>()})
+ .withAllowedTypes<minifi::controllers::SSLContextService>()
.build();
EXTENSIONAPI static constexpr auto ClientAuth = core::PropertyDefinitionBuilder<magic_enum::enum_count<utils::net::ClientAuthOption>()>::createProperty("Client Auth")
.withDescription("The client authentication policy to use for the SSL Context. Only used if an SSL Context Service is provided.")
diff --git a/extensions/standard-processors/processors/PutFile.h b/extensions/standard-processors/processors/PutFile.h
index 26d49571b..9067a1f6c 100644
--- a/extensions/standard-processors/processors/PutFile.h
+++ b/extensions/standard-processors/processors/PutFile.h
@@ -71,7 +71,7 @@ class PutFile : public core::Processor {
.withAllowedValues({CONFLICT_RESOLUTION_STRATEGY_FAIL, CONFLICT_RESOLUTION_STRATEGY_IGNORE, CONFLICT_RESOLUTION_STRATEGY_REPLACE})
.withDefaultValue(CONFLICT_RESOLUTION_STRATEGY_FAIL)
.build();
- EXTENSIONAPI static constexpr auto CreateDirs = core::PropertyDefinitionBuilder<0, 0, 1>::createProperty("Create Missing Directories")
+ EXTENSIONAPI static constexpr auto CreateDirs = core::PropertyDefinitionBuilder<0, 1>::createProperty("Create Missing Directories")
.withDescription("If true, then missing destination directories will be created. If false, flowfiles are penalized and sent to failure.")
.withDefaultValue("true")
.isRequired(true)
diff --git a/extensions/standard-processors/processors/PutTCP.h b/extensions/standard-processors/processors/PutTCP.h
index 923f90a24..5063bae67 100644
--- a/extensions/standard-processors/processors/PutTCP.h
+++ b/extensions/standard-processors/processors/PutTCP.h
@@ -63,55 +63,55 @@ class PutTCP final : public core::Processor {
"An optional \"Connection Per FlowFile\" parameter can be specified to change the behaviour so that each FlowFiles content is transmitted over a single TCP connection "
"which is closed after the FlowFile has been sent.";
-EXTENSIONAPI static constexpr auto Hostname = core::PropertyDefinitionBuilder<>::createProperty("Hostname")
- .withDescription("The ip address or hostname of the destination.")
- .withDefaultValue("localhost")
- .isRequired(true)
- .supportsExpressionLanguage(true)
- .build();
-EXTENSIONAPI static constexpr auto Port = core::PropertyDefinitionBuilder<>::createProperty("Port")
- .withDescription("The port or service on the destination.")
- .isRequired(true)
- .supportsExpressionLanguage(true)
- .build();
-EXTENSIONAPI static constexpr auto IdleConnectionExpiration = core::PropertyDefinitionBuilder<>::createProperty("Idle Connection Expiration")
- .withDescription("The amount of time a connection should be held open without being used before closing the connection. A value of 0 seconds will disable this feature.")
- .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE)
- .withDefaultValue("15 seconds")
- .isRequired(true)
- .supportsExpressionLanguage(true)
- .build();
-EXTENSIONAPI static constexpr auto Timeout = core::PropertyDefinitionBuilder<>::createProperty("Timeout")
- .withDescription("The timeout for connecting to and communicating with the destination.")
- .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE)
- .withDefaultValue("15 seconds")
- .isRequired(true)
- .supportsExpressionLanguage(true)
- .build();
-EXTENSIONAPI static constexpr auto ConnectionPerFlowFile = core::PropertyDefinitionBuilder<>::createProperty("Connection Per FlowFile")
- .withDescription("Specifies whether to send each FlowFile's content on an individual connection.")
- .withPropertyType(core::StandardPropertyTypes::BOOLEAN_TYPE)
- .withDefaultValue("false")
- .isRequired(true)
- .supportsExpressionLanguage(false)
- .build();
-EXTENSIONAPI static constexpr auto OutgoingMessageDelimiter = core::PropertyDefinitionBuilder<>::createProperty("Outgoing Message Delimiter")
- .withDescription("Specifies the delimiter to use when sending messages out over the same TCP stream. "
- "The delimiter is appended to each FlowFile message that is transmitted over the stream so that the receiver can determine when one message ends and the next message begins. "
- "Users should ensure that the FlowFile content does not contain the delimiter character to avoid errors.")
- .isRequired(false)
- .supportsExpressionLanguage(true)
- .build();
-EXTENSIONAPI static constexpr auto SSLContextService = core::PropertyDefinitionBuilder<0, 1>::createProperty("SSL Context Service")
- .withDescription("The Controller Service to use in order to obtain an SSL Context. If this property is set, messages will be sent over a secure connection.")
- .isRequired(false)
- .withAllowedTypes({core::className<minifi::controllers::SSLContextService>()})
- .build();
-EXTENSIONAPI static constexpr auto MaxSizeOfSocketSendBuffer = core::PropertyDefinitionBuilder<>::createProperty("Max Size of Socket Send Buffer")
- .withDescription("The maximum size of the socket send buffer that should be used. This is a suggestion to the Operating System to indicate how big the socket buffer should be.")
- .isRequired(false)
- .withPropertyType(core::StandardPropertyTypes::DATA_SIZE_TYPE)
- .build();
+ EXTENSIONAPI static constexpr auto Hostname = core::PropertyDefinitionBuilder<>::createProperty("Hostname")
+ .withDescription("The ip address or hostname of the destination.")
+ .withDefaultValue("localhost")
+ .isRequired(true)
+ .supportsExpressionLanguage(true)
+ .build();
+ EXTENSIONAPI static constexpr auto Port = core::PropertyDefinitionBuilder<>::createProperty("Port")
+ .withDescription("The port or service on the destination.")
+ .isRequired(true)
+ .supportsExpressionLanguage(true)
+ .build();
+ EXTENSIONAPI static constexpr auto IdleConnectionExpiration = core::PropertyDefinitionBuilder<>::createProperty("Idle Connection Expiration")
+ .withDescription("The amount of time a connection should be held open without being used before closing the connection. A value of 0 seconds will disable this feature.")
+ .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE)
+ .withDefaultValue("15 seconds")
+ .isRequired(true)
+ .supportsExpressionLanguage(true)
+ .build();
+ EXTENSIONAPI static constexpr auto Timeout = core::PropertyDefinitionBuilder<>::createProperty("Timeout")
+ .withDescription("The timeout for connecting to and communicating with the destination.")
+ .withPropertyType(core::StandardPropertyTypes::TIME_PERIOD_TYPE)
+ .withDefaultValue("15 seconds")
+ .isRequired(true)
+ .supportsExpressionLanguage(true)
+ .build();
+ EXTENSIONAPI static constexpr auto ConnectionPerFlowFile = core::PropertyDefinitionBuilder<>::createProperty("Connection Per FlowFile")
+ .withDescription("Specifies whether to send each FlowFile's content on an individual connection.")
+ .withPropertyType(core::StandardPropertyTypes::BOOLEAN_TYPE)
+ .withDefaultValue("false")
+ .isRequired(true)
+ .supportsExpressionLanguage(false)
+ .build();
+ EXTENSIONAPI static constexpr auto OutgoingMessageDelimiter = core::PropertyDefinitionBuilder<>::createProperty("Outgoing Message Delimiter")
+ .withDescription("Specifies the delimiter to use when sending messages out over the same TCP stream. "
+ "The delimiter is appended to each FlowFile message that is transmitted over the stream so that the receiver can determine when one message ends and the next message begins. "
+ "Users should ensure that the FlowFile content does not contain the delimiter character to avoid errors.")
+ .isRequired(false)
+ .supportsExpressionLanguage(true)
+ .build();
+ EXTENSIONAPI static constexpr auto SSLContextService = core::PropertyDefinitionBuilder<>::createProperty("SSL Context Service")
+ .withDescription("The Controller Service to use in order to obtain an SSL Context. If this property is set, messages will be sent over a secure connection.")
+ .isRequired(false)
+ .withAllowedTypes<minifi::controllers::SSLContextService>()
+ .build();
+ EXTENSIONAPI static constexpr auto MaxSizeOfSocketSendBuffer = core::PropertyDefinitionBuilder<>::createProperty("Max Size of Socket Send Buffer")
+ .withDescription("The maximum size of the socket send buffer that should be used. This is a suggestion to the Operating System to indicate how big the socket buffer should be.")
+ .isRequired(false)
+ .withPropertyType(core::StandardPropertyTypes::DATA_SIZE_TYPE)
+ .build();
EXTENSIONAPI static constexpr auto Properties = std::array<core::PropertyReference, 8>{
Hostname,
Port,
diff --git a/extensions/standard-processors/processors/TailFile.h b/extensions/standard-processors/processors/TailFile.h
index 809d80fdc..015eab7d9 100644
--- a/extensions/standard-processors/processors/TailFile.h
+++ b/extensions/standard-processors/processors/TailFile.h
@@ -179,9 +179,9 @@ class TailFile : public core::Processor {
.withDefaultValue(magic_enum::enum_name(InitialStartPositions::BEGINNING_OF_FILE))
.withAllowedValues(magic_enum::enum_names<InitialStartPositions>())
.build();
- EXTENSIONAPI static constexpr auto AttributeProviderService = core::PropertyDefinitionBuilder<0, 1>::createProperty("Attribute Provider Service")
+ EXTENSIONAPI static constexpr auto AttributeProviderService = core::PropertyDefinitionBuilder<>::createProperty("Attribute Provider Service")
.withDescription("Provides a list of key-value pair records which can be used in the Base Directory property using Expression Language. Requires Multiple file mode.")
- .withAllowedTypes({core::className<minifi::controllers::AttributeProviderService>()})
+ .withAllowedTypes<minifi::controllers::AttributeProviderService>()
.build();
EXTENSIONAPI static constexpr auto BatchSize = core::PropertyDefinitionBuilder<>::createProperty("Batch Size")
.withDescription("Maximum number of flowfiles emitted in a single trigger. If set to 0 all new content will be processed.")
diff --git a/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp b/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp
index be7a60c97..deac64ba4 100644
--- a/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp
+++ b/extensions/standard-processors/tests/unit/YamlConfigurationTests.cpp
@@ -585,7 +585,7 @@ TEST_CASE("Test Dependent Property", "[YamlConfigurationDependentProperty]") {
const auto component = std::make_shared<DummyComponent>();
component->setSupportedProperties(std::array<core::PropertyReference, 2>{
core::PropertyDefinitionBuilder<>::createProperty("Prop A").withDescription("Prop A desc").withDefaultValue("val A").isRequired(true).build(),
- core::PropertyDefinitionBuilder<0, 0, 1>::createProperty("Prop B").withDescription("Prop B desc").withDefaultValue("val B").isRequired(true).withDependentProperties({ "Prop A" }).build()
+ core::PropertyDefinitionBuilder<0, 1>::createProperty("Prop B").withDescription("Prop B desc").withDefaultValue("val B").isRequired(true).withDependentProperties({ "Prop A" }).build()
});
yamlConfig.validateComponentProperties(*component, "component A", "section A");
REQUIRE(true); // Expected to get here w/o any exceptions
@@ -598,7 +598,7 @@ TEST_CASE("Test Dependent Property 2", "[YamlConfigurationDependentProperty2]")
const auto component = std::make_shared<DummyComponent>();
component->setSupportedProperties(std::array<core::PropertyReference, 2>{
core::PropertyDefinitionBuilder<>::createProperty("Prop A").withDescription("Prop A desc").isRequired(false).build(),
- core::PropertyDefinitionBuilder<0, 0, 1>::createProperty("Prop B").withDescription("Prop B desc").withDefaultValue("val B").isRequired(true).withDependentProperties({ "Prop A" }).build()
+ core::PropertyDefinitionBuilder<0, 1>::createProperty("Prop B").withDescription("Prop B desc").withDefaultValue("val B").isRequired(true).withDependentProperties({ "Prop A" }).build()
});
bool config_failed = false;
try {
@@ -619,7 +619,7 @@ TEST_CASE("Test Exclusive Property", "[YamlConfigurationExclusiveOfProperty]") {
const auto component = std::make_shared<DummyComponent>();
component->setSupportedProperties(std::array<core::PropertyReference, 2>{
core::PropertyDefinitionBuilder<>::createProperty("Prop A").withDescription("Prop A desc").withDefaultValue("val A").isRequired(true).build(),
- core::PropertyDefinitionBuilder<0, 0, 0, 1>::createProperty("Prop B").withDescription("Prop B desc").withDefaultValue("val B").isRequired(true)
+ core::PropertyDefinitionBuilder<0, 0, 1>::createProperty("Prop B").withDescription("Prop B desc").withDefaultValue("val B").isRequired(true)
.withExclusiveOfProperties({{ { "Prop A", "^abcd.*$" } }}).build()
});
yamlConfig.validateComponentProperties(*component, "component A", "section A");
@@ -633,7 +633,7 @@ TEST_CASE("Test Exclusive Property 2", "[YamlConfigurationExclusiveOfProperty2]"
const auto component = std::make_shared<DummyComponent>();
component->setSupportedProperties(std::array<core::PropertyReference, 2>{
core::PropertyDefinitionBuilder<>::createProperty("Prop A").withDescription("Prop A desc").withDefaultValue("val A").isRequired(true).build(),
- core::PropertyDefinitionBuilder<0, 0, 0, 1>::createProperty("Prop B").withDescription("Prop B desc").withDefaultValue("val B").isRequired(true)
+ core::PropertyDefinitionBuilder<0, 0, 1>::createProperty("Prop B").withDescription("Prop B desc").withDefaultValue("val B").isRequired(true)
.withExclusiveOfProperties({{ { "Prop A", "^val.*$" } }}).build()
});
bool config_failed = false;
diff --git a/libminifi/include/core/OutputAttributeDefinition.h b/libminifi/include/core/OutputAttributeDefinition.h
index 1057c72db..9e4ea9c67 100644
--- a/libminifi/include/core/OutputAttributeDefinition.h
+++ b/libminifi/include/core/OutputAttributeDefinition.h
@@ -17,9 +17,9 @@
#pragma once
#include <array>
+#include <span>
#include <string_view>
-#include "utils/gsl.h"
#include "RelationshipDefinition.h"
namespace org::apache::nifi::minifi::core {
@@ -48,7 +48,7 @@ struct OutputAttributeReference {
}
std::string_view name;
- gsl::span<const RelationshipDefinition> relationships;
+ std::span<const RelationshipDefinition> relationships;
std::string_view description;
};
diff --git a/libminifi/include/core/PropertyDefinition.h b/libminifi/include/core/PropertyDefinition.h
index 99022855d..1af98de75 100644
--- a/libminifi/include/core/PropertyDefinition.h
+++ b/libminifi/include/core/PropertyDefinition.h
@@ -18,22 +18,24 @@
#include <array>
#include <optional>
+#include <span>
#include <string_view>
#include <utility>
+#include "core/Core.h"
#include "core/PropertyType.h"
#include "utils/gsl.h"
namespace org::apache::nifi::minifi::core {
-template<size_t NumAllowedValues = 0, size_t NumAllowedTypes = 0, size_t NumDependentProperties = 0, size_t NumExclusiveOfProperties = 0>
+template<size_t NumAllowedValues = 0, size_t NumDependentProperties = 0, size_t NumExclusiveOfProperties = 0>
struct PropertyDefinition {
std::string_view name;
std::string_view display_name;
std::string_view description;
bool is_required = false;
std::array<std::string_view, NumAllowedValues> allowed_values;
- std::array<std::string_view, NumAllowedTypes> allowed_types;
+ std::span<const std::string_view> allowed_types;
std::array<std::string_view, NumDependentProperties> dependent_properties;
std::array<std::pair<std::string_view, std::string_view>, NumExclusiveOfProperties> exclusive_of_properties;
std::optional<std::string_view> default_value;
@@ -46,18 +48,18 @@ struct PropertyReference {
std::string_view display_name;
std::string_view description;
bool is_required = false;
- gsl::span<const std::string_view> allowed_values;
- gsl::span<const std::string_view> allowed_types;
- gsl::span<const std::string_view> dependent_properties;
- gsl::span<const std::pair<std::string_view, std::string_view>> exclusive_of_properties;
+ std::span<const std::string_view> allowed_values;
+ std::span<const std::string_view> allowed_types;
+ std::span<const std::string_view> dependent_properties;
+ std::span<const std::pair<std::string_view, std::string_view>> exclusive_of_properties;
std::optional<std::string_view> default_value;
gsl::not_null<const PropertyType*> type = gsl::make_not_null(&StandardPropertyTypes::VALID_TYPE);
bool supports_expression_language = false;
constexpr PropertyReference() = default;
- template<size_t NumAllowedValues = 0, size_t NumAllowedTypes = 0, size_t NumDependentProperties = 0, size_t NumExclusiveOfProperties = 0>
- constexpr PropertyReference(const PropertyDefinition<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties>& property_definition) // NOLINT: non-explicit on purpose
+ template<size_t NumAllowedValues = 0, size_t NumDependentProperties = 0, size_t NumExclusiveOfProperties = 0>
+ constexpr PropertyReference(const PropertyDefinition<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties>& property_definition) // NOLINT: non-explicit on purpose
: name{property_definition.name},
display_name{property_definition.display_name},
description{property_definition.description},
diff --git a/libminifi/include/core/PropertyDefinitionBuilder.h b/libminifi/include/core/PropertyDefinitionBuilder.h
index 801faf59e..4ec16838a 100644
--- a/libminifi/include/core/PropertyDefinitionBuilder.h
+++ b/libminifi/include/core/PropertyDefinitionBuilder.h
@@ -23,75 +23,80 @@
namespace org::apache::nifi::minifi::core {
-template <size_t NumAllowedValues = 0, size_t NumAllowedTypes = 0, size_t NumDependentProperties = 0, size_t NumExclusiveOfProperties = 0>
+namespace detail {
+template<typename... Types>
+inline constexpr auto TypeNames = std::array<std::string_view, sizeof...(Types)>{core::className<Types>()...};
+}
+
+template<size_t NumAllowedValues = 0, size_t NumDependentProperties = 0, size_t NumExclusiveOfProperties = 0>
struct PropertyDefinitionBuilder {
- static constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> createProperty(std::string_view name) {
- PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> builder;
+ static constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> createProperty(std::string_view name) {
+ PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> builder;
builder.property.name = name;
return builder;
}
- static constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> createProperty(std::string_view name, std::string_view display_name) {
- PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> builder;
+ static constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> createProperty(std::string_view name, std::string_view display_name) {
+ PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> builder;
builder.property.name = name;
builder.property.display_name = display_name;
return builder;
}
- constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> withDescription(std::string_view description) {
+ constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> withDescription(std::string_view description) {
property.description = description;
return *this;
}
- constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> isRequired(bool required) {
+ constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> isRequired(bool required) {
property.is_required = required;
return *this;
}
- constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> supportsExpressionLanguage(bool supports_expression_language) {
+ constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> supportsExpressionLanguage(bool supports_expression_language) {
property.supports_expression_language = supports_expression_language;
return *this;
}
- constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> withDefaultValue(std::string_view default_value) {
+ constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> withDefaultValue(std::string_view default_value) {
property.default_value = std::optional<std::string_view>{default_value}; // workaround for gcc 11.1; on gcc 11.3 and later, `property.default_value = default_value` works, too
return *this;
}
- constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> withAllowedValues(
+ constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> withAllowedValues(
std::array<std::string_view, NumAllowedValues> allowed_values) {
property.allowed_values = allowed_values;
return *this;
}
- constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> withAllowedTypes(
- std::array<std::string_view, NumAllowedTypes> types) {
- property.allowed_types = types;
+ template<typename... AllowedTypes>
+ constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> withAllowedTypes() {
+ property.allowed_types = {detail::TypeNames<AllowedTypes...>};
return *this;
}
- constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> withDependentProperties(
+ constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> withDependentProperties(
std::array<std::string_view, NumDependentProperties> dependent_properties) {
property.dependent_properties = dependent_properties;
return *this;
}
- constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> withExclusiveOfProperties(
+ constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> withExclusiveOfProperties(
std::array<std::pair<std::string_view, std::string_view>, NumExclusiveOfProperties> exclusive_of_properties) {
property.exclusive_of_properties = exclusive_of_properties;
return *this;
}
- constexpr PropertyDefinitionBuilder<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> withPropertyType(const PropertyType& property_type) {
+ constexpr PropertyDefinitionBuilder<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> withPropertyType(const PropertyType& property_type) {
property.type = gsl::make_not_null(&property_type);
return *this;
}
- constexpr PropertyDefinition<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> build() {
+ constexpr PropertyDefinition<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> build() {
return property;
}
- PropertyDefinition<NumAllowedValues, NumAllowedTypes, NumDependentProperties, NumExclusiveOfProperties> property{};
+ PropertyDefinition<NumAllowedValues, NumDependentProperties, NumExclusiveOfProperties> property{};
};
} // namespace org::apache::nifi::minifi::core
diff --git a/libminifi/src/core/Property.cpp b/libminifi/src/core/Property.cpp
index 017deaffb..79f5fc488 100644
--- a/libminifi/src/core/Property.cpp
+++ b/libminifi/src/core/Property.cpp
@@ -91,18 +91,18 @@ std::vector<std::pair<std::string, std::string>> Property::getExclusiveOfPropert
}
namespace {
-std::vector<PropertyValue> createPropertyValues(gsl::span<const std::string_view> values, const core::PropertyParser& property_parser) {
+std::vector<PropertyValue> createPropertyValues(std::span<const std::string_view> values, const core::PropertyParser& property_parser) {
return ranges::views::transform(values, [&property_parser](const auto& value) {
return property_parser.parse(value);
}) | ranges::to<std::vector>;
}
-inline std::vector<std::string> createStrings(gsl::span<const std::string_view> string_views) {
+inline std::vector<std::string> createStrings(std::span<const std::string_view> string_views) {
return ranges::views::transform(string_views, [](const auto& string_view) { return std::string{string_view}; })
| ranges::to<std::vector>;
}
-inline std::vector<std::pair<std::string, std::string>> createStrings(gsl::span<const std::pair<std::string_view, std::string_view>> pairs_of_string_views) {
+inline std::vector<std::pair<std::string, std::string>> createStrings(std::span<const std::pair<std::string_view, std::string_view>> pairs_of_string_views) {
return ranges::views::transform(pairs_of_string_views, [](const auto& pair_of_string_views) { return std::pair<std::string, std::string>(pair_of_string_views); })
| ranges::to<std::vector>;
}
diff --git a/libminifi/test/unit/ComponentManifestTests.cpp b/libminifi/test/unit/ComponentManifestTests.cpp
index 4f5c24737..907300750 100644
--- a/libminifi/test/unit/ComponentManifestTests.cpp
+++ b/libminifi/test/unit/ComponentManifestTests.cpp
@@ -61,10 +61,10 @@ class ExampleProcessor : public core::Processor {
using Processor::Processor;
static constexpr const char* Description = "An example processor";
- static constexpr auto ExampleProperty = core::PropertyDefinitionBuilder<0, 1>::createProperty("Example Property")
+ static constexpr auto ExampleProperty = core::PropertyDefinitionBuilder<>::createProperty("Example Property")
.withDescription("An example property")
.isRequired(false)
- .withAllowedTypes({core::className<ExampleService>()})
+ .withAllowedTypes<ExampleService>()
.build();
static constexpr auto Properties = std::array<core::PropertyReference, 1>{ExampleProperty};
static constexpr auto Relationships = std::array<core::RelationshipDefinition, 0>{};
diff --git a/libminifi/test/unit/TypeListTests.cpp b/libminifi/test/unit/TypeListTests.cpp
new file mode 100644
index 000000000..463a7a26c
--- /dev/null
+++ b/libminifi/test/unit/TypeListTests.cpp
@@ -0,0 +1,61 @@
+/**
+ * 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 "../Catch.h"
+#include "core/Core.h"
+#include "utils/meta/type_list.h"
+
+class A {};
+
+namespace outer {
+class B {};
+class C {};
+
+namespace inner {
+class C {};
+class D {};
+}
+}
+
+template<typename... Types>
+using type_list = org::apache::nifi::minifi::utils::meta::type_list<Types...>;
+
+TEST_CASE("an empty type_list doesn't contain anything") {
+ STATIC_CHECK_FALSE(type_list<>::contains<A>());
+ STATIC_CHECK_FALSE(type_list<>::contains<outer::B>());
+ STATIC_CHECK_FALSE(type_list<>::contains<outer::inner::D>());
+ STATIC_CHECK_FALSE(type_list<>::contains<std::string>());
+}
+
+TEST_CASE("a non-empty type_list contains what it should") {
+ STATIC_CHECK(type_list<A, outer::B>::contains<A>());
+ STATIC_CHECK(type_list<A, outer::B>::contains<outer::B>());
+ STATIC_CHECK_FALSE(type_list<A, outer::B>::contains<outer::C>());
+ STATIC_CHECK_FALSE(type_list<A, outer::B>::contains<outer::inner::D>());
+
+ STATIC_CHECK(type_list<int, A, std::string>::contains<int>());
+ STATIC_CHECK(type_list<int, A, std::string>::contains<A>());
+ STATIC_CHECK(type_list<int, A, std::string>::contains<std::string>());
+ STATIC_CHECK_FALSE(type_list<int, A, std::string>::contains<double>());
+ STATIC_CHECK_FALSE(type_list<int, A, std::string>::contains<outer::C>());
+
+ namespace inner = outer::inner;
+ STATIC_CHECK(type_list<outer::C>::contains<outer::C>());
+ STATIC_CHECK(type_list<inner::C>::contains<inner::C>());
+ STATIC_CHECK_FALSE(type_list<outer::C>::contains<inner::C>());
+ STATIC_CHECK_FALSE(type_list<inner::C>::contains<outer::C>());
+}
[nifi-minifi-cpp] 04/05: MINIFICPP-2175 Increase CMAKE requirements to 3.24
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 6372d67abf0fdee041fc6402a120d6b05488d96b
Author: Martin Zink <ma...@apache.org>
AuthorDate: Thu Jul 27 21:14:19 2023 +0200
MINIFICPP-2175 Increase CMAKE requirements to 3.24
Co-authored-by: Márton Szász <sz...@gmail.com>
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #1624
---
.github/workflows/ci.yml | 2 +-
CMakeLists.txt | 2 +-
README.md | 2 +-
centos.sh | 8 +-------
docker/Dockerfile | 2 +-
docker/centos/Dockerfile | 2 +-
docker/rockylinux/Dockerfile | 4 ++--
linux.sh | 30 ++++++++++++++++++++++++++++++
8 files changed, 38 insertions(+), 14 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 9eff11a07..5ba42fe6f 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -332,7 +332,7 @@ jobs:
path: /tmp/bin
docker_build:
name: "Docker build for integration tests"
- runs-on: ubuntu-20.04
+ runs-on: ubuntu-22.04
timeout-minutes: 180
steps:
- id: checkout
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 6600abd9c..17ea231da 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -17,7 +17,7 @@
# under the License.
#
-cmake_minimum_required(VERSION 3.17)
+cmake_minimum_required(VERSION 3.24)
cmake_policy(SET CMP0096 NEW) # policy to preserve the leading zeros in PROJECT_VERSION_{MAJOR,MINOR,PATCH,TWEAK}
cmake_policy(SET CMP0065 OLD) # default export policy, required for self-dlopen
if (CMAKE_VERSION VERSION_GREATER_EQUAL 3.24)
diff --git a/README.md b/README.md
index 1866219c2..060eacddf 100644
--- a/README.md
+++ b/README.md
@@ -116,7 +116,7 @@ Through JNI extensions you can run NiFi processors using NARs. The JNI extension
### To build
#### Utilities
-* CMake 3.17 or greater
+* CMake 3.24 or greater
* gcc 11 or greater
* bison 3.0.x+ (3.2 has been shown to fail builds)
* flex 2.6 or greater
diff --git a/centos.sh b/centos.sh
index 1649c775a..000a47800 100644
--- a/centos.sh
+++ b/centos.sh
@@ -57,13 +57,7 @@ install_libusb() {
bootstrap_cmake(){
- case "$OS_MAJOR" in
- 7)
- install_pkgs epel-release
- install_pkgs cmake3
- ;;
- *) install_pkgs cmake ;;
- esac
+ install_cmake_from_binary
}
bootstrap_compiler() {
diff --git a/docker/Dockerfile b/docker/Dockerfile
index 6de245ac7..e3ee1022e 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -16,7 +16,7 @@
# under the License.
#
-ARG BASE_IMAGE="alpine:3.16"
+ARG BASE_IMAGE="alpine:3.17"
# Build image
FROM ${BASE_IMAGE} AS build
diff --git a/docker/centos/Dockerfile b/docker/centos/Dockerfile
index 9ae84e426..19449ff62 100644
--- a/docker/centos/Dockerfile
+++ b/docker/centos/Dockerfile
@@ -69,5 +69,5 @@ RUN cd $MINIFI_BASE_DIR && \
source /opt/rh/devtoolset-11/enable && \
export PATH=/usr/lib64/ccache${PATH:+:${PATH}} && \
export CCACHE_DIR=${MINIFI_BASE_DIR}/.ccache && \
- cmake3 -DSTATIC_BUILD= -DSKIP_TESTS=${DOCKER_SKIP_TESTS} ${MINIFI_OPTIONS} -DAWS_ENABLE_UNITY_BUILD=OFF -DCMAKE_BUILD_TYPE="${CMAKE_BUILD_TYPE}" .. && \
+ cmake -DSTATIC_BUILD= -DSKIP_TESTS=${DOCKER_SKIP_TESTS} ${MINIFI_OPTIONS} -DAWS_ENABLE_UNITY_BUILD=OFF -DCMAKE_BUILD_TYPE="${CMAKE_BUILD_TYPE}" .. && \
make -j "$(nproc)" ${DOCKER_MAKE_TARGET}
diff --git a/docker/rockylinux/Dockerfile b/docker/rockylinux/Dockerfile
index d90112e8f..fe02f5ce8 100644
--- a/docker/rockylinux/Dockerfile
+++ b/docker/rockylinux/Dockerfile
@@ -40,7 +40,7 @@ COPY . ${MINIFI_BASE_DIR}
# Install the system dependencies needed for a build
# gpsd-devel and ccache are in EPEL
-RUN dnf -y install epel-release && dnf -y install sudo git which make libarchive ccache ca-certificates cmake perl && \
+RUN dnf -y install epel-release && dnf -y install sudo git which make libarchive ccache ca-certificates perl && \
if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_GPS=ON"; then dnf -y install gpsd-devel; fi && \
if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_JNI=ON"; then dnf -y install java-1.8.0-openjdk maven; fi && \
if echo "$MINIFI_OPTIONS" | grep -q "ENABLE_PCAP=ON"; then dnf -y install libpcap-devel; fi && \
@@ -66,5 +66,5 @@ RUN cd $MINIFI_BASE_DIR && \
source /opt/rh/gcc-toolset-12/enable && \
export PATH=/usr/lib64/ccache${PATH:+:${PATH}} && \
export CCACHE_DIR=${MINIFI_BASE_DIR}/.ccache && \
- cmake3 -DSTATIC_BUILD= -DSKIP_TESTS=${DOCKER_SKIP_TESTS} ${MINIFI_OPTIONS} -DAWS_ENABLE_UNITY_BUILD=OFF -DCMAKE_BUILD_TYPE="${CMAKE_BUILD_TYPE}" .. && \
+ cmake -DSTATIC_BUILD= -DSKIP_TESTS=${DOCKER_SKIP_TESTS} ${MINIFI_OPTIONS} -DAWS_ENABLE_UNITY_BUILD=OFF -DCMAKE_BUILD_TYPE="${CMAKE_BUILD_TYPE}" .. && \
make -j "$(nproc)" ${DOCKER_MAKE_TARGET}
diff --git a/linux.sh b/linux.sh
index c172bdd47..2d58dcf1c 100644
--- a/linux.sh
+++ b/linux.sh
@@ -19,3 +19,33 @@ verify_gcc_enable(){
#feature="$1"
[ "$COMPILER_MAJOR" -ge 11 ] && echo true || echo false
}
+
+install_cmake_from_binary() {
+ CMAKE_VERSION="3.24.4"
+ CMAKE_URL="https://cmake.org/files/v3.24/cmake-3.24.4-linux-x86_64.tar.gz"
+ EXPECTED_SHA256="cac77d28fb8668c179ac02c283b058aeb846fe2133a57d40b503711281ed9f19"
+
+ TMP_DIR=$(mktemp -d)
+
+ install_pkgs wget
+ wget -P "$TMP_DIR" "$CMAKE_URL"
+
+ ACTUAL_SHA256=$(sha256sum "$TMP_DIR/cmake-$CMAKE_VERSION-linux-x86_64.tar.gz" | cut -d " " -f 1)
+
+ if [ "$ACTUAL_SHA256" != "$EXPECTED_SHA256" ]; then
+ echo "ERROR: SHA-256 verification failed. Aborting."
+ rm -r "$TMP_DIR"
+ exit 1
+ fi
+
+ echo "Installing CMake $CMAKE_VERSION to /opt/cmake-$CMAKE_VERSION..."
+ set -x
+ tar -C "$TMP_DIR" -zxf "$TMP_DIR/cmake-$CMAKE_VERSION-linux-x86_64.tar.gz"
+ sudo mv "$TMP_DIR/cmake-$CMAKE_VERSION-linux-x86_64" /opt/cmake-$CMAKE_VERSION
+
+ sudo ln -s "/opt/cmake-$CMAKE_VERSION/bin/cmake" /usr/local/bin/cmake
+ set +x
+
+ echo "CMake $CMAKE_VERSION has been installed successfully."
+ rm -r "$TMP_DIR"
+}
[nifi-minifi-cpp] 01/05: MINIFICPP-2129 Refactor threadpool
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit deab43e4419fa85a9d15ca0fbfdf23467142bf37
Author: Martin Zink <ma...@apache.org>
AuthorDate: Thu May 25 12:12:58 2023 +0200
MINIFICPP-2129 Refactor threadpool
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #1606
---
extensions/rocksdb-repos/FlowFileLoader.cpp | 6 +-
extensions/rocksdb-repos/FlowFileLoader.h | 2 +-
libminifi/include/CronDrivenSchedulingAgent.h | 2 +-
libminifi/include/EventDrivenSchedulingAgent.h | 2 +-
libminifi/include/FlowController.h | 2 +-
libminifi/include/SchedulingAgent.h | 4 +-
libminifi/include/ThreadedSchedulingAgent.h | 2 +-
libminifi/include/TimerDrivenSchedulingAgent.h | 2 +-
libminifi/include/c2/C2Agent.h | 2 +-
libminifi/include/core/Processor.h | 1 +
libminifi/include/utils/Monitors.h | 82 ++++++---------------
libminifi/include/utils/ThreadPool.h | 98 +++++++-------------------
libminifi/src/CronDrivenSchedulingAgent.cpp | 3 +-
libminifi/src/EventDrivenSchedulingAgent.cpp | 2 +-
libminifi/src/FlowController.cpp | 2 +-
libminifi/src/ThreadedSchedulingAgent.cpp | 17 +----
libminifi/src/TimerDrivenSchedulingAgent.cpp | 2 +-
libminifi/src/c2/C2Agent.cpp | 5 +-
libminifi/src/utils/ThreadPool.cpp | 80 ++++++++-------------
libminifi/test/unit/BackTraceTests.cpp | 97 +++++++++++--------------
libminifi/test/unit/SchedulingAgentTests.cpp | 57 ++++++++-------
libminifi/test/unit/SocketTests.cpp | 46 +++++-------
libminifi/test/unit/ThreadPoolTests.cpp | 64 ++++++-----------
nanofi/include/cxx/Instance.h | 5 +-
24 files changed, 210 insertions(+), 375 deletions(-)
diff --git a/extensions/rocksdb-repos/FlowFileLoader.cpp b/extensions/rocksdb-repos/FlowFileLoader.cpp
index 983cae449..133f7ecda 100644
--- a/extensions/rocksdb-repos/FlowFileLoader.cpp
+++ b/extensions/rocksdb-repos/FlowFileLoader.cpp
@@ -41,11 +41,11 @@ FlowFileLoader::~FlowFileLoader() {
std::future<FlowFileLoader::FlowFilePtrVec> FlowFileLoader::load(std::vector<SwappedFlowFile> flow_files) {
auto promise = std::make_shared<std::promise<FlowFilePtrVec>>();
std::future<FlowFilePtrVec> future = promise->get_future();
- utils::Worker<utils::TaskRescheduleInfo> task{[this, flow_files = std::move(flow_files), promise = std::move(promise)] {
+ utils::Worker task{[this, flow_files = std::move(flow_files), promise = std::move(promise)] {
return loadImpl(flow_files, promise);
},
- "", // doesn't matter that tasks alias by name, as we never actually query their status or stop a single task
- std::make_unique<utils::ComplexMonitor>()};
+ ""}; // doesn't matter that tasks alias by name, as we never actually query their status or stop a single task
+
// the dummy_future is for the return value of the Worker's lambda, rerunning this lambda
// depends on run_determinant + result
// we could create a custom run_determinant to instead determine if/when it should be rerun
diff --git a/extensions/rocksdb-repos/FlowFileLoader.h b/extensions/rocksdb-repos/FlowFileLoader.h
index 9c7ee6d51..2287c7ddf 100644
--- a/extensions/rocksdb-repos/FlowFileLoader.h
+++ b/extensions/rocksdb-repos/FlowFileLoader.h
@@ -52,7 +52,7 @@ class FlowFileLoader {
private:
utils::TaskRescheduleInfo loadImpl(const std::vector<SwappedFlowFile>& flow_files, const std::shared_ptr<std::promise<FlowFilePtrVec>>& output);
- utils::ThreadPool<utils::TaskRescheduleInfo> thread_pool_{thread_count_, false, nullptr, "FlowFileLoaderThreadPool"};
+ utils::ThreadPool thread_pool_{thread_count_, nullptr, "FlowFileLoaderThreadPool"};
gsl::not_null<minifi::internal::RocksDatabase*> db_;
diff --git a/libminifi/include/CronDrivenSchedulingAgent.h b/libminifi/include/CronDrivenSchedulingAgent.h
index 88ee42e6b..7d422e785 100644
--- a/libminifi/include/CronDrivenSchedulingAgent.h
+++ b/libminifi/include/CronDrivenSchedulingAgent.h
@@ -41,7 +41,7 @@ class CronDrivenSchedulingAgent : public ThreadedSchedulingAgent {
std::shared_ptr<core::Repository> flow_repo,
std::shared_ptr<core::ContentRepository> content_repo,
std::shared_ptr<Configure> configuration,
- utils::ThreadPool<utils::TaskRescheduleInfo>& thread_pool)
+ utils::ThreadPool& thread_pool)
: ThreadedSchedulingAgent(controller_service_provider, std::move(repo), std::move(flow_repo), std::move(content_repo), std::move(configuration), thread_pool) {
}
diff --git a/libminifi/include/EventDrivenSchedulingAgent.h b/libminifi/include/EventDrivenSchedulingAgent.h
index 07a173059..f8664cc25 100644
--- a/libminifi/include/EventDrivenSchedulingAgent.h
+++ b/libminifi/include/EventDrivenSchedulingAgent.h
@@ -37,7 +37,7 @@ class EventDrivenSchedulingAgent : public ThreadedSchedulingAgent {
public:
EventDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo,
std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configuration,
- utils::ThreadPool<utils::TaskRescheduleInfo> &thread_pool)
+ utils::ThreadPool &thread_pool)
: ThreadedSchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configuration, thread_pool) {
using namespace std::literals::chrono_literals;
diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h
index 1d3cbf9e5..84724920f 100644
--- a/libminifi/include/FlowController.h
+++ b/libminifi/include/FlowController.h
@@ -200,7 +200,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<FlowController>::getLogger();
// Thread pool for schedulers
- utils::ThreadPool<utils::TaskRescheduleInfo> thread_pool_;
+ utils::ThreadPool thread_pool_;
};
} // namespace org::apache::nifi::minifi
diff --git a/libminifi/include/SchedulingAgent.h b/libminifi/include/SchedulingAgent.h
index fec5e9598..4911f5e90 100644
--- a/libminifi/include/SchedulingAgent.h
+++ b/libminifi/include/SchedulingAgent.h
@@ -53,7 +53,7 @@ namespace org::apache::nifi::minifi {
class SchedulingAgent {
public:
SchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo, std::shared_ptr<core::Repository> flow_repo,
- std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configuration, utils::ThreadPool<utils::TaskRescheduleInfo> &thread_pool)
+ std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configuration, utils::ThreadPool& thread_pool)
: admin_yield_duration_(),
bored_yield_duration_(0),
configure_(configuration),
@@ -122,7 +122,7 @@ class SchedulingAgent {
std::shared_ptr<core::Repository> flow_repo_;
std::shared_ptr<core::ContentRepository> content_repo_;
- utils::ThreadPool<utils::TaskRescheduleInfo> &thread_pool_;
+ utils::ThreadPool& thread_pool_;
gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider_;
private:
diff --git a/libminifi/include/ThreadedSchedulingAgent.h b/libminifi/include/ThreadedSchedulingAgent.h
index e51279154..14a703948 100644
--- a/libminifi/include/ThreadedSchedulingAgent.h
+++ b/libminifi/include/ThreadedSchedulingAgent.h
@@ -40,7 +40,7 @@ class ThreadedSchedulingAgent : public SchedulingAgent {
public:
ThreadedSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo,
std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo,
- std::shared_ptr<Configure> configuration, utils::ThreadPool<utils::TaskRescheduleInfo> &thread_pool)
+ std::shared_ptr<Configure> configuration, utils::ThreadPool &thread_pool)
: SchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configuration, thread_pool) {
}
~ThreadedSchedulingAgent() override = default;
diff --git a/libminifi/include/TimerDrivenSchedulingAgent.h b/libminifi/include/TimerDrivenSchedulingAgent.h
index 9e74fec2d..b91522f96 100644
--- a/libminifi/include/TimerDrivenSchedulingAgent.h
+++ b/libminifi/include/TimerDrivenSchedulingAgent.h
@@ -32,7 +32,7 @@ class TimerDrivenSchedulingAgent : public ThreadedSchedulingAgent {
public:
TimerDrivenSchedulingAgent(const gsl::not_null<core::controller::ControllerServiceProvider*> controller_service_provider, std::shared_ptr<core::Repository> repo,
std::shared_ptr<core::Repository> flow_repo, std::shared_ptr<core::ContentRepository> content_repo, std::shared_ptr<Configure> configure,
- utils::ThreadPool<utils::TaskRescheduleInfo> &thread_pool)
+ utils::ThreadPool &thread_pool)
: ThreadedSchedulingAgent(controller_service_provider, repo, flow_repo, content_repo, configure, thread_pool) {
}
diff --git a/libminifi/include/c2/C2Agent.h b/libminifi/include/c2/C2Agent.h
index 1c2e27b51..92d86afb3 100644
--- a/libminifi/include/c2/C2Agent.h
+++ b/libminifi/include/c2/C2Agent.h
@@ -227,7 +227,7 @@ class C2Agent : public state::UpdateController {
std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<C2Agent>::getLogger();
- utils::ThreadPool<utils::TaskRescheduleInfo> thread_pool_;
+ utils::ThreadPool thread_pool_;
std::vector<utils::Identifier> task_ids_;
diff --git a/libminifi/include/core/Processor.h b/libminifi/include/core/Processor.h
index f2266ea4b..ee3141423 100644
--- a/libminifi/include/core/Processor.h
+++ b/libminifi/include/core/Processor.h
@@ -176,6 +176,7 @@ class Processor : public Connectable, public ConfigurableComponent, public state
void clearYield();
+ std::chrono::steady_clock::time_point getYieldExpirationTime() const { return yield_expiration_; }
std::chrono::steady_clock::duration getYieldTime() const;
// Whether flow file queue full in any of the outgoing connection
bool flowFilesOutGoingFull() const;
diff --git a/libminifi/include/utils/Monitors.h b/libminifi/include/utils/Monitors.h
index aed00ce2d..10e7411c4 100644
--- a/libminifi/include/utils/Monitors.h
+++ b/libminifi/include/utils/Monitors.h
@@ -23,90 +23,50 @@
#if defined(WIN32)
#include <future> // This is required to work around a VS2017 bug, see the details below
#endif
-#include "utils/gsl.h"
namespace org::apache::nifi::minifi::utils {
-/**
- * Worker task helper that determines
- * whether or not we will run
- */
-template<typename T>
-class AfterExecute {
+class TaskRescheduleInfo {
public:
- virtual ~AfterExecute() = default;
-
- AfterExecute() = default;
- AfterExecute(AfterExecute&& /*other*/) noexcept = default;
- virtual bool isFinished(const T &result) = 0;
- virtual bool isCancelled(const T &result) = 0;
- /**
- * Time to wait before re-running this task if necessary
- * @return milliseconds since epoch after which we are eligible to re-run this task.
- */
- virtual std::chrono::steady_clock::duration wait_time() = 0;
-};
-
-/**
- * Uses the wait time for a given worker to determine if it is eligible to run
- */
+ TaskRescheduleInfo(bool result, std::chrono::steady_clock::time_point next_execution_time)
+ : next_execution_time_(next_execution_time), finished_(result) {}
-struct TaskRescheduleInfo {
- TaskRescheduleInfo(bool result, std::chrono::steady_clock::duration wait_time)
- : wait_time_(wait_time), finished_(result) {
- gsl_Expects(wait_time >= std::chrono::milliseconds(0));
+ static TaskRescheduleInfo Done() {
+ return {true, std::chrono::steady_clock::time_point::min()};
}
- std::chrono::steady_clock::duration wait_time_;
- bool finished_;
+ static TaskRescheduleInfo RetryAfter(std::chrono::steady_clock::time_point next_execution_time) {
+ return {false, next_execution_time};
+ }
- static TaskRescheduleInfo Done() {
- return {true, std::chrono::steady_clock::duration(0)};
+ static TaskRescheduleInfo RetryIn(std::chrono::steady_clock::duration duration) {
+ return {false, std::chrono::steady_clock::now()+duration};
}
- static TaskRescheduleInfo RetryIn(std::chrono::steady_clock::duration interval) {
- return {false, interval};
+ static TaskRescheduleInfo RetryImmediately() {
+ return {false, std::chrono::steady_clock::time_point::min()};
}
- static TaskRescheduleInfo RetryAfter(std::chrono::steady_clock::time_point time_point) {
- auto interval = std::max(time_point - std::chrono::steady_clock::now(), std::chrono::steady_clock::duration(0));
- return {false, interval};
+ std::chrono::steady_clock::time_point getNextExecutionTime() const {
+ return next_execution_time_;
}
- static TaskRescheduleInfo RetryImmediately() {
- return {false, std::chrono::steady_clock::duration(0)};
+ bool isFinished() const {
+ return finished_;
}
+ private:
+ std::chrono::steady_clock::time_point next_execution_time_;
+ bool finished_;
+
#if defined(WIN32)
// https://developercommunity.visualstudio.com/content/problem/60897/c-shared-state-futuresstate-default-constructs-the.html
// Because of this bug we need to have this object default constructible, which makes no sense otherwise. Hack.
private:
- TaskRescheduleInfo() : wait_time_(std::chrono::steady_clock::duration(0)), finished_(true) {}
+ TaskRescheduleInfo() : next_execution_time_(std::chrono::steady_clock::time_point::min()), finished_(true) {}
friend class std::_Associated_state<TaskRescheduleInfo>;
#endif
};
-class ComplexMonitor : public utils::AfterExecute<TaskRescheduleInfo> {
- public:
- ComplexMonitor() = default;
-
- bool isFinished(const TaskRescheduleInfo &result) override {
- if (result.finished_) {
- return true;
- }
- current_wait_.store(result.wait_time_);
- return false;
- }
- bool isCancelled(const TaskRescheduleInfo& /*result*/) override {
- return false;
- }
-
- std::chrono::steady_clock::duration wait_time() override {
- return current_wait_.load();
- }
-
- private:
- std::atomic<std::chrono::steady_clock::duration> current_wait_ {std::chrono::steady_clock::duration(0)};
-};
} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/include/utils/ThreadPool.h b/libminifi/include/utils/ThreadPool.h
index dde867909..cafcb2b94 100644
--- a/libminifi/include/utils/ThreadPool.h
+++ b/libminifi/include/utils/ThreadPool.h
@@ -50,27 +50,17 @@ using TaskId = std::string;
* purpose: Provides a wrapper for the functor
* and returns a future based on the template argument.
*/
-template<typename T>
class Worker {
public:
- explicit Worker(const std::function<T()> &task, const TaskId &identifier, std::unique_ptr<AfterExecute<T>> run_determinant)
- : identifier_(identifier),
+ explicit Worker(const std::function<TaskRescheduleInfo()> &task, TaskId identifier)
+ : identifier_(std::move(identifier)),
next_exec_time_(std::chrono::steady_clock::now()),
- task(task),
- run_determinant_(std::move(run_determinant)) {
- promise = std::make_shared<std::promise<T>>();
+ task(task) {
+ promise = std::make_shared<std::promise<TaskRescheduleInfo>>();
}
- explicit Worker(const std::function<T()> &task, const TaskId &identifier)
- : identifier_(identifier),
- next_exec_time_(std::chrono::steady_clock::now()),
- task(task),
- run_determinant_(nullptr) {
- promise = std::make_shared<std::promise<T>>();
- }
-
- explicit Worker(const TaskId& identifier = {})
- : identifier_(identifier),
+ explicit Worker(TaskId identifier = {})
+ : identifier_(std::move(identifier)),
next_exec_time_(std::chrono::steady_clock::now()) {
}
@@ -89,50 +79,40 @@ class Worker {
* true == run again
*/
virtual bool run() {
- T result = task();
- if (run_determinant_ == nullptr || (run_determinant_->isFinished(result) || run_determinant_->isCancelled(result))) {
+ TaskRescheduleInfo result = task();
+ if (result.isFinished()) {
promise->set_value(result);
return false;
}
- next_exec_time_ = std::max(next_exec_time_, std::chrono::steady_clock::now() + run_determinant_->wait_time());
- return true;
- }
- virtual void setIdentifier(const TaskId& identifier) {
- identifier_ = identifier;
+ next_exec_time_ = result.getNextExecutionTime();
+ return true;
}
- virtual std::chrono::steady_clock::time_point getNextExecutionTime() const {
+ [[nodiscard]] virtual std::chrono::steady_clock::time_point getNextExecutionTime() const {
return next_exec_time_;
}
- std::shared_ptr<std::promise<T>> getPromise() const;
+ [[nodiscard]] std::shared_ptr<std::promise<TaskRescheduleInfo>> getPromise() const { return promise; }
- const TaskId &getIdentifier() const {
+ [[nodiscard]] const TaskId &getIdentifier() const {
return identifier_;
}
protected:
TaskId identifier_;
std::chrono::steady_clock::time_point next_exec_time_;
- std::function<T()> task;
- std::unique_ptr<AfterExecute<T>> run_determinant_;
- std::shared_ptr<std::promise<T>> promise;
+ std::function<TaskRescheduleInfo()> task;
+ std::shared_ptr<std::promise<TaskRescheduleInfo>> promise;
};
-template<typename T>
class DelayedTaskComparator {
public:
- bool operator()(Worker<T> &a, Worker<T> &b) {
+ bool operator()(Worker &a, Worker &b) {
return a.getNextExecutionTime() > b.getNextExecutionTime();
}
};
-template<typename T>
-std::shared_ptr<std::promise<T>> Worker<T>::getPromise() const {
- return promise;
-}
-
class WorkerThread {
public:
explicit WorkerThread(std::thread thread, const std::string &name = "NamelessWorker")
@@ -155,16 +135,15 @@ class WorkerThread {
* ThreadPoolExecutor
* Design: Locked control over a manager thread that controls the worker threads
*/
-template<typename T>
class ThreadPool {
public:
- ThreadPool(int max_worker_threads = 2, bool daemon_threads = false,
+ ThreadPool(int max_worker_threads = 2,
core::controller::ControllerServiceProvider* controller_service_provider = nullptr, std::string name = "NamelessPool");
- ThreadPool(const ThreadPool<T> &other) = delete;
- ThreadPool<T>& operator=(const ThreadPool<T> &other) = delete;
- ThreadPool(ThreadPool<T> &&other) = delete;
- ThreadPool<T>& operator=(ThreadPool<T> &&other) = delete;
+ ThreadPool(const ThreadPool &other) = delete;
+ ThreadPool& operator=(const ThreadPool &other) = delete;
+ ThreadPool(ThreadPool &&other) = delete;
+ ThreadPool& operator=(ThreadPool &&other) = delete;
~ThreadPool() {
shutdown();
@@ -177,7 +156,7 @@ class ThreadPool {
* the worker task
* @param future future to move new promise to
*/
- void execute(Worker<T> &&task, std::future<T> &future);
+ void execute(Worker &&task, std::future<TaskRescheduleInfo> &future);
/**
* attempts to stop tasks with the provided identifier.
@@ -286,59 +265,32 @@ class ThreadPool {
}
}
-// determines if threads are detached
- bool daemon_threads_;
std::atomic<int> thread_reduction_count_;
-// max worker threads
int max_worker_threads_;
-// current worker tasks.
std::atomic<int> current_workers_;
-// thread queue
std::vector<std::shared_ptr<WorkerThread>> thread_queue_;
-// manager thread
std::thread manager_thread_;
-// the thread responsible for putting delayed tasks to the worker queue when they had to be put
std::thread delayed_scheduler_thread_;
-// conditional that's used to adjust the threads
std::atomic<bool> adjust_threads_;
-// atomic running boolean
std::atomic<bool> running_;
-// controller service provider
core::controller::ControllerServiceProvider* controller_service_provider_;
-// integrated power manager
std::shared_ptr<controllers::ThreadManagementService> thread_manager_;
- // thread queue for the recently deceased threads.
ConcurrentQueue<std::shared_ptr<WorkerThread>> deceased_thread_queue_;
-// worker queue of worker objects
- ConditionConcurrentQueue<Worker<T>> worker_queue_;
- std::priority_queue<Worker<T>, std::vector<Worker<T>>, DelayedTaskComparator<T>> delayed_worker_queue_;
-// mutex to protect task status and delayed queue
+ ConditionConcurrentQueue<Worker> worker_queue_;
+ std::priority_queue<Worker, std::vector<Worker>, DelayedTaskComparator> delayed_worker_queue_;
std::mutex worker_queue_mutex_;
-// notification for new delayed tasks that's before the current ones
std::condition_variable delayed_task_available_;
-// map to identify if a task should be
std::map<TaskId, bool> task_status_;
-// manager mutex
std::recursive_mutex manager_mutex_;
- // thread pool name
std::string name_;
- // count of running tasks by ID
std::unordered_map<TaskId, uint32_t> running_task_count_by_id_;
- // variable to signal task running completion
std::condition_variable task_run_complete_;
std::shared_ptr<core::logging::Logger> logger_;
- /**
- * Call for the manager to start worker threads
- */
- void manageWorkers();
- /**
- * Runs worker tasks
- */
+ void manageWorkers();
void run_tasks(const std::shared_ptr<WorkerThread>& thread);
-
void manage_delayed_queue();
};
diff --git a/libminifi/src/CronDrivenSchedulingAgent.cpp b/libminifi/src/CronDrivenSchedulingAgent.cpp
index fdc885747..905c309ed 100644
--- a/libminifi/src/CronDrivenSchedulingAgent.cpp
+++ b/libminifi/src/CronDrivenSchedulingAgent.cpp
@@ -32,7 +32,6 @@ utils::TaskRescheduleInfo CronDrivenSchedulingAgent::run(core::Processor* proces
using namespace std::literals::chrono_literals;
using std::chrono::ceil;
using std::chrono::seconds;
- using std::chrono::milliseconds;
using std::chrono::time_point_cast;
using std::chrono::system_clock;
@@ -58,7 +57,7 @@ utils::TaskRescheduleInfo CronDrivenSchedulingAgent::run(core::Processor* proces
last_exec_[uuid] = current_time.get_local_time();
if (processor->isYield())
- return utils::TaskRescheduleInfo::RetryIn(processor->getYieldTime());
+ return utils::TaskRescheduleInfo::RetryAfter(processor->getYieldExpirationTime());
if (auto next_trigger = schedules_.at(uuid).calculateNextTrigger(current_time.get_local_time()))
return utils::TaskRescheduleInfo::RetryIn(*next_trigger-current_time.get_local_time());
diff --git a/libminifi/src/EventDrivenSchedulingAgent.cpp b/libminifi/src/EventDrivenSchedulingAgent.cpp
index b50d2d6f3..38073fa12 100644
--- a/libminifi/src/EventDrivenSchedulingAgent.cpp
+++ b/libminifi/src/EventDrivenSchedulingAgent.cpp
@@ -43,7 +43,7 @@ utils::TaskRescheduleInfo EventDrivenSchedulingAgent::run(core::Processor* proce
while (processor->isRunning() && (std::chrono::steady_clock::now() - start_time < time_slice_)) {
this->onTrigger(processor, processContext, sessionFactory);
if (processor->isYield()) {
- return utils::TaskRescheduleInfo::RetryIn(processor->getYieldTime());
+ return utils::TaskRescheduleInfo::RetryAfter(processor->getYieldExpirationTime());
}
}
return utils::TaskRescheduleInfo::RetryImmediately(); // Let's continue work as soon as a thread is available
diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp
index 5f26a9562..451ac2d18 100644
--- a/libminifi/src/FlowController.cpp
+++ b/libminifi/src/FlowController.cpp
@@ -54,7 +54,7 @@ FlowController::FlowController(std::shared_ptr<core::Repository> provenance_repo
: core::controller::ForwardingControllerServiceProvider(core::className<FlowController>()),
running_(false),
initialized_(false),
- thread_pool_(5, false, nullptr, "Flowcontroller threadpool"),
+ thread_pool_(5, nullptr, "Flowcontroller threadpool"),
configuration_(std::move(configure)),
provenance_repo_(std::move(provenance_repo)),
flow_file_repo_(std::move(flow_file_repo)),
diff --git a/libminifi/src/ThreadedSchedulingAgent.cpp b/libminifi/src/ThreadedSchedulingAgent.cpp
index fe6cf46ad..7f4810162 100644
--- a/libminifi/src/ThreadedSchedulingAgent.cpp
+++ b/libminifi/src/ThreadedSchedulingAgent.cpp
@@ -34,17 +34,13 @@
#include "core/ProcessorNode.h"
#include "core/ProcessContext.h"
#include "core/ProcessContextBuilder.h"
-#include "core/ProcessSession.h"
#include "core/ProcessSessionFactory.h"
#include "utils/ValueParser.h"
using namespace std::literals::chrono_literals;
-namespace org {
-namespace apache {
-namespace nifi {
-namespace minifi {
+namespace org::apache::nifi::minifi {
void ThreadedSchedulingAgent::schedule(core::Processor* processor) {
std::lock_guard<std::mutex> lock(mutex_);
@@ -103,12 +99,8 @@ void ThreadedSchedulingAgent::schedule(core::Processor* processor) {
return agent->run(processor, processContext, sessionFactory);
};
- // create a functor that will be submitted to the thread pool.
- utils::Worker<utils::TaskRescheduleInfo> functor(f_ex, processor->getUUIDStr(), std::make_unique<utils::ComplexMonitor>());
- // move the functor into the thread pool. While a future is returned
- // we aren't terribly concerned with the result.
std::future<utils::TaskRescheduleInfo> future;
- thread_pool_.execute(std::move(functor), future);
+ thread_pool_.execute(utils::Worker{f_ex, processor->getUUIDStr()}, future);
}
logger_->log_debug("Scheduled thread %d concurrent workers for for process %s", processor->getMaxConcurrentTasks(), processor->getName());
processors_running_.insert(processor->getUUID());
@@ -141,7 +133,4 @@ void ThreadedSchedulingAgent::unschedule(core::Processor* processor) {
processors_running_.erase(processor->getUUID());
}
-} /* namespace minifi */
-} /* namespace nifi */
-} /* namespace apache */
-} /* namespace org */
+} // namespace org::apache::nifi::minifi
diff --git a/libminifi/src/TimerDrivenSchedulingAgent.cpp b/libminifi/src/TimerDrivenSchedulingAgent.cpp
index 113cee8f8..cbe5c2306 100644
--- a/libminifi/src/TimerDrivenSchedulingAgent.cpp
+++ b/libminifi/src/TimerDrivenSchedulingAgent.cpp
@@ -31,7 +31,7 @@ utils::TaskRescheduleInfo TimerDrivenSchedulingAgent::run(core::Processor* proce
auto trigger_start_time = std::chrono::steady_clock::now();
this->onTrigger(processor, processContext, sessionFactory);
if (processor->isYield())
- return utils::TaskRescheduleInfo::RetryIn(processor->getYieldTime());
+ return utils::TaskRescheduleInfo::RetryAfter(processor->getYieldExpirationTime());
return utils::TaskRescheduleInfo::RetryAfter(trigger_start_time + processor->getSchedulingPeriod());
}
diff --git a/libminifi/src/c2/C2Agent.cpp b/libminifi/src/c2/C2Agent.cpp
index b3b6fdef0..9891fb9de 100644
--- a/libminifi/src/c2/C2Agent.cpp
+++ b/libminifi/src/c2/C2Agent.cpp
@@ -58,7 +58,7 @@ C2Agent::C2Agent(std::shared_ptr<Configure> configuration,
configuration_(std::move(configuration)),
node_reporter_(std::move(node_reporter)),
filesystem_(std::move(filesystem)),
- thread_pool_(2, false, nullptr, "C2 threadpool"),
+ thread_pool_(2, nullptr, "C2 threadpool"),
request_restart_(std::move(request_restart)),
last_run_(std::chrono::steady_clock::now()) {
if (!configuration_->getAgentClass()) {
@@ -102,9 +102,8 @@ void C2Agent::start() {
for (const auto& function : functions_) {
utils::Identifier uuid = utils::IdGenerator::getIdGenerator()->generate();
task_ids_.push_back(uuid);
- utils::Worker<utils::TaskRescheduleInfo> functor(function, uuid.to_string(), std::make_unique<utils::ComplexMonitor>());
std::future<utils::TaskRescheduleInfo> future;
- thread_pool_.execute(std::move(functor), future);
+ thread_pool_.execute(utils::Worker{function, uuid.to_string()}, future);
}
controller_running_ = true;
thread_pool_.start();
diff --git a/libminifi/src/utils/ThreadPool.cpp b/libminifi/src/utils/ThreadPool.cpp
index 06c2eb4e1..b6343aaa9 100644
--- a/libminifi/src/utils/ThreadPool.cpp
+++ b/libminifi/src/utils/ThreadPool.cpp
@@ -16,28 +16,24 @@
*/
#include "utils/ThreadPool.h"
-#include "core/state/UpdateController.h"
using namespace std::literals::chrono_literals;
namespace org::apache::nifi::minifi::utils {
-template<typename T>
-ThreadPool<T>::ThreadPool(int max_worker_threads, bool daemon_threads, core::controller::ControllerServiceProvider* controller_service_provider, std::string name)
- : daemon_threads_(daemon_threads),
- thread_reduction_count_(0),
+ThreadPool::ThreadPool(int max_worker_threads, core::controller::ControllerServiceProvider* controller_service_provider, std::string name)
+ : thread_reduction_count_(0),
max_worker_threads_(max_worker_threads),
adjust_threads_(false),
running_(false),
controller_service_provider_(controller_service_provider),
name_(std::move(name)),
- logger_(core::logging::LoggerFactory<ThreadPool<T>>::getLogger()) {
+ logger_(core::logging::LoggerFactory<ThreadPool>::getLogger()) {
current_workers_ = 0;
thread_manager_ = nullptr;
}
-template<typename T>
-void ThreadPool<T>::run_tasks(const std::shared_ptr<WorkerThread>& thread) {
+void ThreadPool::run_tasks(const std::shared_ptr<WorkerThread>& thread) {
thread->is_running_ = true;
while (running_.load()) {
if (UNLIKELY(thread_reduction_count_ > 0)) {
@@ -50,7 +46,7 @@ void ThreadPool<T>::run_tasks(const std::shared_ptr<WorkerThread>& thread) {
}
}
- Worker<T> task;
+ Worker task;
if (worker_queue_.dequeueWait(task)) {
{
std::unique_lock<std::mutex> lock(worker_queue_mutex_);
@@ -101,15 +97,14 @@ void ThreadPool<T>::run_tasks(const std::shared_ptr<WorkerThread>& thread) {
current_workers_--;
}
-template<typename T>
-void ThreadPool<T>::manage_delayed_queue() {
+void ThreadPool::manage_delayed_queue() {
while (running_) {
std::unique_lock<std::mutex> lock(worker_queue_mutex_);
// Put the tasks ready to run in the worker queue
while (!delayed_worker_queue_.empty() && delayed_worker_queue_.top().getNextExecutionTime() <= std::chrono::steady_clock::now()) {
// I'm very sorry for this - committee must has been seriously drunk when the interface of prio queue was submitted.
- Worker<T> task = std::move(const_cast<Worker<T>&>(delayed_worker_queue_.top()));
+ Worker task = std::move(const_cast<Worker&>(delayed_worker_queue_.top()));
delayed_worker_queue_.pop();
worker_queue_.enqueue(std::move(task));
}
@@ -122,30 +117,25 @@ void ThreadPool<T>::manage_delayed_queue() {
}
}
-template<typename T>
-void ThreadPool<T>::execute(Worker<T> &&task, std::future<T> &future) {
+void ThreadPool::execute(Worker &&task, std::future<utils::TaskRescheduleInfo> &future) {
{
std::unique_lock<std::mutex> lock(worker_queue_mutex_);
task_status_[task.getIdentifier()] = true;
}
- future = std::move(task.getPromise()->get_future());
+ future = task.getPromise()->get_future();
worker_queue_.enqueue(std::move(task));
}
-template<typename T>
-void ThreadPool<T>::manageWorkers() {
- for (int i = 0; i < max_worker_threads_; i++) {
- std::stringstream thread_name;
- thread_name << name_ << " #" << i;
- auto worker_thread = std::make_shared<WorkerThread>(thread_name.str());
- worker_thread->thread_ = createThread([this, worker_thread] { run_tasks(worker_thread); });
- thread_queue_.push_back(worker_thread);
- current_workers_++;
- }
-
- if (daemon_threads_) {
- for (auto &thread : thread_queue_) {
- thread->thread_.detach();
+void ThreadPool::manageWorkers() {
+ {
+ std::unique_lock<std::mutex> lock(worker_queue_mutex_);
+ for (int i = 0; i < max_worker_threads_; i++) {
+ std::stringstream thread_name;
+ thread_name << name_ << " #" << i;
+ auto worker_thread = std::make_shared<WorkerThread>(thread_name.str());
+ worker_thread->thread_ = createThread([this, worker_thread] { run_tasks(worker_thread); });
+ thread_queue_.push_back(worker_thread);
+ current_workers_++;
}
}
@@ -171,9 +161,6 @@ void ThreadPool<T>::manageWorkers() {
std::unique_lock<std::mutex> worker_queue_lock(worker_queue_mutex_);
auto worker_thread = std::make_shared<WorkerThread>();
worker_thread->thread_ = createThread([this, worker_thread] { run_tasks(worker_thread); });
- if (daemon_threads_) {
- worker_thread->thread_.detach();
- }
thread_queue_.push_back(worker_thread);
current_workers_++;
}
@@ -195,8 +182,7 @@ void ThreadPool<T>::manageWorkers() {
}
}
-template<typename T>
-std::shared_ptr<controllers::ThreadManagementService> ThreadPool<T>::createThreadManager() const {
+std::shared_ptr<controllers::ThreadManagementService> ThreadPool::createThreadManager() const {
if (!controller_service_provider_) {
return nullptr;
}
@@ -213,8 +199,7 @@ std::shared_ptr<controllers::ThreadManagementService> ThreadPool<T>::createThrea
return thread_manager_service;
}
-template<typename T>
-void ThreadPool<T>::start() {
+void ThreadPool::start() {
std::lock_guard<std::recursive_mutex> lock(manager_mutex_);
if (!running_) {
thread_manager_ = createThreadManager();
@@ -224,22 +209,21 @@ void ThreadPool<T>::start() {
manager_thread_ = std::thread(&ThreadPool::manageWorkers, this);
std::lock_guard<std::mutex> quee_lock(worker_queue_mutex_);
- delayed_scheduler_thread_ = std::thread(&ThreadPool<T>::manage_delayed_queue, this);
+ delayed_scheduler_thread_ = std::thread(&ThreadPool::manage_delayed_queue, this);
}
}
-template<typename T>
-void ThreadPool<T>::stopTasks(const TaskId &identifier) {
+void ThreadPool::stopTasks(const TaskId &identifier) {
std::unique_lock<std::mutex> lock(worker_queue_mutex_);
task_status_[identifier] = false;
// remove tasks belonging to identifier from worker_queue_
- worker_queue_.remove([&] (const Worker<T>& worker) { return worker.getIdentifier() == identifier; });
+ worker_queue_.remove([&] (const Worker& worker) { return worker.getIdentifier() == identifier; });
// also remove from delayed_worker_queue_
decltype(delayed_worker_queue_) new_delayed_worker_queue;
while (!delayed_worker_queue_.empty()) {
- Worker<T> task = std::move(const_cast<Worker<T>&>(delayed_worker_queue_.top()));
+ Worker task = std::move(const_cast<Worker&>(delayed_worker_queue_.top()));
delayed_worker_queue_.pop();
if (task.getIdentifier() != identifier) {
new_delayed_worker_queue.push(std::move(task));
@@ -254,22 +238,19 @@ void ThreadPool<T>::stopTasks(const TaskId &identifier) {
});
}
-template<typename T>
-void ThreadPool<T>::resume() {
+void ThreadPool::resume() {
if (!worker_queue_.isRunning()) {
worker_queue_.start();
}
}
-template<typename T>
-void ThreadPool<T>::pause() {
+void ThreadPool::pause() {
if (worker_queue_.isRunning()) {
worker_queue_.stop();
}
}
-template<typename T>
-void ThreadPool<T>::shutdown() {
+void ThreadPool::shutdown() {
if (running_.load()) {
std::lock_guard<std::recursive_mutex> lock(manager_mutex_);
running_.store(false);
@@ -307,9 +288,4 @@ void ThreadPool<T>::shutdown() {
}
}
-template class utils::ThreadPool<utils::TaskRescheduleInfo>;
-template class utils::ThreadPool<int>;
-template class utils::ThreadPool<bool>;
-template class utils::ThreadPool<state::Update>;
-
} // namespace org::apache::nifi::minifi::utils
diff --git a/libminifi/test/unit/BackTraceTests.cpp b/libminifi/test/unit/BackTraceTests.cpp
index 46363638d..87840b13e 100644
--- a/libminifi/test/unit/BackTraceTests.cpp
+++ b/libminifi/test/unit/BackTraceTests.cpp
@@ -24,76 +24,61 @@
#include "utils/BackTrace.h"
#include "utils/Monitors.h"
#include "utils/ThreadPool.h"
+#include "range/v3/algorithm/any_of.hpp"
using namespace std::literals::chrono_literals;
-class WorkerNumberExecutions : public utils::AfterExecute<int> {
- public:
- explicit WorkerNumberExecutions(int tasks)
- : tasks(tasks) {
- }
-
- WorkerNumberExecutions(WorkerNumberExecutions && other) noexcept
- : runs(other.runs),
- tasks(other.tasks) {
- }
-
- bool isFinished(const int &result) override {
- return result <= 0 || ++runs >= tasks;
- }
- bool isCancelled(const int& /*result*/) override {
- return false;
- }
-
- std::chrono::steady_clock::duration wait_time() override {
- return 50ms;
- }
-
- protected:
- int runs{0};
- int tasks;
-};
-
TEST_CASE("BT1", "[TPT1]") {
const BackTrace trace = TraceResolver::getResolver().getBackTrace("BT1");
#ifdef HAS_EXECINFO
- REQUIRE(!trace.getTraces().empty());
+ CHECK(!trace.getTraces().empty());
#endif
}
-TEST_CASE("BT2", "[TPT2]") {
- std::atomic<int> counter = 0;
- utils::ThreadPool<int> pool(4);
- pool.start();
- std::this_thread::sleep_for(std::chrono::milliseconds(150));
- for (int i = 0; i < 3; i++) {
- std::unique_ptr<utils::AfterExecute<int>> after_execute = std::unique_ptr<utils::AfterExecute<int>>(new WorkerNumberExecutions(5));
- utils::Worker<int> functor([&counter]() { return ++counter; }, "id", std::move(after_execute));
+void inner_function(std::atomic_flag& ready_to_check, std::atomic_flag& checking_done) {
+ ready_to_check.test_and_set();
+ ready_to_check.notify_all();
+ checking_done.wait(false);
+}
- std::future<int> fut;
- pool.execute(std::move(functor), fut); // NOLINT(bugprone-use-after-move)
- }
+void outer_function(std::atomic_flag& ready_to_check, std::atomic_flag& checking_done) {
+ inner_function(ready_to_check, checking_done);
+}
- std::unique_ptr<utils::AfterExecute<int>> after_execute = std::unique_ptr<utils::AfterExecute<int>>(new WorkerNumberExecutions(5));
- utils::Worker<int> functor([&counter]() { return ++counter; }, "id", std::move(after_execute));
- std::future<int> fut;
- pool.execute(std::move(functor), fut); // NOLINT(bugprone-use-after-move)
+TEST_CASE("BT2", "[TPT2]") {
+ std::atomic_flag ready_for_checking;
+ std::atomic_flag done_with_checking;
- std::vector<BackTrace> traces = pool.getTraces();
- for (const auto &trace : traces) {
- std::cerr << "Thread name: " << trace.getName() << std::endl;
- const auto &trace_strings = trace.getTraces();
-#ifdef HAS_EXECINFO
- REQUIRE(trace_strings.size() > 2);
- for (const auto& trace_string : trace_strings) {
- std::cerr << " - " << trace_string << std::endl;
- }
- if (trace_strings.at(0).find("sleep_for") != std::string::npos) {
- REQUIRE(trace_strings.at(1).find("counterFunction") != std::string::npos);
+ constexpr std::string_view thread_pool_name = "Winnie the pool";
+ constexpr size_t number_of_worker_threads = 3;
+ utils::ThreadPool pool(number_of_worker_threads, nullptr, thread_pool_name.data());
+ utils::Worker worker([&]() -> utils::TaskRescheduleInfo {
+ outer_function(ready_for_checking, done_with_checking);
+ return utils::TaskRescheduleInfo::Done();
+ }, "id");
+ std::future<utils::TaskRescheduleInfo> future;
+ pool.execute(std::move(worker), future);
+
+ pool.start();
+ {
+ ready_for_checking.wait(false);
+ std::vector<BackTrace> traces = pool.getTraces();
+ CHECK(traces.size() <= number_of_worker_threads);
+ REQUIRE(!traces.empty());
+ for (const auto &trace : traces) {
+ CHECK(trace.getName().starts_with(thread_pool_name));
}
+#ifdef HAS_EXECINFO
+ auto first_worker_trace = traces.front();
+ CHECK(ranges::any_of(first_worker_trace.getTraces(), [](const std::string& trace_line) { return trace_line.find("run_tasks") != trace_line.npos; }));
+#ifdef DEBUG
+ CHECK(ranges::any_of(first_worker_trace.getTraces(), [](const std::string& trace_line) { return trace_line.find("outer_function") != trace_line.npos; }));
+ CHECK(ranges::any_of(first_worker_trace.getTraces(), [](const std::string& trace_line) { return trace_line.find("inner_function") != trace_line.npos; }));
#endif
+#endif
+ done_with_checking.test_and_set();
+ done_with_checking.notify_all();
}
- fut.wait();
+ future.wait();
}
-
diff --git a/libminifi/test/unit/SchedulingAgentTests.cpp b/libminifi/test/unit/SchedulingAgentTests.cpp
index c3cf6638a..b3ffed29d 100644
--- a/libminifi/test/unit/SchedulingAgentTests.cpp
+++ b/libminifi/test/unit/SchedulingAgentTests.cpp
@@ -63,7 +63,7 @@ TEST_CASE("SchedulingAgentTests", "[SchedulingAgent]") {
auto controller_services_ = std::make_shared<minifi::core::controller::ControllerServiceMap>();
auto configuration = std::make_shared<minifi::Configure>();
auto controller_services_provider_ = std::make_shared<minifi::core::controller::StandardControllerServiceProvider>(controller_services_, configuration);
- utils::ThreadPool<utils::TaskRescheduleInfo> thread_pool;
+ utils::ThreadPool thread_pool;
auto count_proc = std::make_shared<CountOnTriggersProcessor>("count_proc");
count_proc->incrementActiveTasks();
count_proc->setScheduledState(core::RUNNING);
@@ -79,21 +79,21 @@ TEST_CASE("SchedulingAgentTests", "[SchedulingAgent]") {
auto timer_driven_agent = std::make_shared<TimerDrivenSchedulingAgent>(gsl::make_not_null(controller_services_provider_.get()), test_repo, test_repo, content_repo, configuration, thread_pool);
timer_driven_agent->start();
auto first_task_reschedule_info = timer_driven_agent->run(count_proc.get(), context, factory);
- CHECK(!first_task_reschedule_info.finished_);
- CHECK(first_task_reschedule_info.wait_time_ <= 125ms);
+ CHECK(!first_task_reschedule_info.isFinished());
+ CHECK(first_task_reschedule_info.getNextExecutionTime() <= std::chrono::steady_clock::now() + 125ms);
CHECK(count_proc->getNumberOfTriggers() == 1);
count_proc->setOnTriggerDuration(50ms);
auto second_task_reschedule_info = timer_driven_agent->run(count_proc.get(), context, factory);
- CHECK(!second_task_reschedule_info.finished_);
- CHECK(second_task_reschedule_info.wait_time_ <= 75ms);
+ CHECK(!second_task_reschedule_info.isFinished());
+ CHECK(first_task_reschedule_info.getNextExecutionTime() <= std::chrono::steady_clock::now() + 75ms);
CHECK(count_proc->getNumberOfTriggers() == 2);
count_proc->setOnTriggerDuration(150ms);
auto third_task_reschedule_info = timer_driven_agent->run(count_proc.get(), context, factory);
- CHECK(!third_task_reschedule_info.finished_);
- CHECK(third_task_reschedule_info.wait_time_ == 0ms);
+ CHECK(!third_task_reschedule_info.isFinished());
+ CHECK(first_task_reschedule_info.getNextExecutionTime() < std::chrono::steady_clock::now());
CHECK(count_proc->getNumberOfTriggers() == 3);
}
@@ -101,33 +101,42 @@ TEST_CASE("SchedulingAgentTests", "[SchedulingAgent]") {
auto event_driven_agent = std::make_shared<EventDrivenSchedulingAgent>(gsl::make_not_null(controller_services_provider_.get()), test_repo, test_repo, content_repo, configuration, thread_pool);
event_driven_agent->start();
auto first_task_reschedule_info = event_driven_agent->run(count_proc.get(), context, factory);
- CHECK(!first_task_reschedule_info.finished_);
- CHECK(first_task_reschedule_info.wait_time_ == 0ms);
+ CHECK(!first_task_reschedule_info.isFinished());
+ CHECK(first_task_reschedule_info.getNextExecutionTime() < std::chrono::steady_clock::now());
auto count_num_after_one_schedule = count_proc->getNumberOfTriggers();
CHECK(count_num_after_one_schedule > 100);
auto second_task_reschedule_info = event_driven_agent->run(count_proc.get(), context, factory);
- CHECK(!second_task_reschedule_info.finished_);
- CHECK(second_task_reschedule_info.wait_time_ == 0ms);
+ CHECK(!second_task_reschedule_info.isFinished());
+ CHECK(second_task_reschedule_info.getNextExecutionTime() < std::chrono::steady_clock::now());
auto count_num_after_two_schedule = count_proc->getNumberOfTriggers();
CHECK(count_num_after_two_schedule > count_num_after_one_schedule+100);
}
SECTION("Cron Driven every year") {
+#ifdef WIN32
+ date::set_install(TZ_DATA_DIR);
+#endif
count_proc->setCronPeriod("0 0 0 1 1 ?");
auto cron_driven_agent = std::make_shared<CronDrivenSchedulingAgent>(gsl::make_not_null(controller_services_provider_.get()), test_repo, test_repo, content_repo, configuration, thread_pool);
cron_driven_agent->start();
auto first_task_reschedule_info = cron_driven_agent->run(count_proc.get(), context, factory);
- CHECK(!first_task_reschedule_info.finished_);
- if (first_task_reschedule_info.wait_time_ > 1min) { // To avoid possibly failing around dec 31 23:59:59
- auto next_run_time_point = std::chrono::round<std::chrono::years>(std::chrono::system_clock::now() + first_task_reschedule_info.wait_time_);
- CHECK(next_run_time_point == std::chrono::ceil<std::chrono::years>(std::chrono::system_clock::now()));
+ CHECK(!first_task_reschedule_info.isFinished());
+ if (first_task_reschedule_info.getNextExecutionTime() > std::chrono::steady_clock::now() + 1min) { // To avoid possibly failing around dec 31 23:59:59
+ auto wait_time_till_next_execution_time = std::chrono::round<std::chrono::seconds>(first_task_reschedule_info.getNextExecutionTime() - std::chrono::steady_clock::now());
+
+ auto current_time = date::make_zoned<std::chrono::seconds>(date::current_zone(), std::chrono::time_point_cast<std::chrono::seconds>(std::chrono::system_clock::now()));
+ auto current_year_month_day = date::year_month_day(date::floor<date::days>(current_time.get_local_time()));
+ auto new_years_day = date::make_zoned<std::chrono::seconds>(date::current_zone(), date::local_days{date::year{current_year_month_day.year()+date::years(1)}/date::January/1});
+
+ auto time_until_new_years_day = new_years_day.get_local_time() - current_time.get_local_time();
+
+ CHECK(std::chrono::round<std::chrono::minutes>(time_until_new_years_day - wait_time_till_next_execution_time) == 0min);
CHECK(count_proc->getNumberOfTriggers() == 0);
auto second_task_reschedule_info = cron_driven_agent->run(count_proc.get(), context, factory);
- CHECK(!second_task_reschedule_info.finished_);
- next_run_time_point = std::chrono::round<std::chrono::years>(std::chrono::system_clock::now() + first_task_reschedule_info.wait_time_);
- CHECK(next_run_time_point == std::chrono::ceil<std::chrono::years>(std::chrono::system_clock::now()));
+ CHECK(!second_task_reschedule_info.isFinished());
+ CHECK(std::chrono::round<std::chrono::minutes>(first_task_reschedule_info.getNextExecutionTime() - second_task_reschedule_info.getNextExecutionTime()) == 0min);
CHECK(count_proc->getNumberOfTriggers() == 0);
}
}
@@ -137,14 +146,14 @@ TEST_CASE("SchedulingAgentTests", "[SchedulingAgent]") {
auto cron_driven_agent = std::make_shared<CronDrivenSchedulingAgent>(gsl::make_not_null(controller_services_provider_.get()), test_repo, test_repo, content_repo, configuration, thread_pool);
cron_driven_agent->start();
auto first_task_reschedule_info = cron_driven_agent->run(count_proc.get(), context, factory);
- CHECK(!first_task_reschedule_info.finished_);
- CHECK(first_task_reschedule_info.wait_time_ <= 1s);
+ CHECK(!first_task_reschedule_info.isFinished());
+ CHECK(first_task_reschedule_info.getNextExecutionTime() <= std::chrono::steady_clock::now() + 1s);
CHECK(count_proc->getNumberOfTriggers() == 0);
- std::this_thread::sleep_for(first_task_reschedule_info.wait_time_ + 1ms);
+ std::this_thread::sleep_until(first_task_reschedule_info.getNextExecutionTime());
auto second_task_reschedule_info = cron_driven_agent->run(count_proc.get(), context, factory);
- CHECK(!second_task_reschedule_info.finished_);
- CHECK(second_task_reschedule_info.wait_time_ <= 1s);
+ CHECK(!second_task_reschedule_info.isFinished());
+ CHECK(second_task_reschedule_info.getNextExecutionTime() <= std::chrono::steady_clock::now() + 1s);
CHECK(count_proc->getNumberOfTriggers() == 1);
}
@@ -153,7 +162,7 @@ TEST_CASE("SchedulingAgentTests", "[SchedulingAgent]") {
auto cron_driven_agent = std::make_shared<CronDrivenSchedulingAgent>(gsl::make_not_null(controller_services_provider_.get()), test_repo, test_repo, content_repo, configuration, thread_pool);
cron_driven_agent->start();
auto first_task_reschedule_info = cron_driven_agent->run(count_proc.get(), context, factory);
- CHECK(first_task_reschedule_info.finished_);
+ CHECK(first_task_reschedule_info.isFinished());
}
}
} // namespace org::apache::nifi::minifi::testing
diff --git a/libminifi/test/unit/SocketTests.cpp b/libminifi/test/unit/SocketTests.cpp
index 5c23e236c..a8d91f44e 100644
--- a/libminifi/test/unit/SocketTests.cpp
+++ b/libminifi/test/unit/SocketTests.cpp
@@ -27,9 +27,13 @@
#include "../Catch.h"
#include "io/StreamFactory.h"
#include "io/Sockets.h"
-#include "utils/ThreadPool.h"
#include "properties/Configuration.h"
+#include <asio/thread_pool.hpp>
+#include <asio/awaitable.hpp>
+#include <asio/co_spawn.hpp>
+#include <asio/use_future.hpp>
+
namespace minifi = org::apache::nifi::minifi;
namespace io = minifi::io;
using io::Socket;
@@ -163,7 +167,7 @@ TEST_CASE("TestSocketWriteTestAfterClose", "[TestSocket7]") {
#ifdef OPENSSL_SUPPORT
std::atomic<uint8_t> counter;
std::mt19937_64 seed { std::random_device { }() };
-bool createSocket() {
+asio::awaitable<bool> createSocket() {
counter++;
std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::Configure>();
@@ -175,57 +179,43 @@ bool createSocket() {
socketA->initialize();
}
- return true;
+ co_return true;
}
-/**
- * MINIFI-320 was created to address reallocations within TLSContext
- * This test will create 20 threads that attempt to create contexts
- * to ensure we no longer see the segfaults.
- */
+
TEST_CASE("TestTLSContextCreation", "[TestSocket8]") {
- utils::ThreadPool<bool> pool(20, true);
+ constexpr size_t number_of_threads = 20;
+ asio::thread_pool pool(number_of_threads);
std::vector<std::future<bool>> futures;
- for (int i = 0; i < 20; i++) {
- std::function<bool()> f_ex = createSocket;
- utils::Worker<bool> functor(f_ex, "id");
- std::future<bool> fut;
- pool.execute(std::move(functor), fut); // NOLINT(bugprone-use-after-move)
- futures.push_back(std::move(fut));
+ futures.reserve(number_of_threads);
+ for (size_t i = 0; i < number_of_threads; i++) {
+ futures.push_back(asio::co_spawn(pool, createSocket(), asio::use_future));
}
- pool.start();
+ pool.join();
for (auto &&future : futures) {
- future.wait();
+ CHECK(future.valid());
}
- REQUIRE(20 == counter.load());
+ REQUIRE(number_of_threads == counter.load());
}
-/**
- * MINIFI-329 was created in regards to an option existing but not
- * being properly evaluated.
- */
TEST_CASE("TestTLSContextCreation2", "[TestSocket9]") {
std::shared_ptr<minifi::Configure> configure = std::make_shared<minifi::Configure>();
configure->set(minifi::Configuration::nifi_remote_input_secure, "false");
auto factory = io::StreamFactory::getInstance(configure);
std::string host = Socket::getMyHostName();
Socket *socket = factory->createSocket(host, 10001).release();
- io::TLSSocket *tls = dynamic_cast<io::TLSSocket*>(socket);
+ auto *tls = dynamic_cast<io::TLSSocket*>(socket);
REQUIRE(tls == nullptr);
}
-/**
- * MINIFI-329 was created in regards to an option existing but not
- * being properly evaluated.
- */
TEST_CASE("TestTLSContextCreationNullptr", "[TestSocket10]") {
std::shared_ptr<minifi::Configure> configure = std::make_shared<minifi::Configure>();
configure->set(minifi::Configuration::nifi_remote_input_secure, "false");
auto factory = io::StreamFactory::getInstance(configure);
std::string host = Socket::getMyHostName();
io::Socket *socket = factory->createSecureSocket(host, 10001, nullptr).release();
- io::TLSSocket *tls = dynamic_cast<minifi::io::TLSSocket*>(socket);
+ auto *tls = dynamic_cast<minifi::io::TLSSocket*>(socket);
REQUIRE(tls == nullptr);
}
#endif // OPENSSL_SUPPORT
diff --git a/libminifi/test/unit/ThreadPoolTests.cpp b/libminifi/test/unit/ThreadPoolTests.cpp
index 948fc73e5..cbf0547b5 100644
--- a/libminifi/test/unit/ThreadPoolTests.cpp
+++ b/libminifi/test/unit/ThreadPoolTests.cpp
@@ -25,67 +25,43 @@
using namespace std::literals::chrono_literals;
-class WorkerNumberExecutions : public utils::AfterExecute<int> {
- public:
- explicit WorkerNumberExecutions(int tasks)
- : tasks(tasks) {
- }
-
- explicit WorkerNumberExecutions(WorkerNumberExecutions && other) noexcept
- : runs(other.runs),
- tasks(other.tasks) {
- }
-
- bool isFinished(const int &result) override {
- return result <= 0 || ++runs >= tasks;
- }
- bool isCancelled(const int& /*result*/) override {
- return false;
- }
-
- std::chrono::steady_clock::duration wait_time() override {
- return 50ms;
- }
-
- protected:
- int runs{0};
- int tasks;
-};
-
TEST_CASE("ThreadPoolTest1", "[TPT1]") {
- utils::ThreadPool<bool> pool(5);
- utils::Worker<bool> functor([](){ return true; }, "id");
+ utils::ThreadPool pool(5);
+ utils::Worker functor([](){ return utils::TaskRescheduleInfo::Done(); }, "id");
pool.start();
- std::future<bool> fut;
+ std::future<utils::TaskRescheduleInfo> fut;
pool.execute(std::move(functor), fut); // NOLINT(bugprone-use-after-move)
- fut.wait();
- REQUIRE(true == fut.get());
+ CHECK(fut.get().isFinished());
}
TEST_CASE("ThreadPoolTest2", "[TPT2]") {
- std::atomic<int> counter = 0;
- utils::ThreadPool<int> pool(5);
- std::unique_ptr<utils::AfterExecute<int>> after_execute = std::unique_ptr<utils::AfterExecute<int>>(new WorkerNumberExecutions(20));
- utils::Worker<int> functor([&counter]() { return ++counter; }, "id", std::move(after_execute));
+ constexpr size_t max_counter = 20;
+ std::atomic<size_t> counter = 0;
+ utils::ThreadPool pool(5);
+ utils::Worker functor([&counter](){
+ if (++counter == max_counter)
+ return utils::TaskRescheduleInfo::Done();
+
+ return utils::TaskRescheduleInfo::RetryImmediately();}, "id");
pool.start();
- std::future<int> fut;
- pool.execute(std::move(functor), fut);
- fut.wait();
- REQUIRE(20 == fut.get());
+ std::future<utils::TaskRescheduleInfo> fut;
+ pool.execute(std::move(functor), fut); // NOLINT(bugprone-use-after-move)
+ CHECK(fut.get().isFinished());
+ REQUIRE(20 == counter);
}
TEST_CASE("Worker wait time should be relative to the last run") {
std::vector<std::chrono::steady_clock::time_point> worker_execution_time_points;
- utils::ThreadPool<utils::TaskRescheduleInfo> pool(1);
+ utils::ThreadPool pool(1);
auto wait_time_between_tasks = 10ms;
- utils::Worker<utils::TaskRescheduleInfo> worker([&]()->utils::TaskRescheduleInfo {
+ utils::Worker worker([&]()->utils::TaskRescheduleInfo {
worker_execution_time_points.push_back(std::chrono::steady_clock::now());
if (worker_execution_time_points.size() == 2) {
return utils::TaskRescheduleInfo::Done();
} else {
return utils::TaskRescheduleInfo::RetryIn(wait_time_between_tasks);
}
- }, "id", std::make_unique<utils::ComplexMonitor>());
+ }, "id");
std::this_thread::sleep_for(wait_time_between_tasks + 1ms); // Pre-waiting should not matter
std::future<utils::TaskRescheduleInfo> task_future;
@@ -94,7 +70,7 @@ TEST_CASE("Worker wait time should be relative to the last run") {
auto final_task_reschedule_info = task_future.get();
- CHECK(final_task_reschedule_info.finished_);
+ CHECK(final_task_reschedule_info.isFinished());
REQUIRE(worker_execution_time_points.size() == 2);
CHECK(worker_execution_time_points[1] - worker_execution_time_points[0] >= wait_time_between_tasks);
}
diff --git a/nanofi/include/cxx/Instance.h b/nanofi/include/cxx/Instance.h
index 427eed117..c17130371 100644
--- a/nanofi/include/cxx/Instance.h
+++ b/nanofi/include/cxx/Instance.h
@@ -148,9 +148,8 @@ class Instance {
// run all functions independently
for (auto function : functions) {
- utils::Worker<utils::TaskRescheduleInfo> functor(function, "listeners");
std::future<utils::TaskRescheduleInfo> future;
- listener_thread_pool_.execute(std::move(functor), future);
+ listener_thread_pool_.execute(utils::Worker{function, "listeners"}, future);
}
}
@@ -169,7 +168,7 @@ class Instance {
std::string url_;
std::shared_ptr<Configure> configure_;
- utils::ThreadPool<utils::TaskRescheduleInfo> listener_thread_pool_;
+ utils::ThreadPool listener_thread_pool_;
};
} // namespace org::apache::nifi::minifi
[nifi-minifi-cpp] 03/05: MINIFICPP-2116 - Support GCC13
Posted by lo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lordgamez pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 4307333d14cde970b2f2dac1525b3ae21e4e0f27
Author: Adam Debreceni <ad...@apache.org>
AuthorDate: Wed Jul 26 07:53:45 2023 +0200
MINIFICPP-2116 - Support GCC13
Signed-off-by: Gabor Gyimesi <ga...@gmail.com>
This closes #1623
---
.github/workflows/gcc13-compat.yml | 45 ++++++++++++++++++++++
cmake/Abseil.cmake | 4 +-
cmake/BundledAzureSdkCpp.cmake | 16 +++++---
cmake/GoogleCloudCpp.cmake | 2 +
cmake/KubernetesClientC.cmake | 3 +-
cmake/curl/dummy/FindCURL.cmake | 1 +
extensions/azure/storage/BlobStorageClient.h | 2 +-
extensions/azure/storage/DataLakeStorageClient.h | 2 +-
extensions/http-curl/tests/HTTPHandlers.h | 2 +-
extensions/sftp/client/SFTPClient.cpp | 5 +--
extensions/sftp/processors/ListSFTP.cpp | 4 +-
libminifi/include/core/Resource.h | 4 +-
libminifi/src/utils/TimeUtil.cpp | 2 +-
libminifi/test/TestBase.cpp | 3 +-
.../azure-sdk-cpp-remove-samples.patch | 23 -----------
thirdparty/azure-sdk-cpp/gcc13.patch | 44 +++++++++++++++++++++
thirdparty/azure-sdk-cpp/remove-amqp.patch | 15 ++++++++
17 files changed, 133 insertions(+), 44 deletions(-)
diff --git a/.github/workflows/gcc13-compat.yml b/.github/workflows/gcc13-compat.yml
new file mode 100644
index 000000000..0c56d2b00
--- /dev/null
+++ b/.github/workflows/gcc13-compat.yml
@@ -0,0 +1,45 @@
+name: "MiNiFi-CPP verify gcc13 compatibility"
+on: [workflow_dispatch]
+jobs:
+ ubuntu_22_04:
+ name: "ubuntu-22.04-gcc-13"
+ runs-on: ubuntu-22.04
+ timeout-minutes: 120
+ steps:
+ - id: checkout
+ uses: actions/checkout@v3
+ - id: install_deps
+ run: |
+ sudo add-apt-repository ppa:ubuntu-toolchain-r/test
+ sudo apt update
+ sudo apt install -y ccache libfl-dev libpcap-dev libboost-all-dev openjdk-8-jdk maven libusb-1.0-0-dev libpng-dev libgps-dev libsqliteodbc lua5.3 liblua5.3-dev flake8 parallel gcc-13
+ echo "PATH=/usr/lib/ccache:$PATH" >> $GITHUB_ENV
+ echo -e "127.0.0.1\t$HOSTNAME" | sudo tee -a /etc/hosts > /dev/null
+ - name: build
+ run: |
+ export CC=gcc-13
+ export CXX=g++-13
+ ./bootstrap.sh -e -t
+ cd build
+ cmake -DUSE_SHARED_LIBS=ON -DCMAKE_BUILD_TYPE=Release -DCI_BUILD=ON -DSTRICT_GSL_CHECKS=AUDIT -DFAIL_ON_WARNINGS=ON -DENABLE_ALL=ON ..
+ make -j$(nproc) VERBOSE=1
+ - name: test
+ id: test
+ run: |
+ # Set core file size limit to 1GiB
+ ulimit -c 1048576
+ cd build && make test ARGS="--timeout 300 -j8 --output-on-failure"
+ - name: check-cores
+ if: ${{ failure() && steps.test.conclusion == 'failure' }}
+ run: |
+ if [ "$(ls -A /var/lib/apport/coredump/)" ]; then echo "CORES_EXIST=true" >> $GITHUB_ENV; fi
+ - uses: actions/upload-artifact@v3.1.2
+ if: ${{ failure() && env.CORES_EXIST == 'true' }}
+ with:
+ name: ubuntu-coredumps
+ path: /var/lib/apport/coredump/
+ - uses: actions/upload-artifact@v3.1.2
+ if: ${{ failure() && env.CORES_EXIST == 'true' }}
+ with:
+ name: ubuntu-binaries
+ path: build/bin
diff --git a/cmake/Abseil.cmake b/cmake/Abseil.cmake
index 48bb01712..1afc06757 100644
--- a/cmake/Abseil.cmake
+++ b/cmake/Abseil.cmake
@@ -22,7 +22,7 @@ set(ABSL_ENABLE_INSTALL ON CACHE INTERNAL "")
set(BUILD_TESTING OFF CACHE STRING "" FORCE)
FetchContent_Declare(
absl
- URL https://github.com/abseil/abseil-cpp/archive/refs/tags/20211102.0.tar.gz
- URL_HASH SHA256=dcf71b9cba8dc0ca9940c4b316a0c796be8fab42b070bb6b7cab62b48f0e66c4
+ URL https://github.com/abseil/abseil-cpp/archive/refs/tags/20220623.1.tar.gz
+ URL_HASH SHA256=91ac87d30cc6d79f9ab974c51874a704de9c2647c40f6932597329a282217ba8
)
FetchContent_MakeAvailable(absl)
diff --git a/cmake/BundledAzureSdkCpp.cmake b/cmake/BundledAzureSdkCpp.cmake
index 847b2c78a..3f1459871 100644
--- a/cmake/BundledAzureSdkCpp.cmake
+++ b/cmake/BundledAzureSdkCpp.cmake
@@ -16,9 +16,13 @@
# under the License.
function(use_bundled_libazure SOURCE_DIR BINARY_DIR)
- set(PATCH_FILE "${SOURCE_DIR}/thirdparty/azure-sdk-cpp/azure-sdk-cpp-remove-samples.patch")
+ set(PATCH_FILE1 "${SOURCE_DIR}/thirdparty/azure-sdk-cpp/remove-amqp.patch")
+ set(PATCH_FILE2 "${SOURCE_DIR}/thirdparty/azure-sdk-cpp/gcc13.patch")
set(PC ${Bash_EXECUTABLE} -c "set -x && \
- (\"${Patch_EXECUTABLE}\" -p1 -R -s -f --dry-run -i \"${PATCH_FILE}\" || \"${Patch_EXECUTABLE}\" -p1 -N -i \"${PATCH_FILE}\")")
+ (\"${Patch_EXECUTABLE}\" -p1 -N -i \"${PATCH_FILE}\")")
+ set(PC ${Bash_EXECUTABLE} -c "set -x &&\
+ (\"${Patch_EXECUTABLE}\" -p1 -N -i \"${PATCH_FILE1}\") &&\
+ (\"${Patch_EXECUTABLE}\" -p1 -N -i \"${PATCH_FILE2}\") ")
# Define byproducts
set(INSTALL_DIR "${BINARY_DIR}/thirdparty/azure-sdk-cpp-install")
if (WIN32)
@@ -55,14 +59,16 @@ function(use_bundled_libazure SOURCE_DIR BINARY_DIR)
set(AZURE_SDK_CMAKE_ARGS ${PASSTHROUGH_CMAKE_ARGS}
-DWARNINGS_AS_ERRORS=OFF
- -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR})
+ -DCMAKE_INSTALL_PREFIX=${INSTALL_DIR}
+ -DDISABLE_AZURE_CORE_OPENTELEMETRY=ON
+ -DBUILD_TRANSPORT_CURL=ON)
append_third_party_passthrough_args(AZURE_SDK_CMAKE_ARGS "${AZURE_SDK_CMAKE_ARGS}")
# Build project
ExternalProject_Add(
asdkext # short for azure-sdk-cpp-external due to windows MAX_PATH limitations
- URL https://github.com/Azure/azure-sdk-for-cpp/archive/refs/tags/azure-storage-files-datalake_12.2.0.tar.gz
- URL_HASH "SHA256=d4e80ea5e786dc689ddd04825d97ab91f5e1ef2787fa88a3d5ee00f0b820433f"
+ URL https://github.com/Azure/azure-sdk-for-cpp/archive/refs/tags/azure-storage-files-datalake_12.7.0.tar.gz
+ URL_HASH "SHA256=42b9c1df0c15d9e0a3f547eeba8c45708aa86819986da66234e267f4ad728112"
SOURCE_DIR "${BINARY_DIR}/thirdparty/azure-sdk-cpp-src"
INSTALL_DIR "${BINARY_DIR}/thirdparty/azure-sdk-cpp-install"
BUILD_BYPRODUCTS "${AZURESDK_LIBRARIES_LIST}"
diff --git a/cmake/GoogleCloudCpp.cmake b/cmake/GoogleCloudCpp.cmake
index aca4ac3e0..4612959b4 100644
--- a/cmake/GoogleCloudCpp.cmake
+++ b/cmake/GoogleCloudCpp.cmake
@@ -65,4 +65,6 @@ if (WIN32)
target_compile_options(google_cloud_cpp_storage PUBLIC /wd4996)
else()
target_compile_options(google_cloud_cpp_storage PUBLIC -Wno-error=deprecated-declarations)
+ target_compile_options(google_cloud_cpp_rest_internal PUBLIC -Wno-error)
endif()
+
diff --git a/cmake/KubernetesClientC.cmake b/cmake/KubernetesClientC.cmake
index 61878d673..155e7b4ff 100644
--- a/cmake/KubernetesClientC.cmake
+++ b/cmake/KubernetesClientC.cmake
@@ -33,12 +33,13 @@ set(LWS_WITH_SHARED OFF CACHE BOOL "" FORCE)
set(LWS_OPENSSL_INCLUDE_DIRS "${OPENSSL_INCLUDE_DIR}" CACHE STRING "" FORCE)
set(LWS_OPENSSL_LIBRARIES "${OPENSSL_LIBRARIES}" CACHE STRING "" FORCE)
set(CMAKE_C_FLAGS "-fpic" CACHE STRING "" FORCE)
+set(DISABLE_WERROR ON CACHE STRING "" FORCE)
set(WEBSOCKETS_PATCH_FILE_1 "${CMAKE_SOURCE_DIR}/thirdparty/libwebsockets/fix-include-dirs.patch")
set(WEBSOCKETS_PATCH_FILE_2 "${CMAKE_SOURCE_DIR}/thirdparty/libwebsockets/openssl3.patch")
set(WEBSOCKETS_PC ${Bash_EXECUTABLE} -c "set -x &&\
(${Patch_EXECUTABLE} -R -p1 -s -f --dry-run -i ${WEBSOCKETS_PATCH_FILE_1} || ${Patch_EXECUTABLE} -p1 -i ${WEBSOCKETS_PATCH_FILE_1}) &&\
- (${Patch_EXECUTABLE} -R -p1 -s -f --dry-run -i ${WEBSOCKETS_PATCH_FILE_2} || ${Patch_EXECUTABLE} -p1 -i ${WEBSOCKETS_PATCH_FILE_2}) ")
+ (${Patch_EXECUTABLE} -R -p1 -s -f --dry-run -i ${WEBSOCKETS_PATCH_FILE_2} || ${Patch_EXECUTABLE} -p1 -i ${WEBSOCKETS_PATCH_FILE_2})")
FetchContent_Declare(websockets
URL https://github.com/warmcat/libwebsockets/archive/refs/tags/v4.3.2.tar.gz
URL_HASH SHA256=6a85a1bccf25acc7e8e5383e4934c9b32a102880d1e4c37c70b27ae2a42406e1
diff --git a/cmake/curl/dummy/FindCURL.cmake b/cmake/curl/dummy/FindCURL.cmake
index 3755b2bd9..77347bac2 100644
--- a/cmake/curl/dummy/FindCURL.cmake
+++ b/cmake/curl/dummy/FindCURL.cmake
@@ -27,4 +27,5 @@ if(NOT TARGET CURL::libcurl)
add_library(CURL::libcurl STATIC IMPORTED)
set_target_properties(CURL::libcurl PROPERTIES IMPORTED_LOCATION "${CURL_LIBRARIES}")
set_property(TARGET CURL::libcurl APPEND PROPERTY INTERFACE_INCLUDE_DIRECTORIES "${CURL_INCLUDE_DIRS}")
+ target_compile_definitions(CURL::libcurl INTERFACE CURL_STATICLIB)
endif()
diff --git a/extensions/azure/storage/BlobStorageClient.h b/extensions/azure/storage/BlobStorageClient.h
index a7d644365..4f28e1acc 100644
--- a/extensions/azure/storage/BlobStorageClient.h
+++ b/extensions/azure/storage/BlobStorageClient.h
@@ -25,7 +25,7 @@
#include <vector>
#include <memory>
-#include "azure/storage/blobs/protocol/blob_rest_client.hpp"
+#include "azure/storage/blobs/blob_responses.hpp"
#include "AzureStorageCredentials.h"
#include "utils/gsl.h"
#include "utils/Enum.h"
diff --git a/extensions/azure/storage/DataLakeStorageClient.h b/extensions/azure/storage/DataLakeStorageClient.h
index 4fd653daf..839302da7 100644
--- a/extensions/azure/storage/DataLakeStorageClient.h
+++ b/extensions/azure/storage/DataLakeStorageClient.h
@@ -28,7 +28,7 @@
#include "utils/gsl.h"
#include "io/InputStream.h"
-#include "azure/storage/files/datalake/protocol/datalake_rest_client.hpp"
+#include "azure/storage/files/datalake/datalake_responses.hpp"
#include "utils/Enum.h"
#include "utils/RegexUtils.h"
diff --git a/extensions/http-curl/tests/HTTPHandlers.h b/extensions/http-curl/tests/HTTPHandlers.h
index ef6a2ca0a..7ab6a5649 100644
--- a/extensions/http-curl/tests/HTTPHandlers.h
+++ b/extensions/http-curl/tests/HTTPHandlers.h
@@ -249,7 +249,7 @@ class FlowFileResponder : public ServerAwareHandler {
}
flow->attributes[name] = value;
}
- uint64_t length;
+ uint64_t length{};
{
const auto read = stream.read(length);
if (!isServerRunning()) return false;
diff --git a/extensions/sftp/client/SFTPClient.cpp b/extensions/sftp/client/SFTPClient.cpp
index e2167bfca..71365c00d 100644
--- a/extensions/sftp/client/SFTPClient.cpp
+++ b/extensions/sftp/client/SFTPClient.cpp
@@ -279,13 +279,10 @@ bool SFTPClient::connect() {
/* Getting socket from curl */
#ifdef WIN32
curl_socket_t sockfd;
- /* Only CURLINFO_ACTIVESOCKET works on Win64 */
- curl_res = curl_easy_getinfo(easy_, CURLINFO_ACTIVESOCKET, &sockfd);
#else
long sockfd; // NOLINT(runtime/int) long due to libcurl API
- /* Some older cURL versions only support CURLINFO_LASTSOCKET */
- curl_res = curl_easy_getinfo(easy_, CURLINFO_LASTSOCKET, &sockfd);
#endif
+ curl_res = curl_easy_getinfo(easy_, CURLINFO_ACTIVESOCKET, &sockfd);
if (curl_res != CURLE_OK) {
return false;
}
diff --git a/extensions/sftp/processors/ListSFTP.cpp b/extensions/sftp/processors/ListSFTP.cpp
index 42490b260..61f823a07 100644
--- a/extensions/sftp/processors/ListSFTP.cpp
+++ b/extensions/sftp/processors/ListSFTP.cpp
@@ -535,9 +535,9 @@ void ListSFTP::listByTrackingTimestamps(
/* Determine the minimum reliable timestamp based on precision */
auto minimum_reliable_timestamp = now - listing_lag;
if (remote_system_timestamp_precision == TARGET_SYSTEM_TIMESTAMP_PRECISION_SECONDS) {
- std::chrono::floor<std::chrono::seconds>(minimum_reliable_timestamp);
+ minimum_reliable_timestamp = std::chrono::floor<std::chrono::seconds>(minimum_reliable_timestamp);
} else {
- std::chrono::floor<std::chrono::minutes>(minimum_reliable_timestamp);
+ minimum_reliable_timestamp = std::chrono::floor<std::chrono::minutes>(minimum_reliable_timestamp);
}
/* If the latest timestamp is not old enough, we wait another cycle */
if (latest_listed_entry_timestamp_this_cycle && minimum_reliable_timestamp < latest_listed_entry_timestamp_this_cycle) {
diff --git a/libminifi/include/core/Resource.h b/libminifi/include/core/Resource.h
index d030c2b61..3a6c56350 100644
--- a/libminifi/include/core/Resource.h
+++ b/libminifi/include/core/Resource.h
@@ -67,11 +67,13 @@ class StaticClassType {
getClassLoader().unregisterClass(construction_name);
}
}
-
+#pragma GCC diagnostic push
+#pragma GCC diagnostic ignored "-Wdangling-reference"
static const StaticClassType& get(const std::string& name, const std::vector<std::string>& construction_names) {
static const StaticClassType instance(name, construction_names);
return instance;
}
+#pragma GCC diagnostic pop
private:
std::string name_;
diff --git a/libminifi/src/utils/TimeUtil.cpp b/libminifi/src/utils/TimeUtil.cpp
index 58aaec3a2..1bde905d2 100644
--- a/libminifi/src/utils/TimeUtil.cpp
+++ b/libminifi/src/utils/TimeUtil.cpp
@@ -37,7 +37,7 @@ void setClock(std::shared_ptr<SteadyClock> clock) {
std::optional<std::chrono::system_clock::time_point> parseRfc3339(const std::string& str) {
std::istringstream stream(str);
- date::year_month_day date_part;
+ date::year_month_day date_part{};
date::from_stream(stream, "%F", date_part);
if (stream.fail())
diff --git a/libminifi/test/TestBase.cpp b/libminifi/test/TestBase.cpp
index 1a08f71b4..71cd2bcdc 100644
--- a/libminifi/test/TestBase.cpp
+++ b/libminifi/test/TestBase.cpp
@@ -52,8 +52,7 @@ std::shared_ptr<LogTestController> LogTestController::getInstance(const std::sha
} else {
// in practice I'd use a derivation here or another paradigm entirely but for the purposes of this test code
// having extra overhead is negligible. this is the most readable and least impactful way
- auto instance = std::shared_ptr<LogTestController>(new LogTestController(logger_properties));
- map.insert(std::make_pair(logger_properties, instance));
+ map.insert(std::make_pair(logger_properties, std::shared_ptr<LogTestController>(new LogTestController(logger_properties))));
return map.find(logger_properties)->second;
}
}
diff --git a/thirdparty/azure-sdk-cpp/azure-sdk-cpp-remove-samples.patch b/thirdparty/azure-sdk-cpp/azure-sdk-cpp-remove-samples.patch
deleted file mode 100644
index c9e5ea658..000000000
--- a/thirdparty/azure-sdk-cpp/azure-sdk-cpp-remove-samples.patch
+++ /dev/null
@@ -1,23 +0,0 @@
-# Samples require OpenSSL library on host that should not be required
-diff --git a/sdk/identity/azure-identity/CMakeLists.txt b/sdk/identity/azure-identity/CMakeLists.txt
-index 5a099b0e..ba8920dc 100644
---- a/sdk/identity/azure-identity/CMakeLists.txt
-+++ b/sdk/identity/azure-identity/CMakeLists.txt
-@@ -92,6 +92,3 @@ if (BUILD_PERFORMANCE_TESTS)
- add_subdirectory(test/perf)
- endif()
-
--if (AZ_ALL_LIBRARIES)
-- add_subdirectory(samples)
--endif()
-
-diff --git a/CMakeLists.txt b/CMakeLists.txt
-index 173bca57..e5e4e9a0 100644
---- a/CMakeLists.txt
-+++ b/CMakeLists.txt
-@@ -95,4 +95,3 @@ add_subdirectory(sdk/identity)
- add_subdirectory(sdk/keyvault)
- add_subdirectory(sdk/storage)
- add_subdirectory(sdk/template)
--add_subdirectory(samples/integration/vcpkg-keyvault)
-
diff --git a/thirdparty/azure-sdk-cpp/gcc13.patch b/thirdparty/azure-sdk-cpp/gcc13.patch
new file mode 100644
index 000000000..41db6ff51
--- /dev/null
+++ b/thirdparty/azure-sdk-cpp/gcc13.patch
@@ -0,0 +1,44 @@
+diff '--color=auto' -rupN azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0/sdk/core/azure-core/inc/azure/core/base64.hpp azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0-patched/sdk/core/azure-core/inc/azure/core/base64.hpp
+--- azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0/sdk/core/azure-core/inc/azure/core/base64.hpp 2023-07-11 04:18:12.000000000 +0200
++++ azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0-patched/sdk/core/azure-core/inc/azure/core/base64.hpp 2023-07-31 10:24:28.267049474 +0200
+@@ -13,6 +13,7 @@
+ #include <stdexcept>
+ #include <string>
+ #include <vector>
++#include <cstdint>
+
+ namespace Azure { namespace Core {
+
+diff '--color=auto' -rupN azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0/sdk/core/azure-core/inc/azure/core/uuid.hpp azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0-patched/sdk/core/azure-core/inc/azure/core/uuid.hpp
+--- azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0/sdk/core/azure-core/inc/azure/core/uuid.hpp 2023-07-11 04:18:12.000000000 +0200
++++ azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0-patched/sdk/core/azure-core/inc/azure/core/uuid.hpp 2023-07-31 10:29:58.670195225 +0200
+@@ -13,6 +13,7 @@
+ #include <array>
+ #include <cstring>
+ #include <string>
++#include <cstdint>
+
+ namespace Azure { namespace Core {
+ /**
+diff '--color=auto' -rupN azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0/sdk/keyvault/azure-security-keyvault-keys/src/private/key_sign_parameters.hpp azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0-patched/sdk/keyvault/azure-security-keyvault-keys/src/private/key_sign_parameters.hpp
+--- azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0/sdk/keyvault/azure-security-keyvault-keys/src/private/key_sign_parameters.hpp 2023-07-11 04:18:12.000000000 +0200
++++ azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0-patched/sdk/keyvault/azure-security-keyvault-keys/src/private/key_sign_parameters.hpp 2023-07-31 10:42:25.125314240 +0200
+@@ -11,6 +11,7 @@
+
+ #include <string>
+ #include <vector>
++#include <cstdint>
+
+ namespace Azure {
+ namespace Security {
+diff '--color=auto' -rupN azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0/sdk/keyvault/azure-security-keyvault-keys/src/private/key_verify_parameters.hpp azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0-patched/sdk/keyvault/azure-security-keyvault-keys/src/private/key_verify_parameters.hpp
+--- azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0/sdk/keyvault/azure-security-keyvault-keys/src/private/key_verify_parameters.hpp 2023-07-11 04:18:12.000000000 +0200
++++ azure-sdk-for-cpp-azure-storage-files-datalake_12.7.0-patched/sdk/keyvault/azure-security-keyvault-keys/src/private/key_verify_parameters.hpp 2023-07-31 11:02:08.775192072 +0200
+@@ -11,6 +11,7 @@
+
+ #include <string>
+ #include <vector>
++#include <cstdint>
+
+ namespace Azure {
+ namespace Security {
diff --git a/thirdparty/azure-sdk-cpp/remove-amqp.patch b/thirdparty/azure-sdk-cpp/remove-amqp.patch
new file mode 100644
index 000000000..daacddd01
--- /dev/null
+++ b/thirdparty/azure-sdk-cpp/remove-amqp.patch
@@ -0,0 +1,15 @@
+diff -rupN a/sdk/core/CMakeLists.txt b/sdk/core/CMakeLists.txt
+--- a/sdk/core/CMakeLists.txt 2023-07-11 04:18:12.000000000 +0200
++++ b/sdk/core/CMakeLists.txt 2023-07-25 16:58:17.919348377 +0200
+@@ -10,11 +10,6 @@ set(CMAKE_WINDOWS_EXPORT_ALL_SYMBOLS ON)
+
+ add_subdirectory(azure-core)
+
+-if (NOT BUILD_WINDOWS_UWP)
+- message(STATUS "Including AMQP library")
+- add_subdirectory(azure-core-amqp)
+-endif()
+-
+ if (NOT DISABLE_AZURE_CORE_OPENTELEMETRY)
+ add_subdirectory(azure-core-tracing-opentelemetry)
+ endif()