Skip to content

MINIFICPP-2906 Add guid nodeid support for OPC processors and some additional fixes #2266

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
lordgamez wants to merge 2 commits into
base: MINIFICPP-2894
Choose a base branch
from
Open

MINIFICPP-2906 Add guid nodeid support for OPC processors and some additional fixes #2266

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
889aae7
MINIFICPP-2906 Add guid nodeid support for OPC processors
lordgamez Sep 17, 2026
7257b9e
Additional node data type support
lordgamez Sep 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 18 additions & 10 deletions .github/references/ubuntu_22_04_clang_arm_manifest.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6570,7 +6570,7 @@
"Namespace index": {
"name": "Namespace index",
"description": "The index of the namespace.",
"validator": "INTEGER_VALIDATOR",
"validator": "NON_NEGATIVE_INTEGER_VALIDATOR",
"required": "true",
"sensitive": "false",
"expressionLanguageScope": "NONE",
Expand Down Expand Up @@ -6746,7 +6746,7 @@
"Namespace index": {
"name": "Namespace index",
"description": "The index of the namespace.",
"validator": "INTEGER_VALIDATOR",
"validator": "NON_NEGATIVE_INTEGER_VALIDATOR",
"required": "true",
"sensitive": "false",
"expressionLanguageScope": "NONE",
Expand Down Expand Up @@ -6779,6 +6779,10 @@
{
"value": "String",
"displayName": "String"
},
{
"value": "Guid",
"displayName": "Guid"
}
]
},
Expand Down Expand Up @@ -6903,17 +6907,17 @@
},
"Parent node ID": {
"name": "Parent node ID",
"description": "Specifies the ID of the root node to traverse",
"description": "Specifies the ID of the root node to traverse. Only required when a new node is created.",
"validator": "VALID",
"required": "true",
"required": "false",
"sensitive": "false",
"expressionLanguageScope": "NONE"
},
"Parent node ID type": {
"name": "Parent node ID type",
"description": "Specifies the type of the provided node ID",
"description": "Specifies the type of the provided node ID. Only required when a new node is created.",
"validator": "VALID",
"required": "true",
"required": "false",
"sensitive": "false",
"expressionLanguageScope": "NONE",
"allowableValues": [
Expand All @@ -6928,14 +6932,18 @@
{
"value": "String",
"displayName": "String"
},
{
"value": "Guid",
"displayName": "Guid"
}
]
},
"Parent node namespace index": {
"name": "Parent node namespace index",
"description": "The index of the namespace of the parent node.",
"validator": "INTEGER_VALIDATOR",
"required": "true",
"description": "The index of the namespace of the parent node. Only used when a new node is created.",
"validator": "NON_NEGATIVE_INTEGER_VALIDATOR",
"required": "false",
"sensitive": "false",
"expressionLanguageScope": "NONE",
"defaultValue": "0"
Expand Down Expand Up @@ -6966,7 +6974,7 @@
},
"Target node ID type": {
"name": "Target node ID type",
"description": "ID type of target node. Allowed values are: Int, String.",
"description": "ID type of target node. Allowed values are: Int, String, Guid.",
"validator": "VALID",
"required": "true",
"sensitive": "false",
Expand Down
39 changes: 20 additions & 19 deletions PROCESSORS.md
View file
Open in desktop

Large diffs are not rendered by default.

6 changes: 4 additions & 2 deletions extensions/opc/include/BaseOPCProcessor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,10 +80,12 @@ class BaseOPCProcessor : public core::ProcessorImpl {
protected:
virtual bool reconnect();
void readPathReferenceTypes(core::ProcessContext& context, const std::string& node_id);
void parseIdType(core::ProcessContext& context, const core::PropertyReference& prop);
void parseNode(core::ProcessContext& context);

std::string node_id_;
int32_t namespace_idx_ = 0;
UA_UInt16 namespace_idx_ = 0;
bool path_node_id_resolved_ = false;
opc::NodeId node_;
opc::OPCNodeIDType id_type_{};

opc::ClientPtr connection_;
Expand Down
4 changes: 1 addition & 3 deletions extensions/opc/include/FetchOPCHistory.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ class FetchOPCHistory final : public BaseOPCProcessor {
EXTENSIONAPI static constexpr auto NameSpaceIndex =
core::PropertyDefinitionBuilder<>::createProperty("Namespace index")
.withDescription("The index of the namespace.")
.withValidator(core::StandardPropertyValidators::INTEGER_VALIDATOR)
.withValidator(core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR)
.withDefaultValue("0")
.isRequired(true)
.build();
Expand Down Expand Up @@ -151,8 +151,6 @@ class FetchOPCHistory final : public BaseOPCProcessor {
std::optional<std::chrono::system_clock::time_point> end_timestamp_;
uint64_t batch_size_ = 0;
std::shared_ptr<core::RecordSetWriter> record_set_writer_;
opc::NodeId node_;
bool path_node_id_resolved_ = false;
};

} // namespace org::apache::nifi::minifi::processors
13 changes: 7 additions & 6 deletions extensions/opc/include/FetchOPCProcessor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,18 +73,18 @@ class FetchOPCProcessor final : public BaseOPCProcessor {

EXTENSIONAPI static constexpr const char* Description = "Fetches OPC-UA node";

EXTENSIONAPI static constexpr auto NodeIDType = core::PropertyDefinitionBuilder<3>::createProperty("Node ID type")
EXTENSIONAPI static constexpr auto NodeIDType = core::PropertyDefinitionBuilder<magic_enum::enum_count<opc::OPCNodeIDType>()>::createProperty("Node ID type")
.withDescription("Specifies the type of the provided node ID")
.isRequired(true)
.withAllowedValues({"Path", "Int", "String"})
.withAllowedValues(magic_enum::enum_names<opc::OPCNodeIDType>())
.build();
EXTENSIONAPI static constexpr auto NodeID = core::PropertyDefinitionBuilder<>::createProperty("Node ID")
.withDescription("Specifies the ID of the root node to traverse. In case of a Path Node ID Type, the path should be provided in the format of 'path/to/node'.")
.isRequired(true)
.build();
EXTENSIONAPI static constexpr auto NameSpaceIndex = core::PropertyDefinitionBuilder<>::createProperty("Namespace index")
.withDescription("The index of the namespace.")
.withValidator(core::StandardPropertyValidators::INTEGER_VALIDATOR)
.withValidator(core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR)
.withDefaultValue("0")
.isRequired(true)
.build();
Expand Down Expand Up @@ -115,6 +115,7 @@ class FetchOPCProcessor final : public BaseOPCProcessor {
EXTENSIONAPI static constexpr auto Relationships = std::array{Success, Failure};

EXTENSIONAPI static constexpr auto NodeIDAttr = core::OutputAttributeDefinition<>{"NodeID", { Success }, "ID of the node."};
EXTENSIONAPI static constexpr auto NamespaceIndexAttr = core::OutputAttributeDefinition<>{"Namespace index", {Success}, "Namespace index of the node."};
EXTENSIONAPI static constexpr auto NodeIDTypeAttr = core::OutputAttributeDefinition<>{"NodeID type", { Success }, "Type of the node ID."};
EXTENSIONAPI static constexpr auto BrowsenameAttr = core::OutputAttributeDefinition<>{"Browsename", { Success }, "The browse name of the node."};
EXTENSIONAPI static constexpr auto FullPathAttr = core::OutputAttributeDefinition<>{"Full path", { Success }, "The full path of the node."};
Expand All @@ -123,8 +124,8 @@ class FetchOPCProcessor final : public BaseOPCProcessor {
EXTENSIONAPI static constexpr auto TypenameAttr = core::OutputAttributeDefinition<>{"Typename", { Success }, "The type name of the node data."};
EXTENSIONAPI static constexpr auto DatasizeAttr = core::OutputAttributeDefinition<>{"Datasize", { Success }, "The size of the node data."};

EXTENSIONAPI static constexpr auto OutputAttributes = std::array<core::OutputAttributeReference, 7> {NodeIDAttr, NodeIDTypeAttr, BrowsenameAttr, FullPathAttr, SourcetimestampAttr,
TypenameAttr, DatasizeAttr};
EXTENSIONAPI static constexpr auto OutputAttributes = std::array<core::OutputAttributeReference, 8> {NodeIDAttr,
NamespaceIndexAttr, NodeIDTypeAttr, BrowsenameAttr, FullPathAttr, SourcetimestampAttr, TypenameAttr, DatasizeAttr};

EXTENSIONAPI static constexpr bool SupportsDynamicProperties = false;
EXTENSIONAPI static constexpr bool SupportsDynamicRelationships = false;
Expand All @@ -149,7 +150,7 @@ class FetchOPCProcessor final : public BaseOPCProcessor {

uint64_t max_depth_ = 0;
LazyModeOptions lazy_mode_ = LazyModeOptions::Off;
std::vector<opc::NodeId> translated_node_ids_; // Only used when user provides path, path->nodeid translation is only done once
std::vector<opc::NodeId> translated_node_ids_; // Only used when user provides path; cached translation, re-resolved on (re)connect (see path_node_id_resolved_)
};

} // namespace org::apache::nifi::minifi::processors
3 changes: 3 additions & 0 deletions extensions/opc/include/OPCCommon.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
#include <map>
#include <vector>
#include <memory>
#include <expected>
#include <string_view>
#include <utility>
#include <optional>
Expand Down Expand Up @@ -202,5 +203,7 @@ void logFunc(void *context, UA_LogLevel level, UA_LogCategory category, const ch

std::optional<UA_UInt32> mapOpcReferenceType(const std::string& ref_type);

std::expected<opc::NodeId, std::string> buildNodeId(opc::OPCNodeIDType id_type, UA_UInt16 namespace_idx, const std::string& node_id);

} // namespace org::apache::nifi::minifi::opc

20 changes: 8 additions & 12 deletions extensions/opc/include/PutOPCProcessor.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,28 +41,25 @@ class PutOPCProcessor final : public BaseOPCProcessor {
public:
EXTENSIONAPI static constexpr const char* Description = "Creates/updates OPC nodes";

EXTENSIONAPI static constexpr auto ParentNodeIDType = core::PropertyDefinitionBuilder<3>::createProperty("Parent node ID type")
.withDescription("Specifies the type of the provided node ID")
.isRequired(true)
.withAllowedValues({"Path", "Int", "String"})
EXTENSIONAPI static constexpr auto ParentNodeIDType = core::PropertyDefinitionBuilder<magic_enum::enum_count<opc::OPCNodeIDType>()>::createProperty("Parent node ID type")
.withDescription("Specifies the type of the provided node ID. Only required when a new node is created.")
.withAllowedValues(magic_enum::enum_names<opc::OPCNodeIDType>())
.build();
EXTENSIONAPI static constexpr auto ParentNodeID = core::PropertyDefinitionBuilder<>::createProperty("Parent node ID")
.withDescription("Specifies the ID of the root node to traverse")
.isRequired(true)
.withDescription("Specifies the ID of the root node to traverse. Only required when a new node is created.")
.build();
EXTENSIONAPI static constexpr auto ParentNameSpaceIndex = core::PropertyDefinitionBuilder<>::createProperty("Parent node namespace index")
.withDescription("The index of the namespace of the parent node.")
.withValidator(core::StandardPropertyValidators::INTEGER_VALIDATOR)
.withDescription("The index of the namespace of the parent node. Only used when a new node is created.")
.withValidator(core::StandardPropertyValidators::UNSIGNED_INTEGER_VALIDATOR)
.withDefaultValue("0")
.isRequired(true)
.build();
EXTENSIONAPI static constexpr auto ValueType = core::PropertyDefinitionBuilder<magic_enum::enum_count<opc::OPCNodeDataType>()>::createProperty("Value type")
.withDescription("Set the OPC value type of the created nodes")
.withAllowedValues(magic_enum::enum_names<opc::OPCNodeDataType>())
.isRequired(true)
.build();
EXTENSIONAPI static constexpr auto TargetNodeIDType = core::PropertyDefinitionBuilder<>::createProperty("Target node ID type")
.withDescription("ID type of target node. Allowed values are: Int, String.")
.withDescription("ID type of target node. Allowed values are: Int, String, Guid.")
.supportsExpressionLanguage(true)
.isRequired(true)
.build();
Expand Down Expand Up @@ -117,14 +114,13 @@ class PutOPCProcessor final : public BaseOPCProcessor {
void initialize() override;

private:
bool readParentNodeId();
std::expected<std::pair<bool, opc::NodeId>, std::string> configureTargetNode(core::ProcessContext& context, core::FlowFile& flow_file) const;
void updateNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessSession& session, const std::shared_ptr<core::FlowFile>& flow_file) const;
void createNode(const UA_NodeId& target_node, const std::string& contentstr, core::ProcessContext& context, core::ProcessSession& session, const std::shared_ptr<core::FlowFile>& flow_file) const;

opc::NodeId parent_node_id_;
opc::OPCNodeDataType node_data_type_{};
UA_UInt32 create_node_reference_type_ = UA_NS0ID_HASCOMPONENT;
bool parent_node_defined_ = false;
};

} // namespace org::apache::nifi::minifi::processors
31 changes: 22 additions & 9 deletions extensions/opc/src/BaseOPCProcessor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,10 +105,15 @@ bool BaseOPCProcessor::reconnect() {
return false;
}
logger_->log_debug("Successfully connected.");

// A new or re-established connection may be to a restarted server that can reassign node ids,
// so if path is used to identify node id, the cached node id should be invalidated
path_node_id_resolved_ = false;
return true;
}

void BaseOPCProcessor::readPathReferenceTypes(core::ProcessContext& context, const std::string& node_id) {
path_reference_types_.clear();
const auto value = context.getProperty(PathReferenceTypes).value_or("");
if (value.empty()) {
return;
Expand All @@ -126,16 +131,24 @@ void BaseOPCProcessor::readPathReferenceTypes(core::ProcessContext& context, con
}
}

void BaseOPCProcessor::parseIdType(core::ProcessContext& context, const core::PropertyReference& prop) {
id_type_ = utils::parseEnumProperty<opc::OPCNodeIDType>(context, prop);

if (id_type_ == opc::OPCNodeIDType::Int) {
try {
static_cast<void>(std::stoi(node_id_));
} catch(const std::exception&) {
auto error_msg = utils::string::join_pack(node_id_, " cannot be used as an int type node ID");
throw Exception(PROCESS_SCHEDULE_EXCEPTION, error_msg);
void BaseOPCProcessor::parseNode(core::ProcessContext& context) {
switch (id_type_) {
case opc::OPCNodeIDType::String:
case opc::OPCNodeIDType::Int:
case opc::OPCNodeIDType::Guid: {
if (auto result = opc::buildNodeId(id_type_, namespace_idx_, node_id_)) {
node_ = std::move(*result);
} else {
throw Exception(PROCESS_SCHEDULE_EXCEPTION, result.error());
}
break;
}
case opc::OPCNodeIDType::Path:
readPathReferenceTypes(context, node_id_);
path_node_id_resolved_ = false;
break;
default:
throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("Unsupported Node ID type: {}", magic_enum::enum_name(id_type_)));
}
}

Expand Down
28 changes: 3 additions & 25 deletions extensions/opc/src/FetchOPCHistory.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -281,31 +281,9 @@ void FetchOPCHistory::onSchedule(core::ProcessContext& context, core::ProcessSes
logger_->log_trace("FetchOPCHistory::onSchedule");
BaseOPCProcessor::onSchedule(context, factory);
node_id_ = utils::parseProperty(context, NodeID);
parseIdType(context, NodeIDType);
namespace_idx_ = gsl::narrow<int32_t>(utils::parseI64Property(context, NameSpaceIndex));

switch (id_type_) {
case opc::OPCNodeIDType::String:
node_ = opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_idx_, node_id_.c_str())};
break;
case opc::OPCNodeIDType::Int:
node_ = opc::NodeId{UA_NODEID_NUMERIC(namespace_idx_, std::stoi(node_id_))};
break;
case opc::OPCNodeIDType::Guid: {
UA_Guid guid;
if (UA_Guid_parse(&guid, UA_STRING(const_cast<char*>(node_id_.c_str()))) != UA_STATUSCODE_GOOD) {
throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("{} cannot be used as a GUID type node ID", node_id_));
}
node_ = opc::NodeId{UA_NODEID_GUID(namespace_idx_, guid)};
break;
}
case opc::OPCNodeIDType::Path:
readPathReferenceTypes(context, node_id_);
path_node_id_resolved_ = false;
break;
default:
throw Exception(PROCESS_SCHEDULE_EXCEPTION, fmt::format("Unsupported Node ID type: {}", magic_enum::enum_name(id_type_)));
}
id_type_ = utils::parseEnumProperty<opc::OPCNodeIDType>(context, NodeIDType);
namespace_idx_ = gsl::narrow<UA_UInt16>(utils::parseU64Property(context, NameSpaceIndex));
parseNode(context);

history_type_ = utils::parseEnumProperty<opc::HistoryReadTypeOption>(context, HistoryReadType);
start_timestamp_ = utils::parseOptionalProperty(context, StartTimestamp) | utils::andThen(utils::timeutils::parseRfc3339);
Expand Down
29 changes: 8 additions & 21 deletions extensions/opc/src/FetchOPCProcessor.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,17 +44,13 @@ void FetchOPCProcessor::onSchedule(core::ProcessContext& context, core::ProcessS
BaseOPCProcessor::onSchedule(context, factory);

node_id_ = utils::parseProperty(context, NodeID);
max_depth_ = utils::parseU64Property(context, MaxDepth);

parseIdType(context, NodeIDType);
id_type_ = utils::parseEnumProperty<opc::OPCNodeIDType>(context, NodeIDType);
namespace_idx_ = gsl::narrow<UA_UInt16>(utils::parseU64Property(context, NameSpaceIndex));
parseNode(context);

namespace_idx_ = gsl::narrow<int32_t>(utils::parseI64Property(context, NameSpaceIndex));
max_depth_ = utils::parseU64Property(context, MaxDepth);

lazy_mode_ = utils::parseEnumProperty<LazyModeOptions>(context, Lazy);

if (id_type_ == opc::OPCNodeIDType::Path) {
readPathReferenceTypes(context, node_id_);
}
}

void FetchOPCProcessor::onTrigger(core::ProcessContext& context, core::ProcessSession& session) {
Expand All @@ -77,26 +73,17 @@ void FetchOPCProcessor::onTrigger(core::ProcessContext& context, core::ProcessSe
};

if (id_type_ != opc::OPCNodeIDType::Path) {
const auto namespace_index = gsl::narrow_cast<UA_UInt16>(namespace_idx_);
opc::NodeId my_id;
if (id_type_ == opc::OPCNodeIDType::Int) {
my_id = opc::NodeId{UA_NODEID_NUMERIC(namespace_index, std::stoi(node_id_))};
} else if (id_type_ == opc::OPCNodeIDType::String) {
my_id = opc::NodeId{UA_NODEID_STRING_ALLOC(namespace_index, node_id_.c_str())};
} else {
logger_->log_error("Unhandled id type: '{}'. No flowfiles are generated.", magic_enum::enum_underlying(id_type_));
context.yield();
return;
}
connection_->traverse(my_id, found_cb, "", max_depth_);
connection_->traverse(node_, found_cb, "", max_depth_);
} else {
if (translated_node_ids_.empty()) {
if (!path_node_id_resolved_) {
translated_node_ids_.clear();
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 {} to node id, no flow files will be generated ({})", node_id_.c_str(), UA_StatusCode_name(sc));
context.yield();
return;
}
path_node_id_resolved_ = true;
}
for (auto& node_id : translated_node_ids_) {
connection_->traverse(node_id, found_cb, node_id_, max_depth_);
Expand Down
Loading