queenshow/src/controllers/natsController.ts

import { queenApi } from "queenjs";
import { connect, StringCodec, Empty, ErrorCode } from "nats.ws";

export class BusController {
  _params = new URLSearchParams(decodeURIComponent(location.search));

  _conn: any;
  _isConned = false;
  _startInit = false;

  getQuery(name: string): string {
    return this._params.get(name) as string;
  }

  async init(host?: string) {
    if (this._startInit) return;
    this._startInit = true;
    
    queenApi.showLoading("服务连接中...");
    const wsHost = host ? host : this.getQuery("host");
    
    console.log("ws host=>", wsHost)
    
    let ret = false;
    try {
      this._conn = await connect({ servers: wsHost });
      this._isConned = !!this._conn;

      ret = true;
    } catch (error) {
      console.log(error);
      queenApi.messageError("连接失败!");
    }
    queenApi.hideLoading();
    this._startInit = false;

    return ret;
  }

  async subscribe(subject: string, callback: any) {
    if (!this._isConned) {
      await this.init();
    }

    if (!this._isConned) {
      console.error("建立连接失败");
      return;
    }

    const sc = StringCodec();
    const sub = this._conn.subscribe(subject);
    console.log(sub);

    (async () => {
      for await (const m of sub) {
        console.log(sub);

        const ret = sc.decode(m.data);
        console.log(subject, "=>recieved");
        try {
          if (ret) {
            const msg = JSON.parse(ret);
            callback(msg);
          } else {
            callback(ret);
          }
        } catch (error) {
          console.log(subject, "=>recieved json parse eror", ret);
          console.log(error);
        }
      }
      console.log(subject, "subscription closed");
    })();
    return function () {
      sub.unsubscribe();
    };
  }
  async requestApi(subject: string, data?: any, timeout?: number) {
     const ret = await this.request(subject, data, timeout)
     console.log("request api=>", ret)
     
     if (ret.error || (ret.result.ErrorNo && ret.result.ErrorNo != 200) ) {
       queenApi.messageError(ret.error || ret.result.ErrorDesc );
       return
     }
     try {
        const retJson = ret.result.Result;
        if (!retJson) return;
        if (retJson[0] != '{' && retJson[0] != '[') return retJson;
        return JSON.parse(retJson)
     } catch (error) {
        console.log( ret );
        console.error(error);
     }
  }

  RequestWebView(name: string, data:any) {
    return new Promise((r)=>{
      let cancel:any = null;
      this.subscribe("webview.close."+name, function(payload:any){
        cancel&&cancel();
        r(payload)
      }).then(c=>{
         cancel = c;
      })
      this.requestApi("webview.open", {...data, name})
    })
  }

  async request(subject: string, data?: any, timeout?: number) {
    if (!this._isConned) {
      await this.init();
    }

    const ret: { error: string; result: any } = { error: "", result: null };
    if (!this._isConned) {
      console.error("建立连接失败");
      ret.error = "建立连接失败";

      queenApi.showConfirm({
        title: "数据请求失败",
        content: "请求数据失败,请重新启动后再试",
        type: "danger",
      });
      return ret;
    }

    const sc = StringCodec();
    try {
      let req = Empty;
      if (data) {
        if (typeof data != "string") {
          req = sc.encode(JSON.stringify(data));
        } else {
          req = sc.encode(data);
        }
      }
      const options = { timeout: 5000 };
      if (timeout) {
        options.timeout = timeout;
      }
      const m = await this._conn.request(subject, req, options);
      let payload = sc.decode(m.data);
      try {
        payload = JSON.parse(payload);
        console.log("m=>", payload);
      } catch (error) {
        console.log(error);
      }
      ret.result = payload;
    } catch (error: any) {
      console.error(error);
      ret.error = error.message || "请求" + subject + "出错";
      if (ret.error == "503") {
        //NoResponders
        ret.error = "网路异常,请重新服务";
      }
    }
    return ret;
  }

  close() {
    if (!this._isConned) {
      return;
    }
    return this._conn.close();
  }

  
  async Public(subject:string, data:any) {
    if (!this._isConned) {
      await this.init();
    }
    const ret: { error: string; result: any } = { error: "", result: null };
    if (!this._isConned) {
      console.error("建立连接失败");
      ret.error = "建立连接失败";
      queenApi.showConfirm({
        title: "数据请求失败",
        content: "请求数据失败,请重新启动后再试",
        type: "danger",
      });
      return false;
    }

    const sc = StringCodec();
    try {
      let req = Empty;
      if (data) {
        if (typeof data != "string") {
          req = sc.encode(JSON.stringify(data));
        } else {
          req = sc.encode(data);
        }
      }
      this._conn.publish(subject,req);

    } catch (error: any) {
      console.log( error);
      return false
    }

    return true;
  }
}