This is an automated email from the ASF dual-hosted git repository. adamdebreceni pushed a commit to branch MINIFICPP-2845 in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit c1f65b296f97d6d1e273ae17550459fdb0dd754a Author: Adam Debreceni <[email protected]> AuthorDate: Wed Jun 3 14:14:09 2026 +0200 MINIFICPP-2840 - More refactor --- extensions/rocksdb-repos/tests/ProvenanceTests.cpp | 10 +-- .../tests/unit/ProcessorTests.cpp | 16 ++--- libminifi/include/provenance/Provenance.h | 39 +++++----- libminifi/src/core/ProcessSession.cpp | 2 +- libminifi/src/provenance/Provenance.cpp | 83 +++++++++++----------- 5 files changed, 73 insertions(+), 77 deletions(-) diff --git a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp index ecd3a77cf..fca1926bb 100644 --- a/extensions/rocksdb-repos/tests/ProvenanceTests.cpp +++ b/extensions/rocksdb-repos/tests/ProvenanceTests.cpp @@ -52,7 +52,7 @@ TEST_CASE("Test Provenance record serialization", "[Testprovenance::ProvenanceEv record1->setEventDuration(sample); testRepository->storeElement(record1); - auto record2 = std::make_shared<provenance::ProvenanceEventRecordImpl>(); + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); REQUIRE(record2->getEventId() == record1->getEventId()); @@ -77,7 +77,7 @@ TEST_CASE("Test Flowfile record added to provenance", "[TestFlowAndProv1]") { record1->setEventDuration(sample); testRepository->storeElement(record1); - auto record2 = std::make_shared<provenance::ProvenanceEventRecordImpl>(); + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); REQUIRE(record1->getChildrenUuids().size() == 1); @@ -101,7 +101,7 @@ TEST_CASE("Test Provenance record serialization Volatile", "[Testprovenance::Pro record1->setEventDuration(sample); testRepository->storeElement(record1); - auto record2 = std::make_shared<provenance::ProvenanceEventRecordImpl>(); + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); REQUIRE(record2->getEventId() == record1->getEventId()); @@ -127,7 +127,7 @@ TEST_CASE("Test Flowfile record added to provenance using Volatile Repo", "[Test record1->setEventDuration(sample); testRepository->storeElement(record1); - auto record2 = std::make_shared<provenance::ProvenanceEventRecordImpl>(); + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == true); REQUIRE(record1->getChildrenUuids().size() == 1); @@ -151,7 +151,7 @@ TEST_CASE("Test Provenance record serialization NoOp", "[Testprovenance::Provena record1->setEventDuration(sample); REQUIRE(testRepository->storeElement(record1)); - auto record2 = std::make_shared<provenance::ProvenanceEventRecordImpl>(); + auto record2 = provenance::ProvenanceEventRecord::create(); record2->setEventId(eventId); REQUIRE(record2->loadFromRepository(testRepository) == false); } diff --git a/extensions/standard-processors/tests/unit/ProcessorTests.cpp b/extensions/standard-processors/tests/unit/ProcessorTests.cpp index cb5905af2..6dd308c7c 100644 --- a/extensions/standard-processors/tests/unit/ProcessorTests.cpp +++ b/extensions/standard-processors/tests/unit/ProcessorTests.cpp @@ -444,18 +444,18 @@ TEST_CASE("Test Find file", "[getfileCreate3]") { REQUIRE(2 == repo->getRepoMap().size()); for (auto entry : repo->getRepoMap()) { - minifi::provenance::ProvenanceEventRecordImpl newRecord; + auto newRecord = minifi::provenance::ProvenanceEventRecordImpl::create(); minifi::io::BufferStream stream(std::as_bytes(std::span(entry.second))); - newRecord.deserialize(stream); + newRecord->deserialize(stream); bool found = false; for (const auto& provRec : records) { - if (provRec->getEventId() == newRecord.getEventId()) { - REQUIRE(provRec->getEventId() == newRecord.getEventId()); - REQUIRE(provRec->getComponentId() == newRecord.getComponentId()); - REQUIRE(provRec->getComponentType() == newRecord.getComponentType()); - REQUIRE(provRec->getDetails() == newRecord.getDetails()); - REQUIRE(provRec->getEventDuration() == newRecord.getEventDuration()); + if (provRec->getEventId() == newRecord->getEventId()) { + REQUIRE(provRec->getEventId() == newRecord->getEventId()); + REQUIRE(provRec->getComponentId() == newRecord->getComponentId()); + REQUIRE(provRec->getComponentType() == newRecord->getComponentType()); + REQUIRE(provRec->getDetails() == newRecord->getDetails()); + REQUIRE(provRec->getEventDuration() == newRecord->getEventDuration()); found = true; break; } diff --git a/libminifi/include/provenance/Provenance.h b/libminifi/include/provenance/Provenance.h index 9d6aa11e2..5539dc5c2 100644 --- a/libminifi/include/provenance/Provenance.h +++ b/libminifi/include/provenance/Provenance.h @@ -45,12 +45,12 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public public: static const char *ProvenanceEventTypeStr[REPLAY + 1]; - ProvenanceEventRecordImpl(ProvenanceEventType event, std::string componentId, std::string componentType); + ProvenanceEventRecordImpl(ProvenanceEventType event, utils::Identifier component_id, std::string component_type); - ProvenanceEventRecordImpl() - : core::SerializableComponentImpl(core::className<ProvenanceEventRecord>()) { - event_time_ = std::chrono::system_clock::now(); - } + ProvenanceEventRecordImpl(const ProvenanceEventRecordImpl&) = delete; + ProvenanceEventRecordImpl(ProvenanceEventRecordImpl&&) = delete; + ProvenanceEventRecordImpl& operator=(const ProvenanceEventRecordImpl&) = delete; + ProvenanceEventRecordImpl& operator=(ProvenanceEventRecordImpl&&) = delete; ~ProvenanceEventRecordImpl() override = default; @@ -245,20 +245,22 @@ class ProvenanceEventRecordImpl : public core::SerializableComponentImpl, public std::string alternate_identifier_uri_; private: - ProvenanceEventRecordImpl(const ProvenanceEventRecordImpl &parent); - ProvenanceEventRecordImpl &operator=(const ProvenanceEventRecordImpl &parent); static std::shared_ptr<core::logging::Logger> logger_; static std::shared_ptr<utils::IdGenerator> id_generator_; }; class ProvenanceReporterImpl : public virtual ProvenanceReporter { public: - ProvenanceReporterImpl(std::shared_ptr<core::Repository> repo, std::string componentId, std::string componentType) - : logger_(core::logging::LoggerFactory<ProvenanceReporter>::getLogger()) { - component_id_ = componentId; - component_type_ = componentType; - repo_ = repo; - } + ProvenanceReporterImpl(std::shared_ptr<core::Repository> repo, utils::Identifier component_id, std::string component_type) + : component_id_(component_id), + component_type_(std::move(component_type)), + logger_(core::logging::LoggerFactory<ProvenanceReporter>::getLogger()), + repo_(std::move(repo)) {} + + ProvenanceReporterImpl(const ProvenanceReporterImpl&) = delete; + ProvenanceReporterImpl(ProvenanceReporterImpl&&) = delete; + ProvenanceReporterImpl& operator=(const ProvenanceReporterImpl&) = delete; + ProvenanceReporterImpl& operator=(ProvenanceReporterImpl&&) = delete; ~ProvenanceReporterImpl() override { clear(); @@ -273,12 +275,10 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter { } void remove(const std::shared_ptr<ProvenanceEventRecord> &event) override { - if (events_.find(event) != events_.end()) { - events_.erase(event); - } + events_.erase(event); } - void clear() override { + void clear() final { events_.clear(); } @@ -308,16 +308,13 @@ class ProvenanceReporterImpl : public virtual ProvenanceReporter { return event; } - std::string component_id_; + utils::Identifier component_id_; std::string component_type_; private: std::shared_ptr<core::logging::Logger> logger_; std::set<std::shared_ptr<ProvenanceEventRecord>> events_; std::shared_ptr<core::Repository> repo_; - - ProvenanceReporterImpl(const ProvenanceReporterImpl &parent); - ProvenanceReporterImpl &operator=(const ProvenanceReporterImpl &parent); }; } // namespace org::apache::nifi::minifi::provenance diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp index d44c71947..31c6a6b29 100644 --- a/libminifi/src/core/ProcessSession.cpp +++ b/libminifi/src/core/ProcessSession.cpp @@ -69,7 +69,7 @@ ProcessSessionImpl::ProcessSessionImpl(std::shared_ptr<ProcessContext> processCo stateManager_(process_context_->getStateManager()) { logger_->log_trace("ProcessSession created for {}", process_context_->getProcessor().getName()); auto repo = process_context_->getProvenanceRepository(); - provenance_report_ = std::make_shared<provenance::ProvenanceReporterImpl>(repo, process_context_->getProcessor().getUUIDStr(), process_context_->getProcessor().getName()); + provenance_report_ = std::make_shared<provenance::ProvenanceReporterImpl>(repo, process_context_->getProcessor().getUUID(), process_context_->getProcessor().getName()); content_session_ = process_context_->getContentRepository()->createSession(); if (stateManager_ && !stateManager_->beginTransaction()) { diff --git a/libminifi/src/provenance/Provenance.cpp b/libminifi/src/provenance/Provenance.cpp index ed4d0638a..66bcff976 100644 --- a/libminifi/src/provenance/Provenance.cpp +++ b/libminifi/src/provenance/Provenance.cpp @@ -40,12 +40,12 @@ std::shared_ptr<core::logging::Logger> ProvenanceEventRecordImpl::logger_ = core const char *ProvenanceEventRecord::ProvenanceEventTypeStr[REPLAY + 1] = { "CREATE", "RECEIVE", "FETCH", "SEND", "DOWNLOAD", // NOLINT(cppcoreguidelines-avoid-c-arrays) "DROP", "EXPIRE", "FORK", "JOIN", "CLONE", "CONTENT_MODIFIED", "ATTRIBUTES_MODIFIED", "ROUTE", "ADDINFO", "REPLAY" }; -ProvenanceEventRecordImpl::ProvenanceEventRecordImpl(ProvenanceEventRecord::ProvenanceEventType event, std::string componentId, std::string componentType) +ProvenanceEventRecordImpl::ProvenanceEventRecordImpl(ProvenanceEventRecord::ProvenanceEventType event, utils::Identifier component_id, std::string component_type) : core::SerializableComponentImpl(core::className<ProvenanceEventRecord>()), event_type_(event), event_time_(std::chrono::system_clock::now()), - component_id_(std::move(componentId)), - component_type_(std::move(componentType)) { + component_id_(std::move(component_id.to_string())), + component_type_(std::move(component_type)) { } bool ProvenanceEventRecordImpl::loadFromRepository(const std::shared_ptr<core::Repository> &repo) { @@ -80,13 +80,13 @@ bool ProvenanceEventRecordImpl::loadFromRepository(const std::shared_ptr<core::R bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) { { - const auto ret = output_stream.write(this->uuid_); + const auto ret = output_stream.write(uuid_); if (ret == 0 || io::isError(ret)) { return false; } } { - uint32_t eventType = this->event_type_; + uint32_t eventType = event_type_; const auto ret = output_stream.write(eventType); if (ret != 4) { return false; @@ -107,7 +107,7 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) { } } { - uint64_t event_duration_ms = this->event_duration_.count(); + uint64_t event_duration_ms = event_duration_.count(); const auto ret = output_stream.write(event_duration_ms); if (ret != 8) { return false; @@ -121,32 +121,32 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) { } } { - const auto ret = output_stream.write(this->component_id_); + const auto ret = output_stream.write(component_id_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = output_stream.write(this->component_type_); + const auto ret = output_stream.write(component_type_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = output_stream.write(this->flow_uuid_); + const auto ret = output_stream.write(flow_uuid_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = output_stream.write(this->details_); + const auto ret = output_stream.write(details_); if (ret == 0 || io::isError(ret)) { return false; } } // write flow attributes { - const auto numAttributes = gsl::narrow<uint32_t>(this->attributes_.size()); + const auto numAttributes = gsl::narrow<uint32_t>(attributes_.size()); const auto ret = output_stream.write(numAttributes); if (ret != 4) { return false; @@ -167,33 +167,33 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) { } } { - const auto ret = output_stream.write(this->content_full_path); + const auto ret = output_stream.write(content_full_path); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = output_stream.write(this->size_); + const auto ret = output_stream.write(size_); if (ret != 8) { return false; } } { - const auto ret = output_stream.write(this->offset_); + const auto ret = output_stream.write(offset_); if (ret != 8) { return false; } } { - const auto ret = output_stream.write(this->source_queue_identifier_); + const auto ret = output_stream.write(source_queue_identifier_); if (ret == 0 || io::isError(ret)) { return false; } } - if (this->event_type_ == ProvenanceEventRecord::FORK || this->event_type_ == ProvenanceEventRecord::CLONE || this->event_type_ == ProvenanceEventRecord::JOIN) { + if (event_type_ == ProvenanceEventRecord::FORK || event_type_ == ProvenanceEventRecord::CLONE || event_type_ == ProvenanceEventRecord::JOIN) { // write UUIDs { - const auto parent_uuids_count = gsl::narrow<uint32_t>(this->parent_uuids_.size()); + const auto parent_uuids_count = gsl::narrow<uint32_t>(parent_uuids_.size()); const auto ret = output_stream.write(parent_uuids_count); if (ret != 4) { return false; @@ -206,7 +206,7 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) { } } { - const auto children_uuids_count = gsl::narrow<uint32_t>(this->children_uuids_.size()); + const auto children_uuids_count = gsl::narrow<uint32_t>(children_uuids_.size()); const auto ret = output_stream.write(children_uuids_count); if (ret != 4) { return false; @@ -218,20 +218,20 @@ bool ProvenanceEventRecordImpl::serialize(io::OutputStream& output_stream) { return false; } } - } else if (this->event_type_ == ProvenanceEventRecord::SEND || this->event_type_ == ProvenanceEventRecord::FETCH) { - const auto ret = output_stream.write(this->transit_uri_); + } else if (event_type_ == ProvenanceEventRecord::SEND || event_type_ == ProvenanceEventRecord::FETCH) { + const auto ret = output_stream.write(transit_uri_); if (ret == 0 || io::isError(ret)) { return false; } - } else if (this->event_type_ == ProvenanceEventRecord::RECEIVE) { + } else if (event_type_ == ProvenanceEventRecord::RECEIVE) { { - const auto ret = output_stream.write(this->transit_uri_); + const auto ret = output_stream.write(transit_uri_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = output_stream.write(this->source_system_flow_file_identifier_); + const auto ret = output_stream.write(source_system_flow_file_identifier_); if (ret == 0 || io::isError(ret)) { return false; } @@ -300,28 +300,28 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { } { - const auto ret = input_stream.read(this->component_id_); + const auto ret = input_stream.read(component_id_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = input_stream.read(this->component_type_); + const auto ret = input_stream.read(component_type_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = input_stream.read(this->flow_uuid_); + const auto ret = input_stream.read(flow_uuid_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = input_stream.read(this->details_); + const auto ret = input_stream.read(details_); if (ret == 0 || io::isError(ret)) { return false; } @@ -351,38 +351,38 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { return false; } } - this->attributes_[key] = value; + attributes_[key] = value; } { - const auto ret = input_stream.read(this->content_full_path); + const auto ret = input_stream.read(content_full_path); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = input_stream.read(this->size_); + const auto ret = input_stream.read(size_); if (ret != 8) { return false; } } { - const auto ret = input_stream.read(this->offset_); + const auto ret = input_stream.read(offset_); if (ret != 8) { return false; } } { - const auto ret = input_stream.read(this->source_queue_identifier_); + const auto ret = input_stream.read(source_queue_identifier_); if (ret == 0 || io::isError(ret)) { return false; } } - if (this->event_type_ == ProvenanceEventRecord::FORK || this->event_type_ == ProvenanceEventRecord::CLONE || this->event_type_ == ProvenanceEventRecord::JOIN) { + if (event_type_ == ProvenanceEventRecord::FORK || event_type_ == ProvenanceEventRecord::CLONE || event_type_ == ProvenanceEventRecord::JOIN) { // read UUIDs uint32_t number = 0; { @@ -400,7 +400,7 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { return false; } } - this->addParentUuid(parentUUID); + addParentUuid(parentUUID); } number = 0; { @@ -417,24 +417,24 @@ bool ProvenanceEventRecordImpl::deserialize(io::InputStream &input_stream) { return false; } } - this->addChildUuid(childUUID); + addChildUuid(childUUID); } - } else if (this->event_type_ == ProvenanceEventRecord::SEND || this->event_type_ == ProvenanceEventRecord::FETCH) { + } else if (event_type_ == ProvenanceEventRecord::SEND || event_type_ == ProvenanceEventRecord::FETCH) { { - const auto ret = input_stream.read(this->transit_uri_); + const auto ret = input_stream.read(transit_uri_); if (ret == 0 || io::isError(ret)) { return false; } } - } else if (this->event_type_ == ProvenanceEventRecord::RECEIVE) { + } else if (event_type_ == ProvenanceEventRecord::RECEIVE) { { - const auto ret = input_stream.read(this->transit_uri_); + const auto ret = input_stream.read(transit_uri_); if (ret == 0 || io::isError(ret)) { return false; } } { - const auto ret = input_stream.read(this->source_system_flow_file_identifier_); + const auto ret = input_stream.read(source_system_flow_file_identifier_); if (ret == 0 || io::isError(ret)) { return false; } @@ -509,7 +509,6 @@ void ProvenanceReporterImpl::clone(const core::FlowFile& parent, const core::Flo if (event) { event->addChildFlowFile(child); - event->addParentFlowFile(parent); add(event); } } @@ -577,7 +576,7 @@ void ProvenanceReporterImpl::fetch(const core::FlowFile& flow_file, const std::s } std::shared_ptr<ProvenanceEventRecord> ProvenanceEventRecord::create() { - return std::make_shared<ProvenanceEventRecordImpl>(); + return std::make_shared<ProvenanceEventRecordImpl>(ProvenanceEventType::CLONE, utils::Identifier{}, ""); } } // namespace org::apache::nifi::minifi::provenance
