wirenboard / wb-mqtt-opcua / 2

Committed 31 Jul 2025 05:54PM UTC coverage: 29.074% (+0.2%) from 28.86%

Build # 2

Build Type

push

github

Committed by

u236

Commit Message

try to create nodes in mqtt driver callback

Run Details

104 of 294 branches covered (35.37%)

0 of 21 new or added lines in 1 file covered. (0.0%)

26 existing lines in 1 file now uncovered.

157 of 540 relevant lines covered (29.07%)

0.88 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/OPCUAServer.cpp

#include "OPCUAServer.h"

#include <open62541/plugin/accesscontrol_default.h>
#include <open62541/server.h>
#include <open62541/server_config_default.h>
#include <open62541/statuscodes.h>

#include <functional>
#include <stdexcept>
#include <vector>

#include "log.h"

#define LOG(logger) ::logger.Log() << "[OPCUA] "

namespace
{
    const char* LogCategoryNames[7] =
        {"network", "channel", "session", "server", "client", "userland", "securitypolicy"};

    void PrintLogMessage(WBMQTT::TLogger& logger, UA_LogCategory category, const char* msg, va_list args)
    {
        va_list args2;
        va_copy(args2, args);
        auto bufSize = 1 + vsnprintf(nullptr, 0, msg, args);
        std::string str(bufSize, '\0');
        vsnprintf(&str[0], bufSize, msg, args2);
        va_end(args2);
        logger.Log() << "[OPCUA] " << LogCategoryNames[category] << ": " << str;
    }

    extern "C" {
    void Log(void* context, UA_LogLevel level, UA_LogCategory category, const char* msg, va_list args)
    {
        switch (level) {
            case UA_LOGLEVEL_TRACE:
            case UA_LOGLEVEL_DEBUG:
                PrintLogMessage(Debug, category, msg, args);
                break;
            case UA_LOGLEVEL_INFO:
                PrintLogMessage(Info, category, msg, args);
                break;
            case UA_LOGLEVEL_WARNING:
                PrintLogMessage(Warn, category, msg, args);
                break;
            case UA_LOGLEVEL_ERROR:
            case UA_LOGLEVEL_FATAL:
                PrintLogMessage(Error, category, msg, args);
                break;
        }
    }

    void LogClear(void* logContext)
    {}

    UA_StatusCode ReadVariableCallback(UA_Server* sserver,
                                       const UA_NodeId* ssessionId,
                                       void* ssessionContext,
                                       const UA_NodeId* snodeId,
                                       void* snodeContext,
                                       UA_Boolean ssourceTimeStamp,
                                       const UA_NumericRange* range,
                                       UA_DataValue* dataValue);

    UA_StatusCode WriteVariableCallback(UA_Server* server,
                                        const UA_NodeId* sessionId,
                                        void* sessionContext,
                                        const UA_NodeId* nodeId,
                                        void* nodeContext,
                                        const UA_NumericRange* range,
                                        const UA_DataValue* data);
    }

    UA_Logger MakeLogger()
    {
        UA_Logger logger = {Log, nullptr, LogClear};
        return logger;
    }

    UA_Byte GetAccessLevel(WBMQTT::PDeviceDriver driver, const std::string& deviceName, const std::string& controlName)
    {
        auto tx = driver->BeginTx();
        auto dev = tx->GetDevice(deviceName);
        if (dev) {
            auto ctrl = dev->GetControl(controlName);
            if (ctrl) {
                return (ctrl->IsReadonly() ? UA_ACCESSLEVELMASK_READ
                                           : UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE);
            }
        }
        return UA_ACCESSLEVELMASK_READ;
    }

    WBMQTT::PControl GetControl(WBMQTT::PDriverTx& tx, const std::string& nodeIdName)
    {
        auto components = WBMQTT::StringSplit(nodeIdName, "/");
        if (components.size() < 2) {
            return WBMQTT::PControl();
        }

        auto dev = tx->GetDevice(components[0]);
        if (!dev) {
            return WBMQTT::PControl();
        }
        return dev->GetControl(components[1]);
    }

