import { Empty, StringCodec, connect } from "nats.ws"; import { queenApi } from "queenjs"; 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) { const wsHost = host ? host : this.getQuery("host"); if (this._startInit || !wsHost || wsHost == "null") return; this._startInit = true; queenApi.showLoading("服务连接中..."); 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; } }