lordgamez commented on code in PR #2287: URL: https://github.com/apache/nifi-minifi-cpp/pull/2287#discussion_r4514187149
########## extensions/opc/src/FetchOPCFile.cpp: ########## @@ -0,0 +1,252 @@ +/** + * 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 "FetchOPCFile.h" + +#include <algorithm> +#include <limits> +#include <memory> +#include <optional> +#include <string> +#include <utility> +#include <vector> + +#include "core/Resource.h" +#include "minifi-cpp/core/ProcessContext.h" +#include "utils/ProcessorConfigUtils.h" +#include "utils/expected.h" + +namespace org::apache::nifi::minifi::processors { + +namespace { +// Used when neither the file node nor the server specifies a MaxByteStringLength +constexpr UA_UInt32 DEFAULT_CHUNK_SIZE = 32 * 1024; + +constexpr const char* NODE_ID_STATE_KEY = "node_id"; +constexpr const char* SIZE_STATE_KEY = "size"; +constexpr const char* LAST_MODIFIED_TIME_STATE_KEY = "last_modified_time"; +} // namespace + +void FetchOPCFile::initialize() { + setSupportedProperties(Properties); + setSupportedRelationships(Relationships); +} + +void FetchOPCFile::onSchedule(core::ProcessContext& context, core::ProcessSessionFactory& factory) { + logger_->log_trace("FetchOPCFile::onSchedule"); + + BaseOPCProcessor::onSchedule(context, factory); + + node_id_ = utils::parseProperty(context, NodeID); + id_type_ = utils::parseEnumProperty<opc::OPCNodeIDType>(context, NodeIDType); + namespace_idx_ = gsl::narrow<UA_UInt16>(utils::parseU64Property(context, NameSpaceIndex)); + parseNode(context, PathReferenceTypes); + lazy_mode_ = utils::parseEnumProperty<FetchOPCFileLazyModeOptions>(context, Lazy); + cached_file_.reset(); +} + +bool FetchOPCFile::reconnect() { + const bool was_connected = connection_ && connection_->isConnected(); + if (!BaseOPCProcessor::reconnect()) { + return false; + } + if (!was_connected) { + cached_file_.reset(); + } + return true; +} + +bool FetchOPCFile::resolvePath() { + if (id_type_ != opc::OPCNodeIDType::Path || path_node_id_resolved_) { + return true; + } + std::vector<opc::NodeId> translated_node_ids; + auto sc = connection_->translateBrowsePathsToNodeIdsRequest(node_id_, translated_node_ids, namespace_idx_, path_reference_types_, logger_); + if (sc != UA_STATUSCODE_GOOD) { + logger_->log_error("Failed to translate path '{}' to a node id: {}", node_id_, UA_StatusCode_name(sc)); + return false; + } + if (translated_node_ids.size() != 1) { + logger_->log_error("Path '{}' resolved to {} node ids; exactly one file node is required for fetch", node_id_, translated_node_ids.size()); + return false; + } + node_ = std::move(translated_node_ids[0]); + path_node_id_resolved_ = true; + return true; +} + +void FetchOPCFile::invalidateFile() { + cached_file_.reset(); + path_node_id_resolved_ = false; +} + +bool FetchOPCFile::resolveFile() { + if (cached_file_) { + return true; + } + auto remote_file = opc::RemoteFile::create(*connection_, opc::NodeId::copyOf(node_)); + if (!remote_file) { + logger_->log_error("Failed to fetch file node '{}' or find its file methods: {}", node_id_, UA_StatusCode_name(remote_file.error())); + invalidateFile(); + return false; + } + const auto chunk_size = gsl::narrow<UA_Int32>(std::min<UA_UInt32>(remote_file->maxByteStringLength().value_or(DEFAULT_CHUNK_SIZE), + static_cast<UA_UInt32>(std::numeric_limits<UA_Int32>::max()))); + std::optional<opc::NodeId> size_node_id; + std::optional<opc::NodeId> last_modified_time_node_id; + if (lazy_mode_ == FetchOPCFileLazyModeOptions::On) { + // Size is a mandatory, LastModifiedTime is an optional property of FileType nodes (OPC UA Part 20, 4.2.1) + size_node_id = connection_->findChild(node_, "Size", 0, UA_NS0ID_HASPROPERTY) | utils::toOptional(); + last_modified_time_node_id = connection_->findChild(node_, "LastModifiedTime", 0, UA_NS0ID_HASPROPERTY) | utils::toOptional(); + if (!last_modified_time_node_id && size_node_id) { + logger_->log_warn("File node '{}' has no LastModifiedTime property, lazy mode can only detect the changes of its Size", node_id_); + } else if (!last_modified_time_node_id) { + logger_->log_warn("File node '{}' has neither LastModifiedTime nor Size property, lazy mode cannot detect its changes, it is fetched on every trigger", node_id_); + } + } + auto browse_name = connection_->readBrowseName(node_) | utils::toOptional(); + if (!browse_name) { + logger_->log_warn("Failed to read browse name of file node '{}'", node_id_); + } + cached_file_.emplace(CachedFile{ + .remote_file = std::move(*remote_file), + .chunk_size = chunk_size, + .browse_name = std::move(browse_name), + .size_node_id = std::move(size_node_id), + .last_modified_time_node_id = std::move(last_modified_time_node_id)}); + return true; +} + +std::optional<core::StateManager::State> FetchOPCFile::readFileState() { + core::StateManager::State state; + if (cached_file_->size_node_id) { + if (auto size = connection_->readUInt64Value(*cached_file_->size_node_id)) { + state[SIZE_STATE_KEY] = std::to_string(*size); + } + } + if (cached_file_->last_modified_time_node_id) { + if (auto last_modified_time = connection_->readDateTimeValue(*cached_file_->last_modified_time_node_id)) { + state[LAST_MODIFIED_TIME_STATE_KEY] = std::to_string(*last_modified_time); + } + } + if (state.empty()) { + return std::nullopt; + } + state[NODE_ID_STATE_KEY] = fmt::format("{};{};{}", magic_enum::enum_name(id_type_), namespace_idx_, node_id_); + return state; +} + +bool FetchOPCFile::writeOPCFileToFlowFile(opc::RemoteFile& remote_file, opc::RemoteFileHandle file_handle, UA_Int32 chunk_size, core::ProcessSession& session, + const std::shared_ptr<core::FlowFile>& flow_file) { + UA_StatusCode read_status = UA_STATUSCODE_GOOD; + session.write(flow_file, [&](const std::shared_ptr<io::OutputStream>& output_stream) -> io::IoResult { + size_t total_size = 0; + while (true) { + auto chunk = remote_file.read(file_handle, chunk_size); + if (!chunk) { + read_status = chunk.error(); + return io::IoResult::cancelled(); + } + const auto data = chunk->data(); + if (data.empty()) { + return io::IoResult::from(total_size); + } + const auto written = output_stream->write(data); + if (io::isError(written)) { + return io::IoResult::error(); + } + total_size += data.size(); + } + }); Review Comment: If session write fails an exception is thrown and the session is rolled back as expected. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
