reliably send msg from client to server when use HTTP

Author: wang0618

pywebio/platform/adaptor/http.py

@@ -14,7 +14,7 @@
 import threading
 import time
 from contextlib import contextmanager
-from typing import Dict, Optional
+from typing import Dict, Optional, List
 from collections import deque
 
 from ..page import make_applications, render_page
@@ -59,16 +59,6 @@ def request_body(self) -> bytes:
         """
         return b''
 
-    def request_json(self) -> Optional[Dict]:
-        """返回当前请求的json反序列化后的内容，若请求数据不为json格式，返回None
-        Return the data (json deserialization) of the currently requested, if the data is not in json format, return None"""
-        try:
-            if self.request_headers().get('content-type') == 'application/octet-stream':
-                return deserialize_binary_event(self.request_body())
-            return json.loads(self.request_body())
-        except Exception:
-            return None
-
     def set_header(self, name, value):
         """为当前响应设置header
         Set a header for the current response"""
@@ -109,7 +99,7 @@ def __init__(self, session: Session, message_window: int = 4):
         self.messages = deque()
         self.window_size = message_window
         self.min_msg_id = 0  # the id of the first message in the window
-        self.next_event_id = 0
+        self.finished_event_id = -1  # the id of the last finished event
 
     @staticmethod
     def close_message(ack):
@@ -118,6 +108,20 @@ def close_message(ack):
             seq=ack + 1
         )
 
+    def push_event(self, events: List[Dict], seq: int) -> int:
+        """Send client events to the session and return the success message count"""
+        if not events:
+            return 0
+
+        submit_cnt = 0
+        for eid, event in enumerate(events, start=seq):
+            if eid > self.finished_event_id:
+                self.finished_event_id = eid  # todo: use lock for check and set operation
+                self.session.send_client_event(event)
+                submit_cnt += 1
+
+        return submit_cnt
+
     def get_response(self, ack=0):
         """
         ack num is the number of messages that the client has received.
@@ -134,7 +138,8 @@ def get_response(self, ack=0):
 
         return dict(
             commands=list(self.messages),
-            seq=self.min_msg_id
+            seq=self.min_msg_id,
+            ack=self.finished_event_id
         )
 
 
@@ -179,6 +184,7 @@ def _remove_expired_sessions(cls, session_expire_seconds):
             if session:
                 session.close(nonblock=True)
                 del cls._webio_sessions[sid]
+                del cls._webio_transports[sid]
 
     @classmethod
     def _remove_webio_session(cls, sid):
@@ -234,6 +240,14 @@ def get_cdn(self, context):
             return False
         return self.cdn
 
+    def read_event_data(self, context: HttpContext) -> List[Dict]:
+        try:
+            if context.request_headers().get('content-type') == 'application/octet-stream':
+                return [deserialize_binary_event(context.request_body())]
+            return json.loads(context.request_body())
+        except Exception:
+            return []
+
     @contextmanager
     def handle_request_context(self, context: HttpContext):
         """called when every http request"""
@@ -304,8 +318,10 @@ def handle_request_context(self, context: HttpContext):
             webio_session = cls._webio_sessions[webio_session_id]
 
         if context.request_method() == 'POST':  # client push event
-            if context.request_json() is not None:
-                webio_session.send_client_event(context.request_json())
+            seq = int(context.request_url_parameter('seq', 0))
+            event_data = self.read_event_data(context)
+            submit_cnt = cls._webio_transports[webio_session_id].push_event(event_data, seq)
+            if submit_cnt > 0:
                 yield type(self).WAIT_MS_ON_POST / 1000.0  # <--- <--- <--- <--- <--- <--- <--- <--- <--- <--- <--- <---
         elif context.request_method() == 'GET':  # client pull messages
             pass

webiojs/src/session.ts

@@ -1,4 +1,4 @@
-import {error_alert, randomid} from "./utils";
+import {error_alert, randomid, ReliableSender} from "./utils";
 import {state} from "./state";
 import {t} from "./i18n";
 
