async-mqtt.js

import * as os from "os";
import * as std from "std";
import { WebSocket } from "/usr/lib/quickjs/websocket.js";

class AsyncClient {
  constructor() {
    globalThis.WebSocket = WebSocket;
    // paho-mqtt requires setTimeout and clearTimeout in global scope
    globalThis.setTimeout = os.setTimeout;
    globalThis.clearTimeout = function (timeout) {
      if (timeout) {
        os.clearTimeout(timeout);
      }
    };
    globalThis.global = globalThis;
    std.loadScript("/usr/lib/usp-js/lib/protobuf.min.js");
    std.loadScript("/usr/lib/usp-js/lib/paho-mqtt.min.js");
    protobuf.Root.prototype.fetch = function (filename, callback) {
      os.setTimeout(function () {
        const data = std.loadFile(filename);
        if (data === null)
          callback(new Error("failed to load file: " + filename));
        else
          callback(null, data);
      }, 0);
    };
  }

  async connectAsync(
    host = "localhost",
    port = 9001,
    user = "admin",
    password = "admin"
  ) {
    this.client = new Paho.Client(
      host,
      port,
      "/mqtt",
      "proto::interop-usp-controller"
    );
    const opts = {
      userName: user,
      password: password,
    };

    const client = this.client;
    const connectPromise = new Promise(function (resolve, reject) {
      opts.onFailure = function (err) {
        reject(new Error(err.errorMessage));
      };

      opts.onSuccess = function () {
        resolve(client);
      };
      client.connect(opts);
    });

    this.client = await connectPromise;
    // this.client.subscribe('/usp/controller/#')

    return this;
  }

  async subscribe(id) {
    this.client.subscribe(id);
  }

  async unsubscribe(id) {
    this.client.unsubscribe(id);
  }

  async on(msg, callback) {
    if (msg === "message") {
      this.client.onMessageArrived = function (msg) {
        callback(msg.payloadBytes);
      };
    } else if (msg === "error") {
      // there is no erro handling in paho mqtt
      this.client.onConnectionLost = function (responseObject) {
        callback(responseObject);
      };
    }
  }

  async publish(topic, msg) {
    this.client.send(topic, msg);
  }

  async end() {
    this.client.disconnect();
  }
}

async function main() {
  const client = new AsyncClient();
  await client.connect();
}

const mqttAsync = new AsyncClient();
export default mqttAsync;