qualiteer/lib/sockets/clients/Initiator.js

import { Manager } from "socket.io-client";
import modes from "../modes.js";
import events from "../events.js";

export { default as events } from "../events.js";
export { default as modes } from "../modes.js";

export default class Initiator {
  constructor(url, options = {}) {
    this.url = url;
    this.mode = modes.INIT;
    this.onLog = options.onLog ?? ((d) => console.log(`job: ${d}`));
    this.onClose = options.onClose ?? (() => {});
    this.onCreate = options.onCreate ?? ((id) => console.log(`job id: ${id}`));
    this.onPipelineClose =
      options.onPipelineClose ??
      (() => {
        console.log("job pipeline closed");
      });
    this.sk = null;
  }

  async newJob(jobRequest, onLog, onClose, onCreate) {
    const mgr = new Manager(this.url, {
      query: { mode: this.mode, job: JSON.stringify(jobRequest) },
    });
    onLog = onLog ?? this.onLog.bind(this);
    onClose = onClose ?? this.onClose.bind(this);
    onCreate = onCreate ?? this.onCreate.bind(this);
    this.sk = mgr.socket("/");
    this.sk.on(events.JOB_LOG, onLog);
    this.sk.on(events.JOB_CLS, onClose);
    return new Promise((res) =>
      this.sk.on(events.JOB_CRT, function onJobCreate(id) {
        onCreate(id);
        res({ ...jobRequest, id });
      })
    );
  }

  async newPipelineJob(
    jobRequest,
    onLog,
    onClose,
    onCreate,
    onPipelineTrigger,
    onPipelineClose
  ) {
    const mgr = new Manager(this.url, {
      query: { mode: this.mode, job: JSON.stringify(jobRequest) },
    });
    onLog = onLog ?? this.onLog.bind(this);
    onClose = onClose ?? this.onClose.bind(this);
    onCreate = onCreate ?? this.onCreate.bind(this);
    onPipelineTrigger =
      onPipelineTrigger ??
      ((pipeline) => {
        console.log("job trg:", pipeline);
        const { triggers } = pipeline;
        if (!Object.keys(triggers).length) onPipelineClose();
        // For each trigger
        for (var testName in triggers) {
          const delay = triggers[testName].__testDelay ?? 0;
          delete triggers[testName].__testDelay;
          const jobReq = {
            ...jobRequest,
            pipeline: {...pipeline, triggers: triggers[testName],__test:testName },
          };
          setTimeout(
            () =>
              this.newPipelineJob(
                jobReq,
                onLog,
                onClose,
                onCreate,
                onPipelineTrigger,
                onPipelineClose
              ),
            delay
          );
        }
      });
    onPipelineClose = onPipelineClose ?? this.onPipelineClose.bind(this);
    this.sk = mgr.socket("/");
    this.sk.on(events.JOB_LOG, onLog);
    this.sk.on(events.JOB_CLS, onClose);
    this.sk.on(events.PPL_TRG, onPipelineTrigger);
    this.sk.on(events.PPL_CLS, onPipelineClose);
    return new Promise((res) =>
      this.sk.on(events.JOB_CRT, function onJobCreate(id) {
        onCreate(id);
        res({ ...jobRequest, id });
      })
    );
  }
}