This is an automated email from the ASF dual-hosted git repository. szaszm pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit cafc3b1a003ef55103d49af09b2283ab93b79384 Author: Martin Zink <[email protected]> AuthorDate: Wed Aug 2 17:54:36 2023 +0200 MINIFICPP-2159 Add heterogeneous lookup to FlatMap Closes #1612 Signed-off-by: Marton Szasz <[email protected]> --- extensions/aws/processors/FetchS3Object.cpp | 2 +- extensions/gcp/tests/DeleteGCSObjectTests.cpp | 12 ++--- extensions/gcp/tests/FetchGCSObjectTests.cpp | 8 ++-- extensions/gcp/tests/PutGCSObjectTests.cpp | 16 +++---- extensions/http-curl/processors/InvokeHTTP.cpp | 16 +++---- extensions/libarchive/CompressContent.cpp | 2 +- extensions/mqtt/processors/ConsumeMQTT.cpp | 4 +- extensions/python/types/PyScriptFlowFile.cpp | 8 ++-- .../standard-processors/tests/unit/GetTCPTests.cpp | 2 +- libminifi/include/core/FlowFile.h | 54 +++++++++------------- libminifi/include/core/ProcessSession.h | 4 +- libminifi/include/utils/FlatMap.h | 40 ++++++++-------- libminifi/src/core/FlowFile.cpp | 31 +++---------- libminifi/src/core/ProcessSession.cpp | 6 +-- libminifi/test/unit/FlatMapTests.cpp | 18 +++++++- 15 files changed, 105 insertions(+), 118 deletions(-) diff --git a/extensions/aws/processors/FetchS3Object.cpp b/extensions/aws/processors/FetchS3Object.cpp index 5389b57bd..311af7cf8 100644 --- a/extensions/aws/processors/FetchS3Object.cpp +++ b/extensions/aws/processors/FetchS3Object.cpp @@ -90,7 +90,7 @@ void FetchS3Object::onTrigger(const std::shared_ptr<core::ProcessContext> &conte }); if (result) { - auto putAttributeIfNotEmpty = [&](const std::string& attribute, const std::string& value) { + auto putAttributeIfNotEmpty = [&](std::string_view attribute, const std::string& value) { if (!value.empty()) { session->putAttribute(flow_file, attribute, value); } diff --git a/extensions/gcp/tests/DeleteGCSObjectTests.cpp b/extensions/gcp/tests/DeleteGCSObjectTests.cpp index 7d4166188..1f54fccf0 100644 --- a/extensions/gcp/tests/DeleteGCSObjectTests.cpp +++ b/extensions/gcp/tests/DeleteGCSObjectTests.cpp @@ -68,8 +68,8 @@ TEST_F(DeleteGCSObjectTests, MissingBucket) { const auto& result = test_controller_.trigger("hello world"); EXPECT_EQ(0, result.at(DeleteGCSObject::Success).size()); ASSERT_EQ(1, result.at(DeleteGCSObject::Failure).size()); - EXPECT_EQ(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_DOMAIN))); - EXPECT_EQ(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_REASON))); + EXPECT_EQ(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN)); + EXPECT_EQ(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON)); EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(DeleteGCSObject::Failure)[0])); } @@ -80,8 +80,8 @@ TEST_F(DeleteGCSObjectTests, ServerGivesPermaError) { const auto& result = test_controller_.trigger("hello world"); EXPECT_EQ(0, result.at(DeleteGCSObject::Success).size()); ASSERT_EQ(1, result.at(DeleteGCSObject::Failure).size()); - EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_DOMAIN))); - EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_REASON))); + EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN)); + EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON)); EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(DeleteGCSObject::Failure)[0])); } @@ -92,8 +92,8 @@ TEST_F(DeleteGCSObjectTests, ServerGivesTransientErrors) { const auto& result = test_controller_.trigger("hello world", {{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"}}); EXPECT_EQ(0, result.at(DeleteGCSObject::Success).size()); ASSERT_EQ(1, result.at(DeleteGCSObject::Failure).size()); - EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_DOMAIN))); - EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_REASON))); + EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN)); + EXPECT_NE(std::nullopt, result.at(DeleteGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON)); EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(DeleteGCSObject::Failure)[0])); } diff --git a/extensions/gcp/tests/FetchGCSObjectTests.cpp b/extensions/gcp/tests/FetchGCSObjectTests.cpp index 98a5df557..c82d1f3e8 100644 --- a/extensions/gcp/tests/FetchGCSObjectTests.cpp +++ b/extensions/gcp/tests/FetchGCSObjectTests.cpp @@ -65,8 +65,8 @@ TEST_F(FetchGCSObjectTests, MissingBucket) { const auto& result = test_controller_.trigger("hello world"); EXPECT_EQ(0, result.at(FetchGCSObject::Success).size()); ASSERT_EQ(1, result.at(FetchGCSObject::Failure).size()); - EXPECT_EQ(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_DOMAIN))); - EXPECT_EQ(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_REASON))); + EXPECT_EQ(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN)); + EXPECT_EQ(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON)); EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(FetchGCSObject::Failure)[0])); } @@ -91,8 +91,8 @@ TEST_F(FetchGCSObjectTests, ServerError) { const auto& result = test_controller_.trigger("hello world", {{std::string(minifi_gcp::GCS_BUCKET_ATTR), "bucket-from-attribute"}}); EXPECT_EQ(0, result.at(FetchGCSObject::Success).size()); ASSERT_EQ(1, result.at(FetchGCSObject::Failure).size()); - EXPECT_NE(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_DOMAIN))); - EXPECT_NE(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_REASON))); + EXPECT_NE(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN)); + EXPECT_NE(std::nullopt, result.at(FetchGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON)); } TEST_F(FetchGCSObjectTests, HappyPath) { diff --git a/extensions/gcp/tests/PutGCSObjectTests.cpp b/extensions/gcp/tests/PutGCSObjectTests.cpp index ee82b281e..266e10238 100644 --- a/extensions/gcp/tests/PutGCSObjectTests.cpp +++ b/extensions/gcp/tests/PutGCSObjectTests.cpp @@ -84,8 +84,8 @@ TEST_F(PutGCSObjectTests, MissingBucket) { const auto& result = test_controller_.trigger("hello world"); EXPECT_EQ(0, result.at(PutGCSObject::Success).size()); ASSERT_EQ(1, result.at(PutGCSObject::Failure).size()); - EXPECT_EQ(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_DOMAIN))); - EXPECT_EQ(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_REASON))); + EXPECT_EQ(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN)); + EXPECT_EQ(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON)); EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(PutGCSObject::Failure)[0])); } @@ -111,8 +111,8 @@ TEST_F(PutGCSObjectTests, ServerGivesTransientErrors) { const auto& result = test_controller_.trigger("hello world"); EXPECT_EQ(0, result.at(PutGCSObject::Success).size()); ASSERT_EQ(1, result.at(PutGCSObject::Failure).size()); - EXPECT_NE(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_DOMAIN))); - EXPECT_NE(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_REASON))); + EXPECT_NE(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN)); + EXPECT_NE(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON)); EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(PutGCSObject::Failure)[0])); } @@ -123,8 +123,8 @@ TEST_F(PutGCSObjectTests, ServerGivesPermaError) { const auto& result = test_controller_.trigger("hello world"); EXPECT_EQ(0, result.at(PutGCSObject::Success).size()); ASSERT_EQ(1, result.at(PutGCSObject::Failure).size()); - EXPECT_NE(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_DOMAIN))); - EXPECT_NE(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(std::string(minifi_gcp::GCS_ERROR_REASON))); + EXPECT_NE(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_DOMAIN)); + EXPECT_NE(std::nullopt, result.at(PutGCSObject::Failure)[0]->getAttribute(minifi_gcp::GCS_ERROR_REASON)); EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(PutGCSObject::Failure)[0])); } @@ -193,8 +193,8 @@ TEST_F(PutGCSObjectTests, ValidServerSideEncryptionTest) { const auto& result = test_controller_.trigger("hello world"); ASSERT_EQ(1, result.at(PutGCSObject::Success).size()); EXPECT_EQ(0, result.at(PutGCSObject::Failure).size()); - EXPECT_NE(std::nullopt, result.at(PutGCSObject::Success)[0]->getAttribute(std::string(minifi_gcp::GCS_ENCRYPTION_SHA256_ATTR))); - EXPECT_NE(std::nullopt, result.at(PutGCSObject::Success)[0]->getAttribute(std::string(minifi_gcp::GCS_ENCRYPTION_ALGORITHM_ATTR))); + EXPECT_NE(std::nullopt, result.at(PutGCSObject::Success)[0]->getAttribute(minifi_gcp::GCS_ENCRYPTION_SHA256_ATTR)); + EXPECT_NE(std::nullopt, result.at(PutGCSObject::Success)[0]->getAttribute(minifi_gcp::GCS_ENCRYPTION_ALGORITHM_ATTR)); EXPECT_EQ("hello world", test_controller_.plan->getContent(result.at(PutGCSObject::Success)[0])); } diff --git a/extensions/http-curl/processors/InvokeHTTP.cpp b/extensions/http-curl/processors/InvokeHTTP.cpp index cb52b3c4d..75f70b19f 100644 --- a/extensions/http-curl/processors/InvokeHTTP.cpp +++ b/extensions/http-curl/processors/InvokeHTTP.cpp @@ -286,10 +286,10 @@ void InvokeHTTP::onTriggerWithClient(const std::shared_ptr<core::ProcessContext> int64_t http_code = client.getResponseCode(); const char* content_type = client.getContentType(); - flow_file->addAttribute(std::string(STATUS_CODE), std::to_string(http_code)); - if (!response_headers.empty()) { flow_file->addAttribute(std::string(STATUS_MESSAGE), response_headers.at(0)); } - flow_file->addAttribute(std::string(REQUEST_URL), client.getURL()); - flow_file->addAttribute(std::string(TRANSACTION_ID), transaction_id); + flow_file->addAttribute(STATUS_CODE, std::to_string(http_code)); + if (!response_headers.empty()) { flow_file->addAttribute(STATUS_MESSAGE, response_headers.at(0)); } + flow_file->addAttribute(REQUEST_URL, client.getURL()); + flow_file->addAttribute(TRANSACTION_ID, transaction_id); bool is_success = ((http_code / 100) == 2); @@ -307,10 +307,10 @@ void InvokeHTTP::onTriggerWithClient(const std::shared_ptr<core::ProcessContext> // if content type isn't returned we should return application/octet-stream // as per RFC 2046 -- 4.5.1 response_flow->addAttribute(core::SpecialFlowAttribute::MIME_TYPE, content_type ? std::string(content_type) : DefaultContentType); - response_flow->addAttribute(std::string(STATUS_CODE), std::to_string(http_code)); - if (!response_headers.empty()) { response_flow->addAttribute(std::string(STATUS_MESSAGE), response_headers.at(0)); } - response_flow->addAttribute(std::string(REQUEST_URL), client.getURL()); - response_flow->addAttribute(std::string(TRANSACTION_ID), transaction_id); + response_flow->addAttribute(STATUS_CODE, std::to_string(http_code)); + if (!response_headers.empty()) { response_flow->addAttribute(STATUS_MESSAGE, response_headers.at(0)); } + response_flow->addAttribute(REQUEST_URL, client.getURL()); + response_flow->addAttribute(TRANSACTION_ID, transaction_id); io::BufferStream stream(gsl::make_span(response_body).as_span<const std::byte>()); // need an import from the data stream. session->importFrom(stream, response_flow); diff --git a/extensions/libarchive/CompressContent.cpp b/extensions/libarchive/CompressContent.cpp index ab08234f9..966fa7de2 100644 --- a/extensions/libarchive/CompressContent.cpp +++ b/extensions/libarchive/CompressContent.cpp @@ -89,7 +89,7 @@ void CompressContent::processFlowFile(const std::shared_ptr<core::FlowFile>& flo std::string attr; flowFile->getAttribute(core::SpecialFlowAttribute::MIME_TYPE, attr); if (attr.empty()) { - logger_->log_error("No %s attribute existed for the flow, route to failure", core::SpecialFlowAttribute::MIME_TYPE); + logger_->log_error("No %s attribute existed for the flow, route to failure", std::string(core::SpecialFlowAttribute::MIME_TYPE)); session->transfer(flowFile, Failure); return; } diff --git a/extensions/mqtt/processors/ConsumeMQTT.cpp b/extensions/mqtt/processors/ConsumeMQTT.cpp index 06436cd6f..f8be50268 100644 --- a/extensions/mqtt/processors/ConsumeMQTT.cpp +++ b/extensions/mqtt/processors/ConsumeMQTT.cpp @@ -102,8 +102,8 @@ void ConsumeMQTT::onTriggerImpl(const std::shared_ptr<core::ProcessContext>& /*c session->remove(flow_file); } else { putUserPropertiesAsAttributes(message, flow_file, session); - session->putAttribute(flow_file, std::string(BrokerOutputAttribute.name), uri_); - session->putAttribute(flow_file, std::string(TopicOutputAttribute.name), message.topic); + session->putAttribute(flow_file, BrokerOutputAttribute.name, uri_); + session->putAttribute(flow_file, TopicOutputAttribute.name, message.topic); fillAttributeFromContentType(message, flow_file, session); logger_->log_debug("ConsumeMQTT processing success for the flow with UUID %s topic %s", flow_file->getUUIDStr(), message.topic); session->transfer(flow_file, Success); diff --git a/extensions/python/types/PyScriptFlowFile.cpp b/extensions/python/types/PyScriptFlowFile.cpp index e0a742fa7..742c747a9 100644 --- a/extensions/python/types/PyScriptFlowFile.cpp +++ b/extensions/python/types/PyScriptFlowFile.cpp @@ -72,7 +72,7 @@ PyObject* PyScriptFlowFile::getAttribute(PyScriptFlowFile* self, PyObject* args) if (!PyArg_ParseTuple(args, "s", &attribute)) { throw PyException(); } - return object::returnReference(flow_file->getAttribute(std::string(attribute)).value_or("")); + return object::returnReference(flow_file->getAttribute(attribute).value_or("")); } PyObject* PyScriptFlowFile::addAttribute(PyScriptFlowFile* self, PyObject* args) { @@ -88,7 +88,7 @@ PyObject* PyScriptFlowFile::addAttribute(PyScriptFlowFile* self, PyObject* args) throw PyException(); } - return object::returnReference(flow_file->addAttribute(std::string(key), std::string(value))); + return object::returnReference(flow_file->addAttribute(key, std::string(value))); } PyObject* PyScriptFlowFile::updateAttribute(PyScriptFlowFile* self, PyObject* args) { @@ -104,7 +104,7 @@ PyObject* PyScriptFlowFile::updateAttribute(PyScriptFlowFile* self, PyObject* ar throw PyException(); } - return object::returnReference(flow_file->updateAttribute(std::string(key), std::string(value))); + return object::returnReference(flow_file->updateAttribute(key, std::string(value))); } PyObject* PyScriptFlowFile::removeAttribute(PyScriptFlowFile* self, PyObject* args) { @@ -118,7 +118,7 @@ PyObject* PyScriptFlowFile::removeAttribute(PyScriptFlowFile* self, PyObject* ar if (!PyArg_ParseTuple(args, "s", &attribute)) { throw PyException(); } - return object::returnReference(flow_file->removeAttribute(std::string(attribute))); + return object::returnReference(flow_file->removeAttribute(attribute)); } PyObject* PyScriptFlowFile::setAttribute(PyScriptFlowFile* self, PyObject* args) { diff --git a/extensions/standard-processors/tests/unit/GetTCPTests.cpp b/extensions/standard-processors/tests/unit/GetTCPTests.cpp index fb95ced9d..48c5b1640 100644 --- a/extensions/standard-processors/tests/unit/GetTCPTests.cpp +++ b/extensions/standard-processors/tests/unit/GetTCPTests.cpp @@ -34,7 +34,7 @@ namespace org::apache::nifi::minifi::test { void check_for_attributes(core::FlowFile& flow_file, uint16_t port) { const auto local_addresses = {"127.0.0.1:" + std::to_string(port), "::ffff:127.0.0.1:" + std::to_string(port), "::1:" + std::to_string(port)}; - CHECK(ranges::contains(local_addresses, flow_file.getAttribute(std::string(GetTCP::SourceEndpoint.name)))); + CHECK(ranges::contains(local_addresses, flow_file.getAttribute(GetTCP::SourceEndpoint.name))); } minifi::utils::net::SslData createSslDataForServer() { diff --git a/libminifi/include/core/FlowFile.h b/libminifi/include/core/FlowFile.h index 53252f97e..867b91180 100644 --- a/libminifi/include/core/FlowFile.h +++ b/libminifi/include/core/FlowFile.h @@ -15,8 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -#ifndef LIBMINIFI_INCLUDE_CORE_FLOWFILE_H_ -#define LIBMINIFI_INCLUDE_CORE_FLOWFILE_H_ +#pragma once #include <map> #include <memory> @@ -34,11 +33,7 @@ #include "utils/FlatMap.h" #include "utils/Export.h" -namespace org { -namespace apache { -namespace nifi { -namespace minifi { -namespace core { +namespace org::apache::nifi::minifi::core { class Connectable; @@ -125,7 +120,7 @@ class FlowFile : public CoreComponent, public ReferenceContainer { * Sets the lineage start date * @param date new lineage start date */ - void setLineageStartDate(const std::chrono::system_clock::time_point date); + void setLineageStartDate(std::chrono::system_clock::time_point date); void setLineageIdentifiers(const std::vector<utils::Identifier>& lineage_Identifiers) { lineage_Identifiers_ = lineage_Identifiers; @@ -137,9 +132,9 @@ class FlowFile : public CoreComponent, public ReferenceContainer { * @param value value to set * @return result of finding key */ - bool getAttribute(const std::string& key, std::string& value) const; + bool getAttribute(std::string_view key, std::string& value) const; - [[nodiscard]] std::optional<std::string> getAttribute(const std::string& key) const; + [[nodiscard]] std::optional<std::string> getAttribute(std::string_view key) const; /** * Updates the value in the attribute map that corresponds @@ -148,14 +143,14 @@ class FlowFile : public CoreComponent, public ReferenceContainer { * @param value value to set to attribute name * @return result of finding key */ - bool updateAttribute(std::string key, std::string value); + bool updateAttribute(std::string_view key, const std::string& value); /** * Removes the attribute * @param key attribute name to remove * @return result of finding key */ - bool removeAttribute(std::string key); + bool removeAttribute(std::string_view key); /** * setAttribute, if attribute already there, update it, else, add it @@ -184,7 +179,7 @@ class FlowFile : public CoreComponent, public ReferenceContainer { * adds an attribute if it does not exist * */ - bool addAttribute(const std::string& key, const std::string& value); + bool addAttribute(std::string_view key, const std::string& value); /** * Set the size of this record. @@ -296,36 +291,29 @@ class FlowFile : public CoreComponent, public ReferenceContainer { // FlowFile Attribute struct SpecialFlowAttribute { // The flowfile's path indicates the relative directory to which a FlowFile belongs and does not contain the filename - MINIFIAPI static const std::string PATH; + MINIFIAPI static constexpr std::string_view PATH = "path"; // The flowfile's absolute path indicates the absolute directory to which a FlowFile belongs and does not contain the filename - MINIFIAPI static const std::string ABSOLUTE_PATH; + MINIFIAPI static constexpr std::string_view ABSOLUTE_PATH = "absolute.path"; // The filename of the FlowFile. The filename should not contain any directory structure. - MINIFIAPI static const std::string FILENAME; + MINIFIAPI static constexpr std::string_view FILENAME = "filename"; // A unique UUID assigned to this FlowFile. - MINIFIAPI static const std::string UUID; + MINIFIAPI static constexpr std::string_view UUID = "uuid"; // A numeric value indicating the FlowFile priority - MINIFIAPI static const std::string priority; + MINIFIAPI static constexpr std::string_view priority = "priority"; // The MIME Type of this FlowFile - MINIFIAPI static const std::string MIME_TYPE; + MINIFIAPI static constexpr std::string_view MIME_TYPE = "mime.type"; // Specifies the reason that a FlowFile is being discarded - MINIFIAPI static const std::string DISCARD_REASON; + MINIFIAPI static constexpr std::string_view DISCARD_REASON = "discard.reason"; // Indicates an identifier other than the FlowFile's UUID that is known to refer to this FlowFile. - MINIFIAPI static const std::string ALTERNATE_IDENTIFIER; + MINIFIAPI static constexpr std::string_view ALTERNATE_IDENTIFIER = "alternate.identifier"; // Flow identifier - MINIFIAPI static const std::string FLOW_ID; + MINIFIAPI static constexpr std::string_view FLOW_ID = "flow.id"; - static const auto& getSpecialFlowAttributes() { - static const std::array<std::string_view, 9> SPECIAL_FLOW_ATTRIBUTES { - PATH, ABSOLUTE_PATH, FILENAME, UUID, priority, MIME_TYPE, DISCARD_REASON, ALTERNATE_IDENTIFIER, FLOW_ID + static constexpr std::array<std::string_view, 9> getSpecialFlowAttributes() { + return { + PATH, ABSOLUTE_PATH, FILENAME, UUID, priority, MIME_TYPE, DISCARD_REASON, ALTERNATE_IDENTIFIER, FLOW_ID }; - return SPECIAL_FLOW_ATTRIBUTES; } }; -} // namespace core -} // namespace minifi -} // namespace nifi -} // namespace apache -} // namespace org - -#endif // LIBMINIFI_INCLUDE_CORE_FLOWFILE_H_ +} // namespace org::apache::nifi::minifi::core diff --git a/libminifi/include/core/ProcessSession.h b/libminifi/include/core/ProcessSession.h index 2ba5f9cda..36076e230 100644 --- a/libminifi/include/core/ProcessSession.h +++ b/libminifi/include/core/ProcessSession.h @@ -87,9 +87,9 @@ class ProcessSession : public ReferenceContainer { // Transfer the FlowFile to the relationship virtual void transfer(const std::shared_ptr<core::FlowFile>& flow, const Relationship& relationship); // Put Attribute - void putAttribute(const std::shared_ptr<core::FlowFile>& flow, const std::string& key, const std::string& value); + void putAttribute(const std::shared_ptr<core::FlowFile>& flow, std::string_view key, const std::string& value); // Remove Attribute - void removeAttribute(const std::shared_ptr<core::FlowFile>& flow, const std::string& key); + void removeAttribute(const std::shared_ptr<core::FlowFile>& flow, std::string_view key); // Remove Flow File void remove(const std::shared_ptr<core::FlowFile> &flow); // Access the contents of the flow file as an input stream; returns null if the flow file has no content claim diff --git a/libminifi/include/utils/FlatMap.h b/libminifi/include/utils/FlatMap.h index b5d3d35aa..adbdf6373 100644 --- a/libminifi/include/utils/FlatMap.h +++ b/libminifi/include/utils/FlatMap.h @@ -23,11 +23,7 @@ #include <vector> #include <utility> -namespace org { -namespace apache { -namespace nifi { -namespace minifi { -namespace utils { +namespace org::apache::nifi::minifi::utils { template<typename K, typename V> class FlatMap{ @@ -128,7 +124,7 @@ class FlatMap{ FlatMap(InputIterator begin, InputIterator end) : data_(begin, end) {} FlatMap& operator=(const FlatMap& source) = default; - FlatMap& operator=(FlatMap&& source) = default; + FlatMap& operator=(FlatMap&& source) noexcept = default; FlatMap& operator=(std::initializer_list<value_type> items) { data_ = items; return *this; @@ -138,7 +134,9 @@ class FlatMap{ return data_.size(); } - V& operator[](const K& key) { + template<typename T> + requires std::constructible_from<K, T> && std::equality_comparable_with<K, T> + V& operator[](const T& key) { auto it = find(key); if (it != end()) { return it->second; @@ -147,7 +145,8 @@ class FlatMap{ return data_.rbegin()->second; } - const V& at(const K& key) const { + template<std::equality_comparable_with<K> T> + const V& at(const T& key) const { auto it = find(key); if (it != end()) { return it->second; @@ -155,7 +154,8 @@ class FlatMap{ throw std::out_of_range("utils::FlatMap::at"); } - V& at(const K& key) { + template<std::equality_comparable_with<K> T> + V& at(const T& key) { auto it = find(key); if (it != end()) { return it->second; @@ -170,7 +170,8 @@ class FlatMap{ return iterator{data_.begin() + offset}; } - std::size_t erase(const K& key) { + template<std::equality_comparable_with<K> T> + std::size_t erase(const T& key) { for (auto it = data_.begin(); it != data_.end(); ++it) { if (it->first == key) { std::swap(*data_.rbegin(), *it); @@ -212,14 +213,16 @@ class FlatMap{ return {iterator{data_.begin() + data_.size() - 1}, true}; } - iterator find(const K& key) { + template<std::equality_comparable_with<K> T> + iterator find(const T& key) { for (auto it = data_.begin(); it != data_.end(); ++it) { if (it->first == key) return iterator{it}; } return end(); } - const_iterator find(const K& key) const { + template<std::equality_comparable_with<K> T> + const_iterator find(const T& key) const { for (auto it = data_.begin(); it != data_.end(); ++it) { if (it->first == key) return const_iterator{it}; } @@ -258,7 +261,7 @@ class FlatMap{ // sort the underlying storage for (const auto& item : *this) { auto it = other.find(item.first); - if (it == other.end() || !(it.second == item.second)) { + if (it == other.end() || it.second != item.second) { return false; } } @@ -282,11 +285,12 @@ class FlatMap{ return data_.max_size(); } - bool empty() const noexcept { + [[nodiscard]] bool empty() const noexcept { return data_.empty(); } - bool contains(const K& key) const { + template<std::equality_comparable_with<K> T> + bool contains(const T& key) const { return find(key) != end(); } @@ -294,8 +298,4 @@ class FlatMap{ Container data_; }; -} // namespace utils -} // namespace minifi -} // namespace nifi -} // namespace apache -} // namespace org +} // namespace org::apache::nifi::minifi::utils diff --git a/libminifi/src/core/FlowFile.cpp b/libminifi/src/core/FlowFile.cpp index 05b626c1d..35d3a257d 100644 --- a/libminifi/src/core/FlowFile.cpp +++ b/libminifi/src/core/FlowFile.cpp @@ -21,15 +21,11 @@ #include <set> #include <cinttypes> #include "core/Repository.h" -#include "core/logging/LoggerConfiguration.h" #include "utils/Id.h" #include "core/FlowFile.h" #include "utils/requirements/Container.h" -namespace org { -namespace apache { -namespace nifi { -namespace minifi { +namespace org::apache::nifi::minifi { namespace core { std::shared_ptr<utils::IdGenerator> FlowFile::id_generator_ = utils::IdGenerator::getIdGenerator(); @@ -145,7 +141,7 @@ std::vector<utils::Identifier> &FlowFile::getlineageIdentifiers() { return lineage_Identifiers_; } -bool FlowFile::getAttribute(const std::string& key, std::string& value) const { +bool FlowFile::getAttribute(std::string_view key, std::string& value) const { const auto attribute = getAttribute(key); if (!attribute) { return false; @@ -154,7 +150,7 @@ bool FlowFile::getAttribute(const std::string& key, std::string& value) const { return true; } -std::optional<std::string> FlowFile::getAttribute(const std::string& key) const { +std::optional<std::string> FlowFile::getAttribute(std::string_view key) const { auto it = attributes_.find(key); if (it != attributes_.end()) { return it->second; @@ -171,7 +167,7 @@ uint64_t FlowFile::getOffset() const { return offset_; } -bool FlowFile::removeAttribute(const std::string key) { +bool FlowFile::removeAttribute(std::string_view key) { auto it = attributes_.find(key); if (it != attributes_.end()) { attributes_.erase(it); @@ -181,7 +177,7 @@ bool FlowFile::removeAttribute(const std::string key) { } } -bool FlowFile::updateAttribute(const std::string key, const std::string value) { +bool FlowFile::updateAttribute(std::string_view key, const std::string& value) { auto it = attributes_.find(key); if (it != attributes_.end()) { it->second = value; @@ -191,7 +187,7 @@ bool FlowFile::updateAttribute(const std::string key, const std::string value) { } } -bool FlowFile::addAttribute(const std::string& key, const std::string& value) { +bool FlowFile::addAttribute(std::string_view key, const std::string& value) { auto it = attributes_.find(key); if (it != attributes_.end()) { // attribute already there in the map @@ -222,23 +218,10 @@ core::Connectable* FlowFile::getConnection() const { return connection_; } -const std::string SpecialFlowAttribute::PATH = "path"; -const std::string SpecialFlowAttribute::ABSOLUTE_PATH = "absolute.path"; -const std::string SpecialFlowAttribute::FILENAME = "filename"; -const std::string SpecialFlowAttribute::UUID = "uuid"; -const std::string SpecialFlowAttribute::priority = "priority"; -const std::string SpecialFlowAttribute::MIME_TYPE = "mime.type"; -const std::string SpecialFlowAttribute::DISCARD_REASON = "discard.reason"; -const std::string SpecialFlowAttribute::ALTERNATE_IDENTIFIER = "alternate.identifier"; -const std::string SpecialFlowAttribute::FLOW_ID = "flow.id"; - } /* namespace core */ namespace utils { template struct assert_container<core::FlowFile::AttributeMap>; } /* namespace utils */ -} /* namespace minifi */ -} /* namespace nifi */ -} /* namespace apache */ -} /* namespace org */ +} // namespace org::apache::nifi::minifi diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp index 5bd15cd4b..56e5e0582 100644 --- a/libminifi/src/core/ProcessSession.cpp +++ b/libminifi/src/core/ProcessSession.cpp @@ -202,17 +202,17 @@ void ProcessSession::remove(const std::shared_ptr<core::FlowFile> &flow) { provenance_report_->drop(flow, reason); } -void ProcessSession::putAttribute(const std::shared_ptr<core::FlowFile>& flow, const std::string& key, const std::string& value) { +void ProcessSession::putAttribute(const std::shared_ptr<core::FlowFile>& flow, std::string_view key, const std::string& value) { flow->setAttribute(key, value); std::stringstream details; details << process_context_->getProcessorNode()->getName() << " modify flow record " << flow->getUUIDStr() << " attribute " << key << ":" << value; provenance_report_->modifyAttributes(flow, details.str()); } -void ProcessSession::removeAttribute(const std::shared_ptr<core::FlowFile>& flow, const std::string& key) { +void ProcessSession::removeAttribute(const std::shared_ptr<core::FlowFile>& flow, std::string_view key) { flow->removeAttribute(key); std::stringstream details; - details << process_context_->getProcessorNode()->getName() << " remove flow record " << flow->getUUIDStr() << " attribute " + key; + details << process_context_->getProcessorNode()->getName() << " remove flow record " << flow->getUUIDStr() << " attribute " << key; provenance_report_->modifyAttributes(flow, details.str()); } diff --git a/libminifi/test/unit/FlatMapTests.cpp b/libminifi/test/unit/FlatMapTests.cpp index 52e2f0572..647821006 100644 --- a/libminifi/test/unit/FlatMapTests.cpp +++ b/libminifi/test/unit/FlatMapTests.cpp @@ -26,7 +26,7 @@ TEST_CASE("FlatMap operator[]", "[flatmap::subscript]") { CHECK(map.contains("valid_key")); CHECK_FALSE(map.contains("invalid_key")); CHECK(map["valid_key"] == "value"); - CHECK(map["invalid_key"] == ""); + CHECK(map["invalid_key"].empty()); CHECK(map.contains("valid_key")); CHECK(map.contains("invalid_key")); } @@ -53,3 +53,19 @@ TEST_CASE("FlatMap const at", "[flatmap::at]") { CHECK(const_map.contains("valid_key")); CHECK_FALSE(const_map.contains("invalid_key")); } + +TEST_CASE("FlatMap supports equality based lookups") { + utils::FlatMap<std::string, std::string> map; + map.insert(std::make_pair("alpha", "value")); + const std::string string_key = "alpha"; + constexpr std::string_view string_view_key = "alpha"; + constexpr std::string_view invalid_string_view_key = "beta"; + CHECK(map.contains(string_key)); + CHECK(map.contains(string_view_key)); + CHECK_FALSE(map.contains(invalid_string_view_key)); + + auto homogeneous_lookup_result = map.find(string_key); + CHECK(homogeneous_lookup_result != map.end()); + CHECK(map.find(string_view_key) == homogeneous_lookup_result); + CHECK(map.find(invalid_string_view_key) != homogeneous_lookup_result); +}
