opcua read e write

red/opcua/opcua-briq.js

@@ -0,0 +1,280 @@
+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);
+}