    void SetVariableAttributes(UA_VariableAttributes& attr,
                               WBMQTT::PDeviceDriver driver,
                               const std::string& deviceName,
                               const std::string& controlName)
    {
        attr.accessLevel = GetAccessLevel(driver, deviceName, controlName);
        attr.displayName = UA_LOCALIZEDTEXT((char*)"en-US", (char*)controlName.c_str());
        attr.valueRank = UA_VALUERANK_SCALAR;
        attr.dataType = UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATATYPE);
        auto tx = driver->BeginTx();
        auto dev = tx->GetDevice(deviceName);
        if (dev) {
            auto ctrl = dev->GetControl(controlName);
            if (ctrl) {
                try {
                    auto v = ctrl->GetValue();
                    if (v.Is<bool>()) {
                        attr.dataType = UA_NODEID_NUMERIC(0, UA_NS0ID_BOOLEAN);
                        return;
                    }
                    if (v.Is<double>()) {
                        attr.dataType = UA_NODEID_NUMERIC(0, UA_NS0ID_DOUBLE);
                        return;
                    }
                    return;
                } catch (...) {
                }
            }
        }
        LOG(Error) << "Can't get data type for node '" + deviceName + "/" + controlName +
                          "'. Fallback to BaseDataType.";
    }

    void ConfigureOpcUaServer(UA_ServerConfig* serverCfg, const OPCUA::TServerConfig& config)
    {
        serverCfg->logger = MakeLogger();

        UA_ServerConfig_setBasics(serverCfg);
        serverCfg->allowEmptyVariables = UA_RULEHANDLING_ACCEPT;

        UA_BuildInfo_clear(&serverCfg->buildInfo);
        UA_ApplicationDescription_clear(&serverCfg->applicationDescription);
        serverCfg->applicationDescription.applicationUri = UA_STRING_ALLOC("urn:wb-mqtt-opcua.server.application");
        serverCfg->applicationDescription.productUri = UA_STRING_ALLOC("https://wirenboard.com");
        serverCfg->applicationDescription.applicationName =
            UA_LOCALIZEDTEXT_ALLOC("en", "Wiren Board MQTT to OPC UA gateway");
        serverCfg->applicationDescription.applicationType = UA_APPLICATIONTYPE_SERVER;

        if (!config.BindIp.empty()) {
            UA_String_clear(&serverCfg->customHostname);
            serverCfg->customHostname = UA_String_fromChars(config.BindIp.c_str());
        }

        auto res = UA_ServerConfig_addNetworkLayerTCP(serverCfg, config.BindPort, 0, 0);
        if (res != UA_STATUSCODE_GOOD) {
            throw std::runtime_error(std::string("OPC UA network layer configuration failed: ") +
                                     UA_StatusCode_name(res));
        }

        res = UA_ServerConfig_addSecurityPolicyNone(serverCfg, nullptr);
        if (res != UA_STATUSCODE_GOOD) {
            throw std::runtime_error(std::string("OPC UA security policy addition failed: ") + UA_StatusCode_name(res));
        }

        res = UA_AccessControl_default(serverCfg,
                                       true,
                                       &serverCfg->securityPolicies[serverCfg->securityPoliciesSize - 1].policyUri,
                                       0,
                                       nullptr);
        if (res != UA_STATUSCODE_GOOD) {
            throw std::runtime_error(std::string("OPC UA access control configuration failed: ") +
                                     UA_StatusCode_name(res));
        }

        res = UA_ServerConfig_addEndpoint(serverCfg, UA_SECURITY_POLICY_NONE_URI, UA_MESSAGESECURITYMODE_NONE);
        if (res != UA_STATUSCODE_GOOD) {
            throw std::runtime_error(std::string("OPC UA server endpoint allocation failed: ") +
                                     UA_StatusCode_name(res));
        }
    }

    /**! Basic gateway implementation.
     *   The server creates ObjectNodes for groups from config and VariableNodes for MQTT controls.
     *   OPC UA variable node id is DEVICE/CONTROL pair string.
     *   The server translates writes to VariableNodes to publishing into appropriate "on" topics.
     */
    class TServerImpl: public OPCUA::IServer
    {
        UA_Server* Server;
        volatile UA_Boolean IsRunning;
        std::thread ServerThread;
        WBMQTT::PDeviceDriver Driver;

        UA_NodeId CreateObjectNode(const std::string& nodeName, const OPCUA::TVariableNodesConfig& variableNodes)
        {
            UA_NodeId nodeId = UA_NODEID_STRING(1, (char*)nodeName.c_str());
            UA_ObjectAttributes oAttr = UA_ObjectAttributes_default;
            oAttr.displayName = UA_LOCALIZEDTEXT((char*)"en-US", (char*)nodeName.c_str());
            auto res = UA_Server_addObjectNode(Server,
                                               nodeId,
                                               UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
                                               UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES),
                                               UA_QUALIFIEDNAME(1, (char*)nodeName.c_str()),
                                               UA_NODEID_NUMERIC(0, UA_NS0ID_BASEOBJECTTYPE),
                                               oAttr,
                                               nullptr,
                                               nullptr);
            if (res != UA_STATUSCODE_GOOD) {
                throw std::runtime_error("Object node '" + nodeName + "' creation failed: " + UA_StatusCode_name(res));
            }
            return nodeId;
        }

        void CreateVariableNode(const UA_NodeId& parentNodeId,
                                const OPCUA::TVariableNodeConfig& variableNode,
                                WBMQTT::PDeviceDriver driver)
        {
            auto components = WBMQTT::StringSplit(variableNode.DeviceControlPair, "/");

            UA_VariableAttributes oAttr = UA_VariableAttributes_default;
            SetVariableAttributes(oAttr, driver, components[0], components[1]);

            UA_DataSource dataSource;
            dataSource.read = ReadVariableCallback;
            dataSource.write = WriteVariableCallback;

            UA_NodeId nodeId = UA_NODEID_STRING(1, (char*)variableNode.DeviceControlPair.c_str());

            auto res = UA_Server_addDataSourceVariableNode(Server,
                                                           nodeId,
                                                           parentNodeId,
                                                           UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT),
                                                           UA_QUALIFIEDNAME(1, (char*)components[1].c_str()),
                                                           UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE),
                                                           oAttr,
                                                           dataSource,
                                                           this,
                                                           nullptr);
            if (res != UA_STATUSCODE_GOOD) {
                throw std::runtime_error("Variable node '" + variableNode.DeviceControlPair +
                                         "' creation failed: " + UA_StatusCode_name(res));
            }
        }

    public:
        TServerImpl(const OPCUA::TServerConfig& config, WBMQTT::PDeviceDriver driver)
            : Server(UA_Server_new()),
              IsRunning(true),
              Driver(driver)
        {
            if (!Server) {
                throw std::runtime_error("OPC UA server initilization failed");
            }

            Driver->On<WBMQTT::TControlValueEvent>([&](const WBMQTT::TControlValueEvent& event) {
                if (event.RawValue.empty()) {
                    return;
                }
                auto it = config.ObjectNodes.find(event.Control->GetDevice()->GetId());
                if (it == config.ObjectNodes.end()) {
                    return;
                }

                auto browseName = UA_QUALIFIEDNAME(1, (char*)it->first.c_str());
                auto res = UA_Server_browseSimplifiedBrowsePath(Server,
                                                                UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
                                                                1,
                                                                &browseName);
                auto parentNodeId = res.statusCode == UA_STATUSCODE_GOOD ? res.targets[0].targetId.nodeId
                                                                         : CreateObjectNode(it->first, it->second);
                std::string nodeName = it->first + "/" + event.Control->GetId();
                for (auto& valueNode: it->second) {
                    if (valueNode.DeviceControlPair != nodeName) {
                        continue;
                    }
                    browseName = UA_QUALIFIEDNAME(1, (char*)event.Control->GetId().c_str());
                    res = UA_Server_browseSimplifiedBrowsePath(Server, parentNodeId, 1, &browseName);
                    if (res.statusCode != UA_STATUSCODE_GOOD) {
                        CreateVariableNode(parentNodeId, valueNode, Driver);
                        break;
                    }
                }
            });

            // Load external controls
            std::vector<std::string> deviceIds;
            for (const auto& device: config.ObjectNodes) {
                LOG(Debug) << "'" << device.first << "' is added to filter";
                deviceIds.emplace_back(device.first);
            }
            Driver->SetFilter(WBMQTT::GetDeviceListFilter(deviceIds));
            Driver->WaitForReady();

            // Setup OPC UA server and nodes
            ConfigureOpcUaServer(UA_Server_getConfig(Server), config);

            // for (auto& node: config.ObjectNodes) {
            //     auto nodeId = CreateObjectNode(node.first, node.second);
            //     for (auto& vn: node.second) {
            //         CreateVariableNode(nodeId, vn, driver);
            //     }
            // }

            ServerThread = std::thread([this]() {
                auto res = UA_Server_run(Server, &IsRunning);
                if (res != UA_STATUSCODE_GOOD) {
                    LOG(Error) << UA_StatusCode_name(res);
                    exit(1);
                }
            });
        }

        ~TServerImpl()
        {
            if (IsRunning) {
                IsRunning = false;
                if (ServerThread.joinable()) {
                    ServerThread.join();
                }
            }
            if (Server) {
                UA_Server_delete(Server);
            }
        }

        UA_StatusCode writeVariable(const UA_NodeId* snodeId, const UA_DataValue* dataValue)
        {
            std::string nodeIdName((const char*)snodeId->identifier.string.data, snodeId->identifier.string.length);
            auto tx = Driver->BeginTx();
            auto ctrl = GetControl(tx, nodeIdName);
            if (!ctrl || ctrl->IsReadonly()) {
                LOG(Error) << "Variable node '" + nodeIdName + "' writing failed. "
                           << (ctrl ? "It is read only" : "It is not presented in MQTT");
                return UA_STATUSCODE_BADDEVICEFAILURE;
            }
            try {
                if (dataValue->hasValue) {
                    if (UA_Variant_hasScalarType(&dataValue->value, &UA_TYPES[UA_TYPES_BOOLEAN])) {
                        auto value = *(UA_Boolean*)dataValue->value.data;
                        ctrl->SetValue(tx, value).Sync();
                        LOG(Info) << "Variable node '" + nodeIdName + "' = " << value;
                        return UA_STATUSCODE_GOOD;
                    }
                    if (UA_Variant_hasScalarType(&dataValue->value, &UA_TYPES[UA_TYPES_DOUBLE])) {
                        auto value = *(UA_Double*)dataValue->value.data;
                        ctrl->SetValue(tx, value).Sync();
                        LOG(Info) << "Variable node '" + nodeIdName + "' = " << value;
                        return UA_STATUSCODE_GOOD;
                    }
                    if (UA_Variant_hasScalarType(&dataValue->value, &UA_TYPES[UA_TYPES_STRING])) {
                        auto value = (char*)((UA_String*)dataValue->value.data)->data;
                        ctrl->SetRawValue(tx, value).Sync();
                        LOG(Info) << "Variable node '" + nodeIdName + "' = " << value;
                        return UA_STATUSCODE_GOOD;
                    }
                }
                return UA_STATUSCODE_BADDATATYPEIDUNKNOWN;
            } catch (const std::exception& e) {
                LOG(Error) << "Variable node '" + nodeIdName + "' write error: " << e.what();
                return UA_STATUSCODE_BADDEVICEFAILURE;
            }
        }

        UA_StatusCode readVariable(const UA_NodeId* snodeId, UA_DataValue* dataValue)
        {
            std::string nodeIdName((const char*)snodeId->identifier.string.data, snodeId->identifier.string.length);
            auto tx = Driver->BeginTx();
            auto ctrl = GetControl(tx, nodeIdName);
            if (!ctrl) {
                LOG(Error) << "Control is not found '" + nodeIdName + "'";
                dataValue->hasStatus = true;
                dataValue->status = UA_STATUSCODE_BADNOCOMMUNICATION;
                return UA_STATUSCODE_GOOD;
            }
            try {
                dataValue->hasStatus = true;
                if (ctrl->GetError().find("r") != std::string::npos) {
                    dataValue->status = UA_STATUSCODE_BAD;
                } else {
                    dataValue->status = UA_STATUSCODE_GOOD;
                }
                auto v = ctrl->GetValue();
                if (v.Is<bool>()) {
                    auto value = v.As<bool>();
                    UA_Variant_setScalarCopy(&dataValue->value, &value, &UA_TYPES[UA_TYPES_BOOLEAN]);
                } else {
                    if (v.Is<double>()) {
                        auto value = v.As<double>();
                        UA_Variant_setScalarCopy(&dataValue->value, &value, &UA_TYPES[UA_TYPES_DOUBLE]);
                    } else {
                        UA_String stringValue = UA_String_fromChars((char*)v.As<std::string>().c_str());
                        UA_Variant_setScalarCopy(&dataValue->value, &stringValue, &UA_TYPES[UA_TYPES_STRING]);
                        UA_String_clear(&stringValue);
                    }
                }
                dataValue->hasValue = true;
            } catch (const std::exception& e) {
                LOG(Error) << "Variable node '" + nodeIdName + "' read error: " << e.what();
                dataValue->hasStatus = true;
                dataValue->status = UA_STATUSCODE_BADNOCOMMUNICATION;
            }
            return UA_STATUSCODE_GOOD;
        }
    };

    extern "C" {
    UA_StatusCode ReadVariableCallback(UA_Server* sserver,
                                       const UA_NodeId* ssessionId,
                                       void* ssessionContext,
                                       const UA_NodeId* snodeId,
                                       void* snodeContext,
                                       UA_Boolean ssourceTimeStamp,
                                       const UA_NumericRange* range,
                                       UA_DataValue* dataValue)
    {
        TServerImpl* server = (TServerImpl*)(snodeContext);
        return server->readVariable(snodeId, dataValue);
    }

    UA_StatusCode WriteVariableCallback(UA_Server* server,
                                        const UA_NodeId* sessionId,
                                        void* sessionContext,
                                        const UA_NodeId* nodeId,
                                        void* nodeContext,
                                        const UA_NumericRange* range,
                                        const UA_DataValue* data)
    {
        TServerImpl* s = (TServerImpl*)(nodeContext);
        return s->writeVariable(nodeId, data);
    }
    }
}