@@ -181,6 +181,7 @@ export class HttpSession implements Session {
     webio_session_id: string = '';
     debug = false;
 
+    private sender: ReliableSender = null;
     private _executed_command_msg_id = -1;
     private _closed = false;
     private _session_create_callbacks: (() => void)[] = [];
@@ -193,6 +194,7 @@ export class HttpSession implements Session {
         let url = new URL(api_url, window.location.href);
         url.search = "?app=" + app_name;
         this.api_url = url.href;
+        this.sender = new ReliableSender(this._send.bind(this));
     }
 
     on_session_create(callback: () => void): void {
@@ -224,21 +226,21 @@ export class HttpSession implements Session {
             contentType: "application/json; charset=utf-8",
             dataType: "json",
             headers: {"webio-session-id": this.webio_session_id},
-            success: function (data: { commands: Command[][], seq: number, event: number },
+            success: function (data: { commands: Command[][], seq: number, event: number, ack: number },
                                textStatus: string, jqXHR: JQuery.jqXHR) {
                 safe_poprun_callbacks(that._session_create_callbacks, 'session_create_callback');
                 that._on_request_success(data, textStatus, jqXHR);
-                if(that.webio_session_id.startsWith("NEW-")){
+                if (that.webio_session_id.startsWith("NEW-")) {
                     that.webio_session_id = that.webio_session_id.substring(4);
                 }
-            },
-            error: function () {
-                console.error('Http pulling failed');
             }
         })
     }
 
-    private _on_request_success(data: { commands: Command[][], seq: number }, textStatus: string, jqXHR: JQuery.jqXHR) {
+    private _on_request_success(data: { commands: Command[][], seq: number, ack: number },
+                                textStatus: string, jqXHR: JQuery.jqXHR) {
+        this.sender.ack(data.ack);
+
         let msg_start_idx = this._executed_command_msg_id - data.seq + 1;
         if (data.commands.length <= msg_start_idx)
             return;
@@ -258,54 +260,78 @@ export class HttpSession implements Session {
 
     send_message(msg: ClientEvent, onprogress?: (loaded: number, total: number) => void): void {
         if (this.debug) console.info('<<<', msg);
-        this._send({
-            data: JSON.stringify(msg),
-            contentType: "application/json; charset=utf-8",
-        }, onprogress);
+        this.sender.add_send_task({
+            data: msg,
+            json: true,
+            onprogress: onprogress,
+        })
     }
 
     send_buffer(data: Blob, onprogress?: (loaded: number, total: number) => void): void {
         if (this.debug) console.info('<<< Blob data...');
-        this._send({
+        this.sender.add_send_task({
             data: data,
-            cache: false,
-            processData: false,
-            contentType: 'application/octet-stream',
-        }, onprogress);
+            json: false,
+            onprogress: onprogress,
+        }, false)
     }
 
-    _send(options: { [key: string]: any; }, onprogress?: (loaded: number, total: number) => void): void {
-        if (this.closed())
-            return error_alert(t("disconnected_with_server"));
-
-        $.ajax({
-            ...options,
-            type: "POST",
-            url: `${this.api_url}&ack=${this._executed_command_msg_id}`,
-            dataType: "json",
-            headers: {"webio-session-id": this.webio_session_id},
-            success: this._on_request_success.bind(this),
-            xhr: function () {
-                let xhr = new window.XMLHttpRequest();
-                // Upload progress
-                xhr.upload.addEventListener("progress", function (evt) {
-                    if (evt.lengthComputable && onprogress) {
-                        onprogress(evt.loaded, evt.total);
-                    }
-                }, false);
-                return xhr;
-            },
-            error: function () {
-                console.error('Http push blob data failed');
-                error_alert(t("connect_fail"));
+    _send(params: { [key: string]: any; }[], seq: number): Promise<void> {
+        if (this.closed()) {
+            this.sender.stop();
+            error_alert(t("disconnected_with_server"));
+            return Promise.reject();
+        }
+        let data: any, ajax_options: any;
+        let json = params.some(p => p.json);
+        if (json) {
+            data = JSON.stringify(params.map(p => p.data));
+            ajax_options = {
+                contentType: "application/json; charset=utf-8",
             }
+        } else {
+            data = params[0].data;
+            ajax_options = {
+                cache: false,
+                processData: false,
+                contentType: 'application/octet-stream',
+            }
+        }
+        return new Promise((resolve, reject) => {
+            $.ajax({
+                data: data,
+                ...ajax_options,
+                type: "POST",
+                url: `${this.api_url}&ack=${this._executed_command_msg_id}&seq=${seq}`,
+                dataType: "json",
+                headers: {"webio-session-id": this.webio_session_id},
+                success: this._on_request_success.bind(this),
+                xhr: function () {
+                    let xhr = new window.XMLHttpRequest();
+                    // Upload progress
+                    xhr.upload.addEventListener("progress", function (evt) {
+                        if (evt.lengthComputable) {
+                            params.forEach(p => {
+                                if (p.onprogress) // only the first one
+                                    p.onprogress(evt.loaded, evt.total);
+                                p.onprogress = null;
+                            });
+                        }
+                    }, false);
+                    return xhr;
+                },
+                error: function () {
+                    console.error('Http push event failed, will retry');
+                }
+            }).always(() => resolve());
         });
     }
 
     close_session(): void {
         this._closed = true;
         safe_poprun_callbacks(this._session_close_callbacks, 'session_close_callback');
         clearInterval(this.interval_pull_id);
+        this.sender.stop();
     }
 
     closed(): boolean {

webiojs/src/utils.ts

@@ -183,4 +183,82 @@ export function is_mobile() {
     if (navigator.userAgentData) return navigator.userAgentData.mobile;
     const ipadOS = (navigator.platform === 'MacIntel' && navigator.maxTouchPoints > 1); /* iPad OS 13 */
     return /android|webos|iphone|ipad|ipod|blackberry|iemobile|opera mini/i.test(navigator.userAgent.toLowerCase()) || ipadOS;
+}
+
+// put send task to a queue and run it one by one
+export class ReliableSender {
+    private seq = 0;
+    private queue: { enable_batch: boolean, param: any }[] = [];
+    private send_running = false
+    private _stop = false;
+
+    constructor(
+        private readonly sender: (params: any[], seq: number) => Promise<void>,
+        private window_size: number = 8,
+        init_seq = 0, private timeout = 2000
+    ) {
+        this.sender = sender;
+        this.window_size = window_size;
+        this.timeout = timeout;
+        this.seq = init_seq;
+        this.queue = [];
+    }
+
+    /*
+    * for continuous batch_send tasks in queue, they will be sent in one sender, the sending will retry when it finished or timeout.
+    * for non-batch task, each will be sent in a single sender, the sending will retry when it finished.
+    * */
+    add_send_task(param: any, allow_batch_send = true) {
+        if (this._stop) return;
+        this.queue.push({
+            enable_batch: allow_batch_send,
+            param: param
+        });
+        if (!this.send_running)
+            this.start_send();
+    }
+
+    private start_send() {
+        if (this._stop || this.queue.length === 0) {
+            this.send_running = false;
+            return;
+        }
+        this.send_running = true;
+        let params: any[] = [];
+        for (let item of this.queue) {
+            if (!item.enable_batch)
+                break;
+            params.push(item.param);
+        }
+        let batch_send = true;
+        if (params.length === 0 && !this.queue[0].enable_batch) {
+            batch_send = false;
+            params.push(this.queue[0].param);
+        }
+        if (params.length === 0) {
+            this.send_running = false;
+            return;
+        }
+
+        let promises = [this.sender(params, this.seq)];
+        if (batch_send)
+            promises.push(new Promise((resolve) => setTimeout(resolve, this.timeout)));
+
+        Promise.race(promises).then(() => {
+            this.start_send();
+        });
+    }
+
+    // seq for each ack call must be larger than the previous one, otherwise the ack will be ignored
+    ack(seq: number) {
+        if (seq < this.seq)
+            return;
+        let pop_count = seq - this.seq + 1;
+        this.queue = this.queue.slice(pop_count);
+        this.seq = seq + 1;
+    }
+
+    stop() {
+        this._stop = true;
+    }
 }