perf(web ui):keep alive consumer (#2325)

web-ui/arthasWebConsole/ui/src/arthas.d.ts

@@ -422,6 +422,7 @@ type CommandResult = {
 } | {
   options: GlobalOptions[];
   changeResult: {
+
     "afterValue": unknown,
     "beforeValue": unknown,
     "name": string
@@ -709,7 +710,9 @@ type ArthasRes = CommonRes | SessionRes | FailRes | AsyncRes;
 type BindQS =
   | { req: CommandReq; res: CommonRes }
   | { req: SessionReq; res: SessionRes }
-  | { req: AsyncReq; res: AsyncRes };
+  | { req: AsyncReq; res: AsyncRes }
+  | { req: PullResults; res: ArthasRes };
+
 // autoComplete
 type Item = { name: string; value: unknown };
 

web-ui/arthasWebConsole/ui/src/components/NavHeader.vue

@@ -26,7 +26,10 @@ publicS.getCommonResEffect(fetchM, body => {
   const result = body.results[0]
   if (result.type === "version") {
     version.value = result.version
-  }})
+
+  }
+})
+
 
 watchEffect(() => {
   if (!fetchS.wait) restBtnclass.value = "animate-spin-rev-pause"
@@ -68,20 +71,12 @@ const logout = async () => {
 
   interruptEvent()
 
-  sessionM.send("SUBMIT", {
-    value: {
-      action: "close_session",
-      sessionId: undefined
-    }
-  })
+  fetchS.closeSession()
   restBtnclass.value = "animate-spin-rev-pause"
 }
-const login = async () => {
-  sessionM.send("SUBMIT", {
-    value: {
-      action: "init_session"
-    }
-  })
+
+const login = () => {
+  fetchS.initSession()
 }
 const shutdown = () => {
   publicS.warnMessage = "Are you sure to stop the arthas? All the Arthas clients connecting to this server will be disconnected."
@@ -134,6 +129,7 @@ const tabs = [
     icon: TerminalIcon
   },
   {
+
     name:'terminal',
     url:'terminal',
     icon:TerminalIcon

web-ui/arthasWebConsole/ui/src/machines/perRequestMachine.ts

@@ -443,7 +443,7 @@ const permachine = createMachine({
     isCommon: (context) => {
       if (!context) return false;
       if (
-        ["exec"]
+        ["exec","pull_results"]
           .includes(context.inputValue!.action)
       ) {
         console.log("isCommon");
@@ -454,7 +454,7 @@ const permachine = createMachine({
     isAsync: (context) => {
       if (!context) return false;
       if (
-        ["async_exec", "pull_results"]
+        ["async_exec"]
           .includes(context.inputValue!.action)
       ) {
         console.log("isAsync");

web-ui/arthasWebConsole/ui/src/router/index.ts

@@ -12,6 +12,10 @@ const router = createRouter({
 router.beforeEach((to, from, next) => {
   fetchStore()
     .interruptJob()
+    .catch(_=>{
+      // console.error(e)
+      // 拦截调试台的错误
+    })
     .finally(() => {
       next();
     });

web-ui/arthasWebConsole/ui/src/stores/fetch.ts

@@ -27,7 +27,8 @@ type PollingLoop = {
   invoke(): void;
 };
 const nullLoop: PollingLoop = {
-  open() {},
+  open() {
+  },
   close() {},
   isOn() {
     return false;
@@ -48,11 +49,11 @@ export const fetchStore = defineStore("fetch", {
     // 所有用pollingLoop都要
     jobRunning: false,
     // 对于 pullresults可能会拉同一个结果很多次
-    jobIdSet: new Set<string>(),
+    // jobIdSet: new Set<string>(),
     //由于轮询只会轮询一个命令，可以直接挂载当前的轮询机
     curPolling: nullLoop,
-    //获取osName 通过dashboard
-    osName:""
+    //对session init以后4分钟重轮一次。
+    // 如果curPolling是nullLoop,就要停掉状态维持来防止消耗异步请求结果的行为
   }),
   getters: {
     getRequest: (state) =>
@@ -94,6 +95,15 @@ export const fetchStore = defineStore("fetch", {
       },
   },
   actions: {
+    /**
+     * @param hander 要使用的函数
+     * @param options step：间隔时间，globalIntrupt: 是否需要全局的打断按钮
+     * @returns 一个轮询对象
+     * 需要open以后才挂载到curPolling
+     * close以后会重置为nullLoop
+     * curloop的状态转移
+     * nullloop->keepalive loop->polling loop
+     */
     getPollingLoop(
       hander: Function,
       options: { step?: number; globalIntrupt?: boolean } = {
@@ -104,10 +114,16 @@ export const fetchStore = defineStore("fetch", {
       let id = -1;
       const { step, globalIntrupt } = options;
       const that = this;
-      const pollingLoop = {
+      // 很有可能是keepalive session loop
+      let preLoop = this.curPolling;
+      const pollingLoop: PollingLoop = {
         // 自动轮询的可能会被错误打断
         open() {
           if (!this.isOn()) {
+            // 切换为当前用到的 pollingLoop
+            preLoop.close();
+            that.curPolling = pollingLoop;
+
             if (globalIntrupt) that.jobRunning = true;
             hander();
             id = setInterval(
@@ -131,6 +147,10 @@ export const fetchStore = defineStore("fetch", {
             if (globalIntrupt) that.jobRunning = false;
             clearInterval(id);
             id = -1;
+            // 重置为默认的nullloop
+            that.curPolling = preLoop;
+            // 继续keepalive
+            preLoop.open();
           }
         },
         isOn() {
@@ -143,7 +163,6 @@ export const fetchStore = defineStore("fetch", {
           hander();
         },
       };
-      this.curPolling = pollingLoop;
       return pollingLoop;
     },
     pullResultsLoop(pollingM: Machine, globalIntrupt: boolean = true) {
@@ -163,6 +182,9 @@ export const fetchStore = defineStore("fetch", {
         },
       );
     },
+    /**
+     * 用来提示是否正在请求中
+     */
     onWait() {
       if (!this.wait) this.wait = true;
     },
@@ -245,10 +267,71 @@ export const fetchStore = defineStore("fetch", {
         },
       );
     },
+    keepaliveSession() {
+      const kl = () => {
+        let m = interpret(permachine);
+        m.start();
+        m.send("INIT");
+        m.send({
+          type: "SUBMIT",
+          value: {
+            action: "pull_results",
+            sessionId: undefined,
+            consumerId: undefined,
+          },
+        });
+        
+      };
+      kl();
+      let id = -1;
+      let that = this;
+      let loop = {
+        open() {
+          if (!this.isOn()) {
+            // 切换为kl_loop
+            that.curPolling.close();
+            that.curPolling = loop;
+            kl();
+            id = setInterval(
+              (() => {
+                if (
+                  // 不在线或者意外报错就停掉
+                  publicStore().isErr || !that.online
+                ) {
+                  this.close();
+                } else {
+                  kl();
+                }
+              }) as TimerHandler,
+              60_000,
+            );
+          }
+        },
+        close() {
+          if (this.isOn()) {
+            clearInterval(id);
+            id = -1;
+            // 重置为默认的nullloop
+            that.curPolling = nullLoop;
+          }
+        },
+        isOn() {
+          return id !== -1;
+        },
+        invoke() {},
+      };
+      // 会先运行一次，当有任务执行
+      loop.open();
+      return loop;
+    },
     initSession() {
-      return this.baseSubmit(interpret(permachine), {
+      let p1 = this.baseSubmit(interpret(permachine), {
         action: "init_session",
+      }).then((res) => {
+        // 自动调度,维持session活性
+        this.keepaliveSession();
       });
+      return p1
     },
     asyncInit() {
       if (!this.online) {
@@ -263,5 +346,11 @@ export const fetchStore = defineStore("fetch", {
       }
       return Promise.resolve("alrealy init");
     },
+    closeSession() {
+      return this.baseSubmit(interpret(permachine), {
+        action: "close_session",
+        sessionId: undefined,
+      });
+    },
   },
 });