280 lines
9.2 KiB
JavaScript
280 lines
9.2 KiB
JavaScript
const { OPCUAClient, AttributeIds, ReadRequest, ReadResponse, promoteOpaqueStructure, WriteRequest, WriteResponse, StatusCodes, DataType } = require("node-opcua");
|
|
|
|
|
|
let connection_registry = new Map();
|
|
let client_is_closed = new WeakSet();
|
|
|
|
class Queue {
|
|
constructor(single_consumer) {
|
|
this.q = [];
|
|
this.promises = [];
|
|
this.single_consumer = false;
|
|
this.hangup_done = false;
|
|
}
|
|
async _get() {
|
|
if (this.hangup_done) {
|
|
throw new Error('hangup');
|
|
}
|
|
if (this.q.length > 0) {
|
|
return this.q.shift()
|
|
}
|
|
return new Promise((resolve, reject) => {
|
|
this.promises.push({ resolve, reject })
|
|
});
|
|
}
|
|
async process(func) {
|
|
let q_el = await this._get();
|
|
try {
|
|
let out = await func(q_el.element);
|
|
q_el.resolve(out);
|
|
} catch (e) {
|
|
q_el.reject(e);
|
|
}
|
|
}
|
|
async submit_and_wait_processed(element) {
|
|
return await new Promise((resolve, reject) => {
|
|
let queue_el = {
|
|
element,
|
|
resolve,
|
|
reject
|
|
}
|
|
this._put(queue_el)
|
|
})
|
|
}
|
|
_put(element) {
|
|
if (this.promises.length > 0) {
|
|
this.promises.shift().resolve(element);
|
|
} else {
|
|
this.q.push(element);
|
|
}
|
|
}
|
|
hangup() {
|
|
this.hangup_done = true;
|
|
for (let prom of this.promises) {
|
|
prom.reject('hangp');
|
|
}
|
|
}
|
|
}
|
|
|
|
async function perform_request_loop(queue, session) {
|
|
try {
|
|
while (true) {
|
|
await queue.process(async (request) => {
|
|
return await new Promise((resolve, reject) => {
|
|
session.performMessageTransaction(request, (err, response) => {
|
|
/* c8 ignore next */
|
|
if (err) {
|
|
reject(err);
|
|
} else {
|
|
resolve(response);
|
|
}
|
|
})
|
|
})
|
|
})
|
|
}
|
|
} catch (e) {
|
|
if (e.message == 'hangup') {
|
|
} else {
|
|
throw e;
|
|
}
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
async function get_client_and_session(opcua_conn_data) {
|
|
if (!opcua_conn_data?.id) {
|
|
throw new Error("'id' missing in opcua connection settings");
|
|
}
|
|
let client_id = opcua_conn_data.id;
|
|
if (connection_registry.has(client_id)) {
|
|
let entry = connection_registry.get(client_id);
|
|
let { client, } = entry;
|
|
if (!client_is_closed.has(client)) {
|
|
return entry;
|
|
}
|
|
}
|
|
const client_settings = {
|
|
endpointMustExist: false,
|
|
connectionStrategy: {
|
|
initialDelay: 1,
|
|
maxRetry: 3,
|
|
...(opcua_conn_data?.client_settings?.connectionStrategy || {})
|
|
},
|
|
reconnectOnFailure: false,
|
|
...(opcua_conn_data?.client_settings || {})
|
|
};
|
|
const client = OPCUAClient.create(client_settings);
|
|
let queue = new Queue();
|
|
client.on("connection_lost", async () => {
|
|
client_is_closed.add(client);
|
|
try {
|
|
await session.close()
|
|
} catch (e) {
|
|
} finally {
|
|
await client.disconnect();
|
|
}
|
|
});
|
|
await client.connect(opcua_conn_data.url);
|
|
const session = await client.createSession();
|
|
perform_request_loop(queue, session);
|
|
let entry = { session, client, queue };
|
|
connection_registry.set(client_id, entry)
|
|
return entry;
|
|
}
|
|
|
|
function mapAttributeIds(list_in) {
|
|
let list_out = [];
|
|
for (let n of list_in) {
|
|
let attrid = n?.attributeId;
|
|
if (typeof attrid == 'string') {
|
|
attrid = AttributeIds[n.attributeId];
|
|
if (!attrid) {
|
|
throw new Error("unknown attribute id: " + n?.attributeId + " for nodeId: " + n?.nodeId)
|
|
}
|
|
}
|
|
if (attrid === false || attrid === undefined || attrid === null) {
|
|
attrid = AttributeIds.Value;
|
|
}
|
|
list_out.push({
|
|
...n,
|
|
attributeId: attrid
|
|
})
|
|
}
|
|
return list_out;
|
|
}
|
|
function mapDataTypes(list_in) {
|
|
let list_out = [];
|
|
for (let n of list_in) {
|
|
if (
|
|
typeof (n?.value) === 'object' &&
|
|
!Array.isArray(n?.value) &&
|
|
n?.value !== null) {
|
|
// se è già dizionario, facciamo conto che sia
|
|
// nel formato {value: {value: x, dataType:"lkdsjflks"}}
|
|
list_out.push(n)
|
|
} else {
|
|
// altrimenti leggiamo dataType e lo convertiamo.
|
|
let dtype = n?.dataType;
|
|
|
|
if (typeof dtype == 'string') {
|
|
dtype = DataType[dtype];
|
|
if (!dtype) {
|
|
throw new Error("unknown datatype id: " + dtype)
|
|
}
|
|
}
|
|
if (dtype === false || dtype === undefined || dtype === null) {
|
|
throw new Error("missing datatype for nodeId: " + n?.nodeId);
|
|
}
|
|
list_out.push({
|
|
...n,
|
|
value: { value: { dataType: dtype, value: n?.value } },
|
|
})
|
|
}
|
|
}
|
|
return list_out;
|
|
}
|
|
|
|
module.exports = function (RED) {
|
|
function OpcUaBriqRead(config) {
|
|
RED.nodes.createNode(this, config);
|
|
var node = this;
|
|
node.on('input', async function (msg) {
|
|
msg.payload = msg.payload || {};
|
|
try {
|
|
let conn = await get_client_and_session(msg.opcua_server);
|
|
let nodesToRead = mapAttributeIds(msg?.nodesToRead || []);
|
|
let response = await conn.queue.submit_and_wait_processed(
|
|
new ReadRequest({
|
|
nodesToRead: nodesToRead,
|
|
})
|
|
)
|
|
if (!response || !(response instanceof ReadResponse)) {
|
|
throw new Error("Internal Error");
|
|
}
|
|
|
|
if (nodesToRead.length !== response.results.length) {
|
|
throw new Error("wrong response length");
|
|
}
|
|
msg.raw_response = response;
|
|
let parsed = {};
|
|
for (let i = 0; i < nodesToRead.length; i++) {
|
|
let label = nodesToRead[i]?.label;
|
|
if (!label) {
|
|
label = nodesToRead[i]?.nodeId;
|
|
}
|
|
parsed[label] = response.results[i].value.value;
|
|
}
|
|
msg.payload = parsed;
|
|
msg.error = undefined;
|
|
node.send(msg);
|
|
} catch (e) {
|
|
msg.error = e.message + '' + e.stack;
|
|
node.send(msg);
|
|
}
|
|
});
|
|
}
|
|
RED.nodes.registerType("opcua-briq-read", OpcUaBriqRead);
|
|
|
|
|
|
function OpcUaBriqWrite(config) {
|
|
RED.nodes.createNode(this, config);
|
|
var node = this;
|
|
node.on('input', async function (msg) {
|
|
msg.payload = msg.payload || {};
|
|
try {
|
|
let conn = await get_client_and_session(msg.opcua_server);
|
|
let nodesToWrite = mapAttributeIds(msg?.nodesToWrite || []);
|
|
nodesToWrite = mapDataTypes(nodesToWrite);
|
|
let response = await conn.queue.submit_and_wait_processed(
|
|
new WriteRequest({
|
|
nodesToWrite: nodesToWrite,
|
|
})
|
|
)
|
|
if (!response || !(response instanceof WriteResponse)) {
|
|
throw new Error("Internal Error");
|
|
}
|
|
|
|
if (response.responseHeader.serviceResult.isNot(StatusCodes.Good)) {
|
|
throw new Error(response.responseHeader.serviceResult.toString());
|
|
}
|
|
|
|
if (nodesToWrite.length !== response.results.length) {
|
|
throw new Error("wrong response length");
|
|
}
|
|
msg.raw_response = response;
|
|
if (msg?.verifyWrittenValues) {
|
|
// verifica valori appena scritti
|
|
let response = await conn.queue.submit_and_wait_processed(
|
|
new ReadRequest({
|
|
nodesToRead: nodesToWrite,
|
|
})
|
|
)
|
|
if (!response || !(response instanceof ReadResponse)) {
|
|
throw new Error("Internal Error");
|
|
}
|
|
if (nodesToWrite.length !== response.results.length) {
|
|
throw new Error("wrong response length");
|
|
}
|
|
for (let i = 0; i < nodesToWrite.length; i++) {
|
|
let left = nodesToWrite[i].value.value.value; // Lol
|
|
let right = response.results[i].value.value;
|
|
if (left !== right) {
|
|
throw new Error(`verify failed: ${left} != ${right}`);
|
|
}
|
|
}
|
|
msg.payload = "write ok & verified";
|
|
} else {
|
|
msg.payload = "write ok";
|
|
}
|
|
msg.error = undefined;
|
|
node.send(msg);
|
|
} catch (e) {
|
|
msg.error = e.message + '' + e.stack;
|
|
node.send(msg);
|
|
}
|
|
});
|
|
}
|
|
RED.nodes.registerType("opcua-briq-write", OpcUaBriqWrite);
|
|
} |