std::unique_ptr<OPCUA::IServer> OPCUA::MakeServer(const OPCUA::TServerConfig& config, WBMQTT::PDeviceDriver driver)
{
    return std::unique_ptr<OPCUA::IServer>(new TServerImpl(config, driver));
}

1	#include "OPCUAServer.h"
2
3	#include <open62541/plugin/accesscontrol_default.h>
4	#include <open62541/server.h>
5	#include <open62541/server_config_default.h>
6	#include <open62541/statuscodes.h>
7
8	#include <functional>
9	#include <stdexcept>
10	#include <vector>
11
12	#include "log.h"
13
14	#define LOG(logger) ::logger.Log() << "[OPCUA] "
15
16	namespace
17	{
18	const char* LogCategoryNames[7] =
19	{"network", "channel", "session", "server", "client", "userland", "securitypolicy"};
20
21	void PrintLogMessage(WBMQTT::TLogger& logger, UA_LogCategory category, const char* msg, va_list args)	×
22	{
23	va_list args2;
24	va_copy(args2, args);	×
25	auto bufSize = 1 + vsnprintf(nullptr, 0, msg, args);	×
26	std::string str(bufSize, '\0');	×
27	vsnprintf(&str[0], bufSize, msg, args2);	×
28	va_end(args2);	×
29	logger.Log() << "[OPCUA] " << LogCategoryNames[category] << ": " << str;	×
30	}
31
32	extern "C" {
33	void Log(void* context, UA_LogLevel level, UA_LogCategory category, const char* msg, va_list args)	×
34	{
35	switch (level) {	×
36	case UA_LOGLEVEL_TRACE:	×
37	case UA_LOGLEVEL_DEBUG:
38	PrintLogMessage(Debug, category, msg, args);	×
39	break;	×
40	case UA_LOGLEVEL_INFO:	×
41	PrintLogMessage(Info, category, msg, args);	×
42	break;	×
43	case UA_LOGLEVEL_WARNING:	×
44	PrintLogMessage(Warn, category, msg, args);	×
45	break;	×
46	case UA_LOGLEVEL_ERROR:	×
47	case UA_LOGLEVEL_FATAL:
48	PrintLogMessage(Error, category, msg, args);	×
49	break;	×
50	}
51	}
52
53	void LogClear(void* logContext)	×
54	{}
55
56	UA_StatusCode ReadVariableCallback(UA_Server* sserver,
57	const UA_NodeId* ssessionId,
58	void* ssessionContext,
59	const UA_NodeId* snodeId,
60	void* snodeContext,
61	UA_Boolean ssourceTimeStamp,
62	const UA_NumericRange* range,
63	UA_DataValue* dataValue);
64
65	UA_StatusCode WriteVariableCallback(UA_Server* server,
66	const UA_NodeId* sessionId,
67	void* sessionContext,
68	const UA_NodeId* nodeId,
69	void* nodeContext,
70	const UA_NumericRange* range,
71	const UA_DataValue* data);
72	}
73
74	UA_Logger MakeLogger()	×
75	{
76	UA_Logger logger = {Log, nullptr, LogClear};	×
77	return logger;	×
78	}
79
80	UA_Byte GetAccessLevel(WBMQTT::PDeviceDriver driver, const std::string& deviceName, const std::string& controlName)	×
81	{
82	auto tx = driver->BeginTx();	×
83	auto dev = tx->GetDevice(deviceName);	×
84	if (dev) {	×
85	auto ctrl = dev->GetControl(controlName);	×
86	if (ctrl) {	×
87	return (ctrl->IsReadonly() ? UA_ACCESSLEVELMASK_READ	×
88	: UA_ACCESSLEVELMASK_READ \| UA_ACCESSLEVELMASK_WRITE);	×
89	}
90	}
91	return UA_ACCESSLEVELMASK_READ;	×
92	}
93
94	WBMQTT::PControl GetControl(WBMQTT::PDriverTx& tx, const std::string& nodeIdName)	×
95	{
96	auto components = WBMQTT::StringSplit(nodeIdName, "/");	×
97	if (components.size() < 2) {	×
98	return WBMQTT::PControl();	×
99	}
100
101	auto dev = tx->GetDevice(components[0]);	×
102	if (!dev) {	×
103	return WBMQTT::PControl();	×
104	}
105	return dev->GetControl(components[1]);	×
106	}
107
108	void SetVariableAttributes(UA_VariableAttributes& attr,	×
109	WBMQTT::PDeviceDriver driver,
110	const std::string& deviceName,
111	const std::string& controlName)
112	{
113	attr.accessLevel = GetAccessLevel(driver, deviceName, controlName);	×
114	attr.displayName = UA_LOCALIZEDTEXT((char)"en-US", (char)controlName.c_str());	×
115	attr.valueRank = UA_VALUERANK_SCALAR;	×
116	attr.dataType = UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATATYPE);	×
117	auto tx = driver->BeginTx();	×
118	auto dev = tx->GetDevice(deviceName);	×
119	if (dev) {	×
120	auto ctrl = dev->GetControl(controlName);	×
121	if (ctrl) {	×
122	try {
123	auto v = ctrl->GetValue();	×
124	if (v.Is<bool>()) {	×
125	attr.dataType = UA_NODEID_NUMERIC(0, UA_NS0ID_BOOLEAN);	×
126	return;	×
127	}
128	if (v.Is<double>()) {	×
129	attr.dataType = UA_NODEID_NUMERIC(0, UA_NS0ID_DOUBLE);	×
130	return;	×
131	}
132	return;	×
133	} catch (...) {	×
134	}
135	}
136	}
137	LOG(Error) << "Can't get data type for node '" + deviceName + "/" + controlName +	×
138	"'. Fallback to BaseDataType.";	×
139	}
140
141	void ConfigureOpcUaServer(UA_ServerConfig* serverCfg, const OPCUA::TServerConfig& config)	×
142	{
143	serverCfg->logger = MakeLogger();	×
144
145	UA_ServerConfig_setBasics(serverCfg);	×
146	serverCfg->allowEmptyVariables = UA_RULEHANDLING_ACCEPT;	×
147
148	UA_BuildInfo_clear(&serverCfg->buildInfo);	×
149	UA_ApplicationDescription_clear(&serverCfg->applicationDescription);	×
150	serverCfg->applicationDescription.applicationUri = UA_STRING_ALLOC("urn:wb-mqtt-opcua.server.application");	×
151	serverCfg->applicationDescription.productUri = UA_STRING_ALLOC("https://wirenboard.com");	×
152	serverCfg->applicationDescription.applicationName =
153	UA_LOCALIZEDTEXT_ALLOC("en", "Wiren Board MQTT to OPC UA gateway");	×
154	serverCfg->applicationDescription.applicationType = UA_APPLICATIONTYPE_SERVER;	×
155
156	if (!config.BindIp.empty()) {	×
157	UA_String_clear(&serverCfg->customHostname);	×
158	serverCfg->customHostname = UA_String_fromChars(config.BindIp.c_str());	×
159	}
160
161	auto res = UA_ServerConfig_addNetworkLayerTCP(serverCfg, config.BindPort, 0, 0);	×
162	if (res != UA_STATUSCODE_GOOD) {	×
163	throw std::runtime_error(std::string("OPC UA network layer configuration failed: ") +	×
164	UA_StatusCode_name(res));	×
165	}
166
167	res = UA_ServerConfig_addSecurityPolicyNone(serverCfg, nullptr);	×
168	if (res != UA_STATUSCODE_GOOD) {	×
169	throw std::runtime_error(std::string("OPC UA security policy addition failed: ") + UA_StatusCode_name(res));	×
170	}
171
172	res = UA_AccessControl_default(serverCfg,	×
173	true,
174	&serverCfg->securityPolicies[serverCfg->securityPoliciesSize - 1].policyUri,	×
175	0,
176	nullptr);
177	if (res != UA_STATUSCODE_GOOD) {	×
178	throw std::runtime_error(std::string("OPC UA access control configuration failed: ") +	×
179	UA_StatusCode_name(res));	×
180	}
181
182	res = UA_ServerConfig_addEndpoint(serverCfg, UA_SECURITY_POLICY_NONE_URI, UA_MESSAGESECURITYMODE_NONE);	×
183	if (res != UA_STATUSCODE_GOOD) {	×
184	throw std::runtime_error(std::string("OPC UA server endpoint allocation failed: ") +	×
185	UA_StatusCode_name(res));	×
186	}
187	}
188
189	/**! Basic gateway implementation.
190	* The server creates ObjectNodes for groups from config and VariableNodes for MQTT controls.
191	* OPC UA variable node id is DEVICE/CONTROL pair string.
192	* The server translates writes to VariableNodes to publishing into appropriate "on" topics.
193	*/
194	class TServerImpl: public OPCUA::IServer
195	{
196	UA_Server* Server;
197	volatile UA_Boolean IsRunning;
198	std::thread ServerThread;
199	WBMQTT::PDeviceDriver Driver;
200
201	UA_NodeId CreateObjectNode(const std::string& nodeName, const OPCUA::TVariableNodesConfig& variableNodes)	×
202	{
203	UA_NodeId nodeId = UA_NODEID_STRING(1, (char*)nodeName.c_str());	×
204	UA_ObjectAttributes oAttr = UA_ObjectAttributes_default;	×
205	oAttr.displayName = UA_LOCALIZEDTEXT((char)"en-US", (char)nodeName.c_str());	×
206	auto res = UA_Server_addObjectNode(Server,	×
207	nodeId,
208	UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
209	UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES),
210	UA_QUALIFIEDNAME(1, (char*)nodeName.c_str()),	×
211	UA_NODEID_NUMERIC(0, UA_NS0ID_BASEOBJECTTYPE),
212	oAttr,
213	nullptr,
214	nullptr);
215	if (res != UA_STATUSCODE_GOOD) {	×
216	throw std::runtime_error("Object node '" + nodeName + "' creation failed: " + UA_StatusCode_name(res));	×
217	}
218	return nodeId;	×
219	}
220
221	void CreateVariableNode(const UA_NodeId& parentNodeId,	×
222	const OPCUA::TVariableNodeConfig& variableNode,
223	WBMQTT::PDeviceDriver driver)
224	{
225	auto components = WBMQTT::StringSplit(variableNode.DeviceControlPair, "/");	×
226
227	UA_VariableAttributes oAttr = UA_VariableAttributes_default;	×
228	SetVariableAttributes(oAttr, driver, components[0], components[1]);	×
229
230	UA_DataSource dataSource;
231	dataSource.read = ReadVariableCallback;	×
232	dataSource.write = WriteVariableCallback;	×
233
234	UA_NodeId nodeId = UA_NODEID_STRING(1, (char*)variableNode.DeviceControlPair.c_str());	×
235
236	auto res = UA_Server_addDataSourceVariableNode(Server,	×
237	nodeId,
238	parentNodeId,
239	UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT),
240	UA_QUALIFIEDNAME(1, (char*)components[1].c_str()),	×
241	UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE),
242	oAttr,
243	dataSource,
244	this,
245	nullptr);
246	if (res != UA_STATUSCODE_GOOD) {	×
247	throw std::runtime_error("Variable node '" + variableNode.DeviceControlPair +	×
248	"' creation failed: " + UA_StatusCode_name(res));	×
249	}
250	}
251
252	public:
253	TServerImpl(const OPCUA::TServerConfig& config, WBMQTT::PDeviceDriver driver)	×
254	: Server(UA_Server_new()),	×
255	IsRunning(true),
256	Driver(driver)	×
257	{
258	if (!Server) {	×
259	throw std::runtime_error("OPC UA server initilization failed");	×
260	}
261
NEW 262	Driver->On<WBMQTT::TControlValueEvent>([&](const WBMQTT::TControlValueEvent& event) {	×
NEW 263	if (event.RawValue.empty()) {	×
NEW 264	return;	×
265	}
NEW 266	auto it = config.ObjectNodes.find(event.Control->GetDevice()->GetId());	×
NEW 267	if (it == config.ObjectNodes.end()) {	×
NEW 268	return;	×
269	}
270
NEW 271	auto browseName = UA_QUALIFIEDNAME(1, (char*)it->first.c_str());	×
NEW 272	auto res = UA_Server_browseSimplifiedBrowsePath(Server,	×
273	UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
274	1,
NEW 275	&browseName);	×
NEW 276	auto parentNodeId = res.statusCode == UA_STATUSCODE_GOOD ? res.targets[0].targetId.nodeId	×
NEW 277	: CreateObjectNode(it->first, it->second);	×
NEW 278	std::string nodeName = it->first + "/" + event.Control->GetId();	×
NEW 279	for (auto& valueNode: it->second) {	×
NEW 280	if (valueNode.DeviceControlPair != nodeName) {	×
NEW 281	continue;	×
282	}
NEW 283	browseName = UA_QUALIFIEDNAME(1, (char*)event.Control->GetId().c_str());	×
NEW 284	res = UA_Server_browseSimplifiedBrowsePath(Server, parentNodeId, 1, &browseName);	×
NEW 285	if (res.statusCode != UA_STATUSCODE_GOOD) {	×
NEW 286	CreateVariableNode(parentNodeId, valueNode, Driver);	×
NEW 287	break;	×
288	}
289	}
NEW 290	});	×
291
292	// Load external controls
293	std::vector<std::string> deviceIds;	×
294	for (const auto& device: config.ObjectNodes) {	×
UNCOV 295	LOG(Debug) << "'" << device.first << "' is added to filter";	×
296	deviceIds.emplace_back(device.first);	×
297	}
298	Driver->SetFilter(WBMQTT::GetDeviceListFilter(deviceIds));	×
UNCOV 299	Driver->WaitForReady();	×
300
301	// Setup OPC UA server and nodes
302	ConfigureOpcUaServer(UA_Server_getConfig(Server), config);	×
303
304	// for (auto& node: config.ObjectNodes) {
305	// auto nodeId = CreateObjectNode(node.first, node.second);
306	// for (auto& vn: node.second) {
307	// CreateVariableNode(nodeId, vn, driver);
308	// }
309	// }
310
311	ServerThread = std::thread([this]() {	×
UNCOV 312	auto res = UA_Server_run(Server, &IsRunning);	×
313	if (res != UA_STATUSCODE_GOOD) {	×
314	LOG(Error) << UA_StatusCode_name(res);	×
315	exit(1);	×
316	}
317	});	×
318	}
319
320	~TServerImpl()	×
321	{
UNCOV 322	if (IsRunning) {	×
323	IsRunning = false;	×
324	if (ServerThread.joinable()) {	×
325	ServerThread.join();	×
326	}
327	}
328	if (Server) {	×
329	UA_Server_delete(Server);	×
330	}
331	}
332
UNCOV 333	UA_StatusCode writeVariable(const UA_NodeId* snodeId, const UA_DataValue* dataValue)	×
334	{
335	std::string nodeIdName((const char*)snodeId->identifier.string.data, snodeId->identifier.string.length);	×
336	auto tx = Driver->BeginTx();	×
337	auto ctrl = GetControl(tx, nodeIdName);	×
UNCOV 338	if (!ctrl \|\| ctrl->IsReadonly()) {	×
UNCOV 339	LOG(Error) << "Variable node '" + nodeIdName + "' writing failed. "	×
UNCOV 340	<< (ctrl ? "It is read only" : "It is not presented in MQTT");	×
341	return UA_STATUSCODE_BADDEVICEFAILURE;	×
342	}
343	try {
344	if (dataValue->hasValue) {	×
345	if (UA_Variant_hasScalarType(&dataValue->value, &UA_TYPES[UA_TYPES_BOOLEAN])) {	×
UNCOV 346	auto value = (UA_Boolean)dataValue->value.data;	×
347	ctrl->SetValue(tx, value).Sync();	×
UNCOV 348	LOG(Info) << "Variable node '" + nodeIdName + "' = " << value;	×
UNCOV 349	return UA_STATUSCODE_GOOD;	×
350	}
UNCOV 351	if (UA_Variant_hasScalarType(&dataValue->value, &UA_TYPES[UA_TYPES_DOUBLE])) {	×
352	auto value = (UA_Double)dataValue->value.data;	×
353	ctrl->SetValue(tx, value).Sync();	×
354	LOG(Info) << "Variable node '" + nodeIdName + "' = " << value;	×
355	return UA_STATUSCODE_GOOD;	×
356	}
UNCOV 357	if (UA_Variant_hasScalarType(&dataValue->value, &UA_TYPES[UA_TYPES_STRING])) {	×
358	auto value = (char)((UA_String)dataValue->value.data)->data;	×
359	ctrl->SetRawValue(tx, value).Sync();	×
UNCOV 360	LOG(Info) << "Variable node '" + nodeIdName + "' = " << value;	×
UNCOV 361	return UA_STATUSCODE_GOOD;	×
362	}
363	}
UNCOV 364	return UA_STATUSCODE_BADDATATYPEIDUNKNOWN;	×
365	} catch (const std::exception& e) {	×
366	LOG(Error) << "Variable node '" + nodeIdName + "' write error: " << e.what();	×
367	return UA_STATUSCODE_BADDEVICEFAILURE;	×
368	}
369	}
370
371	UA_StatusCode readVariable(const UA_NodeId* snodeId, UA_DataValue* dataValue)	×
372	{
UNCOV 373	std::string nodeIdName((const char*)snodeId->identifier.string.data, snodeId->identifier.string.length);	×
374	auto tx = Driver->BeginTx();	×
375	auto ctrl = GetControl(tx, nodeIdName);	×
376	if (!ctrl) {	×
377	LOG(Error) << "Control is not found '" + nodeIdName + "'";	×
378	dataValue->hasStatus = true;	×
379	dataValue->status = UA_STATUSCODE_BADNOCOMMUNICATION;	×
UNCOV 380	return UA_STATUSCODE_GOOD;	×
381	}
382	try {
383	dataValue->hasStatus = true;	×
384	if (ctrl->GetError().find("r") != std::string::npos) {	×
385	dataValue->status = UA_STATUSCODE_BAD;	×
386	} else {
387	dataValue->status = UA_STATUSCODE_GOOD;	×
388	}
389	auto v = ctrl->GetValue();	×
390	if (v.Is<bool>()) {	×
391	auto value = v.As<bool>();	×
UNCOV 392	UA_Variant_setScalarCopy(&dataValue->value, &value, &UA_TYPES[UA_TYPES_BOOLEAN]);	×
393	} else {
394	if (v.Is<double>()) {	×
395	auto value = v.As<double>();	×
396	UA_Variant_setScalarCopy(&dataValue->value, &value, &UA_TYPES[UA_TYPES_DOUBLE]);	×
397	} else {
UNCOV 398	UA_String stringValue = UA_String_fromChars((char*)v.As<std::string>().c_str());	×
UNCOV 399	UA_Variant_setScalarCopy(&dataValue->value, &stringValue, &UA_TYPES[UA_TYPES_STRING]);	×
UNCOV 400	UA_String_clear(&stringValue);	×
401	}
402	}
403	dataValue->hasValue = true;	×
404	} catch (const std::exception& e) {	×
405	LOG(Error) << "Variable node '" + nodeIdName + "' read error: " << e.what();	×
406	dataValue->hasStatus = true;	×
407	dataValue->status = UA_STATUSCODE_BADNOCOMMUNICATION;	×
408	}
409	return UA_STATUSCODE_GOOD;	×
410	}
411	};
412
413	extern "C" {
414	UA_StatusCode ReadVariableCallback(UA_Server* sserver,	×
415	const UA_NodeId* ssessionId,
416	void* ssessionContext,
417	const UA_NodeId* snodeId,
418	void* snodeContext,
419	UA_Boolean ssourceTimeStamp,
420	const UA_NumericRange* range,
421	UA_DataValue* dataValue)
422	{
UNCOV 423	TServerImpl* server = (TServerImpl*)(snodeContext);	×
424	return server->readVariable(snodeId, dataValue);	×
425	}
426
UNCOV 427	UA_StatusCode WriteVariableCallback(UA_Server* server,	×
428	const UA_NodeId* sessionId,
429	void* sessionContext,
430	const UA_NodeId* nodeId,
431	void* nodeContext,
432	const UA_NumericRange* range,
433	const UA_DataValue* data)
434	{
435	TServerImpl* s = (TServerImpl*)(nodeContext);	×
436	return s->writeVariable(nodeId, data);	×
437	}
438	}
439	}
440
UNCOV 441	std::unique_ptr<OPCUA::IServer> OPCUA::MakeServer(const OPCUA::TServerConfig& config, WBMQTT::PDeviceDriver driver)	×
442	{
UNCOV 443	return std::unique_ptr<OPCUA::IServer>(new TServerImpl(config, driver));	×
444	}

wirenboard / wb-mqtt-opcua / 2

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous