Save socket.io missing messages during reconnection (#937)

* Save socket.io missing messages during reconnection

* Change protocol version for compatibility

* Add more description in documents

Author: starwarfan

doc/Client-Portal Protocol.md

@@ -6,6 +6,7 @@
 |28-06-2017 |Xiande|1.0 reviewed|
 |08-04-2018 |Tianfang|1.0 final|
 |08-09-2019 |ChenLi|1.1 reviewed|
+|03-10-2021 |ChenLi|1.2 reviewed|
 -->
 # 1. Overview
 This documentation covers all signaling messages between Client and MCU component through Socket.io connections.
@@ -90,7 +91,21 @@ This a format for client reconnects.
 
 **RequestName**: "relogin"<br>
 
-**RequestData**: The ReconnectionTicket object defined in 3.3.1 section.<br>
+**RequestData**: The ReconnectionTicket object defined in 3.3.1 section and PendingMessages:
+
+```
+  object(ReloginResponse)::
+    {
+      ticket: string(Base64Encoded(object(ReconnectionTicket)))
+      messages: [
+        {
+          event: string(NotificationName),
+          data: object(NotificationData),
+          seq: number(InternalSeqNo) // Back to 0 if exceed max value 2147483647
+        }
+      ]
+    }
+```
 
 **ResponseData**: A refreshed base64-encoded ReconnectionTicket object if ResponseStatus is "ok".
 ## 3.3 Conferencing

doc/design/pics/relogin1.png

@@ -0,0 +1,29 @@
+Server and client both maintain an **internal sequence number** for
+sent/received messages once a connection established.
+
+For Server:
+
+1.  After a client has logged in, allocate a sequence number with
+    initial value 0 for this connection.
+
+2.  Once a message is sent (no matter success or not), bind the message
+    with current sequence number and time, save them in a buffer.
+    Increase the current sequence number by 1. Discard the old messages after
+    a certain period if necessary.
+
+3.  Once client “relogin”, sent those messages with sequence number in
+    buffer in response.
+
+For Client:
+
+1.  After login succeed, allocate a sequence number with initial
+    value 0.
+
+2.  Once a message is received, increase the current sequence number
+    by 1.
+
+3.  Once client “relogin” succeed, filter messages with sequence
+    number >= client sequence number, process those messages in
+    order. Update its own current sequence number.
+
+

doc/design/pics/relogin2.png

@@ -41,6 +41,11 @@ var Connection = function(spec, socket, reconnectionKey, portal, dock) {
   var client_id;
   var protocol_version;
   var waiting_for_reconnecting_timer = null;
+  // [{ event, data, seq, time }]
+  // Use queue instead of array if its size is very large
+  var message_seq = 0;
+  var message_buffer = [];
+  var message_keep_time = spec.pingInterval + spec.pingTimeout + 1;
   var pending_messages = [];
 
   let reconnection = {
@@ -225,14 +230,24 @@ var Connection = function(spec, socket, reconnectionKey, portal, dock) {
           client_id = connectionInfo.clientId + '';
           protocol_version = connectionInfo.protocolVersion + '';
           pending_messages = connectionInfo.pendingMessages;
+          message_seq = connectionInfo.messageSeq;
+          message_buffer = connectionInfo.messageBuffer;
           reconnection.enabled = true;
           return client.resetConnection(that);
         }
       }).then(() => {
         let ticket = generateReconnectionTicket();
+        let messages = message_buffer.map(msg => {
+          delete msg.time;
+          return msg;
+        });
         state = 'connected';
-        safeCall(callback, okWord(), ticket);
-        drainPendingMessages();
+        if (protocol_version === '1.2') {
+          safeCall(callback, okWord(), {ticket, messages});
+        } else {
+          safeCall(callback, okWord(), ticket);
+          drainPendingMessages();
+        }
       }).catch((err) => {
         state = 'initialized';
         const err_message = getErrorMessage(err);
@@ -298,6 +313,8 @@ var Connection = function(spec, socket, reconnectionKey, portal, dock) {
 
     return {
       pendingMessages: pending_messages,
+      messageSeq: message_seq,
+      messageBuffer: message_buffer,
       clientId: client_id,
       protocolVersion: protocol_version,
       reconnection: reconnection
@@ -315,6 +332,16 @@ var Connection = function(spec, socket, reconnectionKey, portal, dock) {
     } else {
       pending_messages.push({event: event, data: data});
     }
+    let currentTime = process.hrtime()[0];
+    message_seq++;
+    message_buffer.push({event, data, seq: message_seq, time: currentTime});
+    while (message_buffer[0]) {
+      if (currentTime - message_buffer[0].time > message_keep_time) {
+        message_buffer.shift();
+      } else {
+        break;
+      }
+    }
   };
 
   that.close = (ifLeave) => {