node-red-contrib-xilica/nodes/xilica-subscribe.js

const net = require("net");

module.exports = function (RED) {
  function buildControlNamesFromRules(rules) {
    const names = [];
    if (!Array.isArray(rules)) {
      return names;
    }

    rules.forEach((rule) => {
      if (!rule) {
        return;
      }

      const type = (rule.t || rule.type || "eq").toString();

      if (type === "idx") {
        const base = (rule.v || rule.base || "").trim();
        if (!base) {
          return;
        }

        let from = parseInt(rule.from, 10);
        let to = parseInt(rule.to, 10);

        if (Number.isNaN(from) && Number.isNaN(to)) {
          return;
        }
        if (Number.isNaN(from)) {
          from = to;
        }
        if (Number.isNaN(to)) {
          to = from;
        }
        if (from > to) {
          const tmp = from;
          from = to;
          to = tmp;
        }

        for (let i = from; i <= to; i += 1) {
          names.push(base + i);
        }
      } else {
        const name = (rule.v || rule.value || "").trim();
        if (name) {
          names.push(name);
        }
      }
    });

    return names;
  }

  function normaliseAction(action) {
    const a = (action || "").toString().toLowerCase();
    if (a === "unsubscribe" || a === "unsub") {
      return "UNSUBSCRIBE";
    }
    return "SUBSCRIBE";
  }

  function XilicaSubscribe(config) {
    RED.nodes.createNode(this, config);
    const node = this;

    node.action = config.action || "subscribe";
    node.rules = Array.isArray(config.rules) ? config.rules : [];
    node.transport = config.transport || "TCP";
    node.connectionConfig = RED.nodes.getNode(config.connection);

    node.on("input", (msg, send, done) => {
      send =
        send ||
        function () {
          node.send.apply(node, arguments);
        };
      done = done || function () {};

      const action = normaliseAction(msg.action || node.action);
      const rules = Array.isArray(msg.rules) ? msg.rules : node.rules;

      const transport =
        typeof msg.transport === "string" && msg.transport.trim().length
          ? msg.transport.trim()
          : node.transport;

      const names = buildControlNamesFromRules(rules);

      if (!names.length) {
        node.warn("xilica-subscribe: no rules defined");
        send(msg);
        done();
        return;
      }

      const lines = names.map(
        (name) => action + " " + name + ' "' + transport + '"',
      );

      const rawCommand = lines.join("\r");

      const conn = node.connectionConfig;
      if (!conn || !conn.host) {
        node.status({
          fill: "red",
          shape: "ring",
          text: "no connection",
        });
        node.warn(
          "xilica-subscribe: no connection configured, outputting commands only",
        );
        msg.payload = rawCommand;
        send(msg);
        done();
        return;
      }

      const host = conn.host;
      const port = conn.port || 10007;
      const timeout = conn.timeout || 5000;
      const password =
        conn.credentials && conn.credentials.password
          ? conn.credentials.password
          : null;

      const client = net.createConnection({ host, port });
      let buffer = "";
      let phase = password ? "login" : "command";
      let finished = false;

      function finishSuccess() {
        if (finished) {
          return;
        }
        finished = true;
        client.end();
        node.status({});

        msg.raw = buffer;
        msg.payload = rawCommand;

        send(msg);
        done();
      }

      function finishError(err) {
        if (finished) {
          return;
        }
        finished = true;
        client.destroy();
        node.status({ fill: "red", shape: "ring", text: "error" });
        node.error(err, msg);
        done(err);
      }

      client.setTimeout(timeout);

      client.on("connect", () => {
        node.status({ fill: "green", shape: "dot", text: "connected" });
        try {
          if (phase === "login") {
            client.write("LOGIN " + password + "\r");
          } else {
            client.write(rawCommand + "\r");
          }
        } catch (err) {
          finishError(err);
        }
      });

      client.on("data", (dataBuf) => {
        buffer += dataBuf.toString("utf8");
        let idx = buffer.indexOf("\r");
        while (idx !== -1) {
          const line = buffer.slice(0, idx);
          buffer = buffer.slice(idx + 1);

          if (phase === "login") {
            if (line && line.startsWith("ERROR=")) {
              finishError(new Error("Login failed: " + line));
              return;
            }
            phase = "command";
            try {
              client.write(rawCommand + "\r");
            } catch (err) {
              finishError(err);
              return;
            }
          }

          idx = buffer.indexOf("\r");
        }
      });

      client.on("timeout", () => {
        if (phase === "command") {
          finishSuccess();
        } else {
          finishError(new Error("Timeout sending SUBSCRIBE to Xilica"));
        }
      });

      client.on("error", (err) => {
        finishError(err);
      });

      client.on("end", () => {
        if (!finished) {
          finishSuccess();
        }
      });
    });
  }

  RED.nodes.registerType("xilica-subscribe", XilicaSubscribe);
};