use receiver window in http session

Author: wang0618

pywebio/platform/adaptor/http.py

@@ -15,11 +15,12 @@
 import time
 from contextlib import contextmanager
 from typing import Dict, Optional
+from collections import deque
 
 from ..page import make_applications, render_page
 from ..utils import deserialize_binary_event
 from ...session import CoroutineBasedSession, ThreadBasedSession, register_session_implement_for_target
-from ...session.base import get_session_info_from_headers
+from ...session.base import get_session_info_from_headers, Session
 from ...utils import random_str, LRUDict, isgeneratorfunction, iscoroutinefunction, check_webio_js
 
 
@@ -102,6 +103,41 @@ def get_client_ip(self):
 _event_loop = None
 
 
+class ReliableTransport:
+    def __init__(self, session: Session, message_window: int = 4):
+        self.session = session
+        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
+
+    @staticmethod
+    def close_message(ack):
+        return dict(
+            commands=[[dict(command='close_session')]],
+            seq=ack+1
+        )
+
+    def get_response(self, ack=0):
+        """
+        ack num is the number of messages that the client has received.
+        response is a list of messages that the client should receive, along with their min id `seq`.
+        """
+        while ack >= self.min_msg_id and self.messages:
+            self.messages.popleft()
+            self.min_msg_id += 1
+
+        if len(self.messages) < self.window_size:
+            msgs = self.session.get_task_commands()
+            if msgs:
+                self.messages.append(msgs)
+
+        return dict(
+            commands=list(self.messages),
+            seq=self.min_msg_id
+        )
+
+
 # todo: use lock to avoid thread race condition
 class HttpHandler:
     """基于HTTP的后端Handler实现
@@ -112,7 +148,7 @@ class HttpHandler:
 
     """
     _webio_sessions = {}  # WebIOSessionID -> WebIOSession()
-    _webio_last_commands = {}  # WebIOSessionID -> (last commands, commands sequence id)
+    _webio_transports = {}  # WebIOSessionID -> ReliableTransport(), type: Dict[str, ReliableTransport]
     _webio_expire = LRUDict()  # WebIOSessionID -> last active timestamp. In increasing order of last active time
     _webio_expire_lock = threading.Lock()
 
@@ -149,17 +185,6 @@ def _remove_webio_session(cls, sid):
         cls._webio_sessions.pop(sid, None)
         cls._webio_expire.pop(sid, None)
 
-    @classmethod
-    def get_response(cls, sid, ack=0):
-        commands, seq = cls._webio_last_commands.get(sid, ([], 0))
-        if ack == seq:
-            webio_session = cls._webio_sessions[sid]
-            commands = webio_session.get_task_commands()
-            seq += 1
-            cls._webio_last_commands[sid] = (commands, seq)
-
-        return {'commands': commands, 'seq': seq}
-
     def _process_cors(self, context: HttpContext):
         """Handling cross-domain requests: check the source of the request and set headers"""
         origin = context.request_headers().get('Origin', '')
@@ -240,8 +265,8 @@ def handle_request_context(self, context: HttpContext):
             context.set_content(html)
             return context.get_response()
 
+        ack = int(context.request_url_parameter('ack', 0))
         webio_session_id = None
-
         # 初始请求，创建新 Session
         if not request_headers['webio-session-id'] or request_headers['webio-session-id'] == 'NEW':
             if context.request_method() == 'POST':  # 不能在POST请求中创建Session，防止CSRF攻击
@@ -264,9 +289,11 @@ def handle_request_context(self, context: HttpContext):
                 session_cls = ThreadBasedSession
             webio_session = session_cls(application, session_info=session_info)
             cls._webio_sessions[webio_session_id] = webio_session
+            cls._webio_transports[webio_session_id] = ReliableTransport(webio_session)
             yield type(self).WAIT_MS_ON_POST / 1000.0  # <--- <--- <--- <--- <--- <--- <--- <--- <--- <--- <--- <---
         elif request_headers['webio-session-id'] not in cls._webio_sessions:  # WebIOSession deleted
-            context.set_content([dict(command='close_session')], json_type=True)
+            close_msg = ReliableTransport.close_message(ack)
+            context.set_content(close_msg, json_type=True)
             return context.get_response()
         else:
             webio_session_id = request_headers['webio-session-id']
@@ -283,8 +310,8 @@ def handle_request_context(self, context: HttpContext):
 
         self.interval_cleaning()
 
-        ack = int(context.request_url_parameter('ack', 0))
-        context.set_content(type(self).get_response(webio_session_id, ack=ack), json_type=True)
+        resp = cls._webio_transports[webio_session_id].get_response(ack)
+        context.set_content(resp, json_type=True)
 
         if webio_session.closed():
             self._remove_webio_session(webio_session_id)

webiojs/src/session.ts

@@ -181,7 +181,7 @@ export class HttpSession implements Session {
     webio_session_id: string = 'NEW';
     debug = false;
 
-    private _executed_command_msg_id = 0;
+    private _executed_command_msg_id = -1;
     private _closed = false;
     private _session_create_callbacks: (() => void)[] = [];
     private _session_close_callbacks: (() => void)[] = [];
@@ -223,7 +223,7 @@ 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 }, textStatus: string, jqXHR: JQuery.jqXHR) {
+            success: function (data: { commands: Command[][], seq: number }, textStatus: string, jqXHR: JQuery.jqXHR) {
                 safe_poprun_callbacks(that._session_create_callbacks, 'session_create_callback');
                 that._on_request_success(data, textStatus, jqXHR);
             },
@@ -233,18 +233,21 @@ export class HttpSession implements Session {
         })
     }
 
-    private _on_request_success(data: { commands: Command[], seq: number }, textStatus: string, jqXHR: JQuery.jqXHR) {
-        if (data.seq == this._executed_command_msg_id)
+    private _on_request_success(data: { commands: Command[][], seq: number }, textStatus: string, jqXHR: JQuery.jqXHR) {
+        let msg_start_idx = this._executed_command_msg_id - data.seq + 1;
+        if (data.commands.length <= msg_start_idx)
             return;
-        this._executed_command_msg_id = data.seq;
+        this._executed_command_msg_id = data.seq + data.commands.length - 1;
 
         let sid = jqXHR.getResponseHeader('webio-session-id');
         if (sid)
             this.webio_session_id = sid;
 
-        for (let msg of data.commands) {
-            if (this.debug) console.info('>>>', msg);
-            this._on_server_message(msg);
+        for (let msgs of data.commands.slice(msg_start_idx)) {
+            for (let msg of msgs) {
+                if (this.debug) console.info('>>>', msg);
+                this._on_server_message(msg);
+            }
         }
     };