Copilot commented on code in PR #2287:
URL: https://github.com/apache/nifi-minifi-cpp/pull/2287#discussion_r4513723053
##########
extensions/opc/include/OPCCommon.h:
##########
@@ -29,6 +29,7 @@
#include <vector>
#include <memory>
#include <expected>
+#include <span>
#include <string_view>
#include <utility>
#include <optional>
Review Comment:
`nodeIdAttributes` is declared in this header with a `std::map` return type,
but `<map>` is not included here. This can cause compilation failures depending
on include order. Add `#include <map>` to this header.
##########
extensions/opc/src/OPCCommon.cpp:
##########
@@ -464,6 +470,90 @@ UA_StatusCode
Client::translateBrowsePathsToNodeIdsRequest(const std::string& pa
}
}
+std::expected<NodeId, UA_StatusCode> Client::findChild(const UA_NodeId&
parent_node_id, std::string_view browse_name,
+ UA_UInt16 browse_name_namespace_index, UA_UInt32 ref_type_id) {
+ std::string name(browse_name);
+ UA_RelativePathElement element;
+ UA_RelativePathElement_init(&element);
+ element.referenceTypeId = UA_NODEID_NUMERIC(0, ref_type_id);
+ element.includeSubtypes = true;
+ element.targetName = UA_QUALIFIEDNAME(browse_name_namespace_index,
name.data());
+
+ UA_BrowsePath browse_path;
+ UA_BrowsePath_init(&browse_path);
+ browse_path.startingNode = parent_node_id;
+ browse_path.relativePath.elements = &element;
+ browse_path.relativePath.elementsSize = 1;
+
+ UA_TranslateBrowsePathsToNodeIdsRequest request;
+ UA_TranslateBrowsePathsToNodeIdsRequest_init(&request);
+ request.browsePaths = &browse_path;
+ request.browsePathsSize = 1;
+
+ UA_TranslateBrowsePathsToNodeIdsResponse response =
UA_Client_Service_translateBrowsePathsToNodeIds(client_, request);
+ const auto guard = gsl::finally([&response]() {
UA_TranslateBrowsePathsToNodeIdsResponse_clear(&response); });
+
+ if (response.responseHeader.serviceResult != UA_STATUSCODE_GOOD) {
+ return std::unexpected{response.responseHeader.serviceResult};
+ }
+ if (response.resultsSize != 1) {
+ return std::unexpected{UA_STATUSCODE_BADUNEXPECTEDERROR};
+ }
+ if (response.results[0].statusCode != UA_STATUSCODE_GOOD) {
+ return std::unexpected{response.results[0].statusCode};
+ }
+ if (response.results[0].targetsSize < 1) {
+ return std::unexpected{UA_STATUSCODE_BADNOMATCH};
+ }
Review Comment:
`Client::findChild` silently returns the first target even when multiple
targets match the browse path. That can hide ambiguous address spaces and lead
to calling the wrong method/property node. Consider returning an error when
`targetsSize != 1` (e.g., `UA_STATUSCODE_BADTOOMANYMATCHES` or a similar
status) so callers can handle ambiguity deterministically.
##########
extensions/opc/include/OPCFileTransfer.h:
##########
@@ -0,0 +1,118 @@
+/**
+ * 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.
+ */
+#pragma once
+
+#include <cstddef>
+#include <expected>
+#include <optional>
+#include <span>
+#include <string_view>
+#include <utility>
+
+#include "OPCCommon.h"
+
+namespace org::apache::nifi::minifi::opc {
+
+// Open mode bits of the FileType Open method (OPC UA Part 20, 4.2.2)
+inline constexpr UA_Byte FILE_MODE_READ = 0x01;
+inline constexpr UA_Byte FILE_MODE_WRITE = 0x02;
+inline constexpr UA_Byte FILE_MODE_ERASE_EXISTING = 0x04;
+inline constexpr UA_Byte FILE_MODE_APPEND = 0x08;
+
+class RemoteFileHandle;
+
+class FileChunk {
+ public:
+ FileChunk() = default;
+ FileChunk(MethodOutput output, std::span<const std::byte> data) :
output_(std::move(output)), data_(data) {
+ }
+
+ [[nodiscard]] std::span<const std::byte> data() const noexcept {
+ return data_;
+ }
+
+ private:
+ MethodOutput output_;
+ std::span<const std::byte> data_;
+};
+
+class RemoteFile {
+ public:
+ static std::expected<RemoteFile, UA_StatusCode> create(Client& client,
NodeId file_node_id);
+
+ std::expected<RemoteFileHandle, UA_StatusCode> open(UA_Byte mode);
+ std::expected<FileChunk, UA_StatusCode> read(const RemoteFileHandle& handle,
UA_Int32 length);
+
+ [[nodiscard]] const NodeId& nodeId() const noexcept {
+ return file_node_id_;
+ }
+
+ // The MaxByteStringLength property of the file, or the ServerCapabilities
MaxByteStringLength if the file does not specify one (OPC UA Part 20, 4.2.1)
+ [[nodiscard]] std::optional<UA_UInt32> maxByteStringLength() const noexcept {
+ return max_byte_string_length_;
+ }
+
+ private:
+ RemoteFile(Client& client, NodeId file_node_id) : client_(&client),
file_node_id_(std::move(file_node_id)) {
+ }
+
+ friend class RemoteFileHandle;
+ std::expected<void, UA_StatusCode> close(RemoteFileHandle& handle);
+
+ Client* client_;
+ NodeId file_node_id_;
+ NodeId open_method_id_;
+ NodeId close_method_id_;
+ NodeId read_method_id_;
+ NodeId write_method_id_;
+ std::optional<UA_UInt32> max_byte_string_length_;
+};
+
+class RemoteFileHandle {
+ public:
+ RemoteFileHandle(RemoteFile& file, UA_UInt32 handle) : file_(file),
handle_(handle) {
+ }
+ RemoteFileHandle(const RemoteFileHandle&) = delete;
+ RemoteFileHandle& operator=(const RemoteFileHandle&) = delete;
+ RemoteFileHandle(RemoteFileHandle&& other) noexcept : file_(other.file_),
handle_(other.handle_), open_(std::exchange(other.open_, false)) {
+ }
+ RemoteFileHandle& operator=(RemoteFileHandle&&) = delete;
+ ~RemoteFileHandle() {
+ if (open_) {
+ std::ignore = file_.close(*this);
Review Comment:
`std::ignore` requires including `<tuple>`. As written, this header may fail
to compile in translation units that don't already include `<tuple>`. Either
include `<tuple>` in this header or replace `std::ignore = ...` with a simpler
discard pattern (e.g., `(void)file_.close(*this);`).
##########
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:
Write failures (`output_stream->write` returning an error) are not handled
after `session.write` completes. In that path, `read_status` stays
`UA_STATUSCODE_GOOD`, so the function may return success and transfer a
partial/empty FlowFile. Track write errors explicitly (e.g., a `bool
write_failed` set inside the lambda) and treat them like a failure: remove the
FlowFile and return `false`.
--
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]
