fix: 修改了 dashboard 的获取数据方式

Author: tianxiu2b2t

assets/js/index.js

@@ -17,38 +17,71 @@ class Utils {
                 }
             }
         }
-        return uuid.join('');
+        return uuid.join('').toLocaleLowerCase();
     }
 }
 
 class Socket {
     constructor(baseurl) {
-        this._baseurl = baseurl;
         this._ws = null;
         this._wsReconnectTask = null;
+        this._clientId = Utils.uuid();
+        this._baseurl = baseurl;
+        this._url = this._baseurl + "?id=" + this._clientId
+        this._keepaliveTask = null;
+        this._echoCallbacks = {}
         this._wsConnect()
-        window.addEventListener("beforeinput", () => {
+        window.addEventListener("beforeunload", () => {
+            this.send("disconnect")
             this._ws?.close()
         })
+        setInterval(() => this._keepalive(), 5000);
+        this._keepalive();
     }
-    async _sendByFetch(event, data) {
-        var resp = await fetch({
-            "url": this._baseurl,
-            "body": JSON.stringify(
-                {
-                    "event": event,
-                    "data": data,
-                    "echo_id": Utils.uuid()
-                }
-            )
+    async _keepalive() {
+        console.log(await this.send("keepalive", {
+            "timestamp": new Date()
+        }))
+    }
+    async _sendByXHR(event, data) {
+        var xhr = new XMLHttpRequest();
+        var echo_id = Utils.uuid();
+        xhr.addEventListener("readystatechange", (event) => {
+            if (event.target.readyState == XMLHttpRequest.DONE) {
+                this._dispatchData(JSON.parse(xhr.response)); 
+            }
         })
-        return await resp.json()
+        xhr.open("POST", this._url);
+        return this._setEchoCallback(echo_id, () => {
+            xhr.send(JSON.stringify({
+                "event": event,
+                "data": data,
+                echo_id
+            }));
+            setTimeout(() => {
+                xhr.abort();
+            }, 10000)
+        });
+    }
+    _dispatchData(responses) {
+        responses.forEach(response => {
+            const { echo_id, event, data } = response;
+            console.log(echo_id, event)
+            if (echo_id == null) { // global dispatch event
+
+            } else if (echo_id in this._echoCallbacks) {
+                var { resolve, reject, timer } = this._echoCallbacks[echo_id];
+                delete this._echoCallbacks[echo_id];
+                clearTimeout(timer);
+                resolve(data);
+            }
+        });
     }
     _wsConnect() {
         clearTimeout(this._wsReconnectTask)
         this._ws?.close()
         this._ws = new WebSocket(
-            this._baseurl
+            this._url
         )
         this._ws.addEventListener("close", () => {
             console.warn("The websocket has disconnected. After 5s to reconnect.")
@@ -58,29 +91,36 @@ class Socket {
         })
         this._ws.addEventListener("message", (event) => {
             var raw_data = JSON.parse(event.data);
+            this._dispatchData(raw_data)
         })
     }
-    async _sendByWs(event, data) {
-        if (this._ws?.state != WebSocket.OPEN) return;
+    _sendByWs(event, data) {
+        if (this._ws?.readyState != WebSocket.OPEN) return;
         var echo_id = Utils.uuid();
-        this._ws.send({
-            "event": event,
-            "data": data,
-            echo_id
-        })
-        var promise = new Promise((resolve, reject) => {
-            
+        return this._setEchoCallback(echo_id, () => {
+            this._ws.send(JSON.stringify(
+                {
+                    "event": event,
+                    "data": data,
+                    echo_id
+                }
+            ))
         });
-        return promise;
+    }
+    _setEchoCallback(id, executor) {
+        return new Promise((resolve, reject) => {
+            this._echoCallbacks[id] = { resolve, reject, timer: setTimeout(() => {
+                reject("Timeout Error.")
+            }, 10000)};
+            executor();
+        })
     }
     async send(event, data) {
-        if (this._ws != null && this._ws.state == WebSocket.OPEN) {
-            return this._sendByWs(event, data)
-        } else {
-            return this._sendByFetch(event, data)
-        }
+        var handler = this._ws?.readyState == WebSocket.OPEN ? this._sendByWs : this._sendByXHR
+        return handler.bind(this)(event, data);
     }
 
 }
 
-const $socket = new Socket(window.location.origin + "/api")
+const $socket = new Socket(window.location.origin + "/api");
+$socket.send("echo", "hello world")

core/cluster.py

@@ -735,12 +735,14 @@ async def callback(data: tuple[Any, Any]):
         await self.sio.emit(
             event, data, callback=callback
         )
+        timeout_task = None
         if timeout is not None:
-            scheduler.run_later(lambda: fut.set_exception(asyncio.TimeoutError), timeout)
+            timeout_task = scheduler.run_later(lambda: not fut.done() and fut.set_exception(asyncio.TimeoutError), timeout)
         try:
             await fut
         except:
             raise
+        scheduler.cancel(timeout_task)
         return fut.result()
 
 @dataclass
@@ -785,7 +787,7 @@ def __init__(self, hash: str, size: int, mtime: float, data: bytes) -> None:
 
 ROOT = Path(__file__).parent.parent
 
-API_VERSION = "1.11.0"
+API_VERSION = "1.12.0"
 USER_AGENT = f"openbmclapi/{API_VERSION} python-openbmclapi/3.0"
 CHECK_FILE_CONTENT = "Python OpenBMCLAPI"
 CHECK_FILE_MD5 = hashlib.md5(CHECK_FILE_CONTENT.encode("utf-8")).hexdigest()

core/dashboard.py

@@ -3,11 +3,12 @@
 import json
 import os
 import socket
+from typing import Any, Optional
 
 import aiohttp
 import psutil
 
-from core import config
+from core import cache, config
 
 from . import utils
 
@@ -76,7 +77,54 @@ def get_json(self):
                  "qps": item.value.qps
              }} for item in self._data
         ]
+    
+class ClientRoom:
+    def __init__(self) -> None:
+        self.clients = cache.MemoryStorage()
+        self.__default_data = {
+            "lastKeepalive": 0,
+            "messages": []
+        }
+
+    @property
+    def _default_data(self):
+        return self.__default_data.copy()
 
+    def join_ws(self, id: str):
+        self.clients.set(id, self._default_data)
+    def join_http(self, id: str):
+        self.clients.set(id, self._default_data, 600)
+
+    def keepalive(self, id: str):
+        client = self.clients.get(id, self._default_data)
+        client["lastKeepalive"] = utils.get_runtime()
+        self.clients.set(id, client)
+
+    
+    def exit(self, id: str):
+        self.clients.delete(f"ws-{id}")
+        self.clients.delete(f"http-{id}")
+
+    def send(self, event: str, data: Any, id: Optional[str] = None):
+        ids = [_id for _id in self.clients.get_keys() if id is None or (id is not None and _id == id)]
+        for id in ids:
+            client = self.clients.get(id, self._default_data)
+            client["messages"].append({
+                "event": event,
+                "data": data
+            })
+            self.clients.set(id, client)
+
+    def get_messages(self, id: str):
+        client = self.clients.get(id, self._default_data)
+        data: list[Any] = client["messages"].copy()
+        client["messages"].clear()
+        self.clients.set(id, client)
+        return data
+    
+    
+
+room = ClientRoom()
 counter = Counter()
 process = psutil.Process(os.getpid())
 
@@ -110,27 +158,52 @@ async def _(request: web.Request):
 @route.get("/api")
 async def _(request: web.Request):
     if request.headers.get("Connection", "").lower() == "upgrade" and request.headers.get("Upgrade", "").lower() == "websocket":
-        ws = web.WebSocketResponse()
-        ws.can_prepare(request)
-        await ws.prepare(request)
-        await ws.send_json({
-            "event": "echo",
-            "data": "hello world",
-            "echo_id": None
-        })
-        while not ws.closed:
-            try:
-                raw_data = await ws.receive()
+        try:
+            ws = web.WebSocketResponse()
+            ws.can_prepare(request)
+            await ws.prepare(request)
+            id: str = request.query.get("id") # type: ignore
+            room.join_ws(id)
+            while not ws.closed:
                 try:
-                    print(raw_data)
+                    raw_data = json.loads((await ws.receive()).data)
+                    messages = room.get_messages(id)
+                    messages.append({
+                        "echo_id": raw_data.get("echo_id"),
+                        "event": raw_data["event"],
+                        "data": raw_data
+                    })
+                    if raw_data["event"] == "keepalive":
+                        room.keepalive(id)
+                    await ws.send_json(messages)
                 except:
-                    ...
-            except:
-                break
-        return ws
-    else:
-        return web.json_response({
+                    break
+            room.exit(id)
+            return ws
+        except:
+            ...
+    return web.HTTPNotFound()
+
+@route.post("/api")
+async def _(request: web.Request):
+    try:
+        id: str = request.query.get("id") # type: ignore
+        room.join_http(id)
+        body = await request.read()
+        raw_data = json.loads(body)
+        messages = room.get_messages(id)
+        messages.append({
+            "event": raw_data["event"],
+            "echo_id": raw_data.get("echo_id"),
+            "data": raw_data
         })
+        if raw_data["event"] == "disconnect":
+            room.exit(id)
+        if raw_data["event"] == "keepalive":
+            room.keepalive(id)
+        return web.json_response(messages)
+    except:
+        return web.HTTPInternalServerError()
 
 
 route.static("/assets", "./assets")

core/storages/__init__.py

@@ -338,6 +338,7 @@ def empty():
                         r["name"]
                     )
                     files[f"{root_id:02x}"].append(file)
+                await asyncio.sleep(0)
                 update_tqdm()
 
         return files