html: implement flow control for xterm

https://xtermjs.org/docs/guides/flowcontrol/#flow-control-over-websockets

html/src/components/terminal/index.tsx

@@ -7,7 +7,7 @@ import { WebglAddon } from 'xterm-addon-webgl';
 import { WebLinksAddon } from 'xterm-addon-web-links';
 
 import { OverlayAddon } from './overlay';
-import { ZmodemAddon } from '../zmodem';
+import { ZmodemAddon, FlowControl } from '../zmodem';
 
 import 'xterm/css/xterm.css';
 
@@ -30,6 +30,8 @@ const enum Command {
     // client side
     INPUT = '0',
     RESIZE_TERMINAL = '1',
+    PAUSE = '2',
+    RESUME = '3',
 }
 
 export interface ClientOptions {
@@ -112,13 +114,33 @@ export class Xterm extends Component<Props> {
     }
 
     render({ id }: Props) {
+        const control = {
+            limit: 100000,
+            highWater: 10,
+            lowWater: 4,
+            pause: () => this.pause(),
+            resume: () => this.resume(),
+        } as FlowControl;
+
         return (
             <div id={id} ref={c => (this.container = c)}>
-                <ZmodemAddon ref={c => (this.zmodemAddon = c)} sender={this.sendData} />
+                <ZmodemAddon ref={c => (this.zmodemAddon = c)} sender={this.sendData} control={control} />
             </div>
         );
     }
 
+    @bind
+    private pause() {
+        const { textEncoder, socket } = this;
+        socket.send(textEncoder.encode(Command.PAUSE));
+    }
+
+    @bind
+    private resume() {
+        const { textEncoder, socket } = this;
+        socket.send(textEncoder.encode(Command.RESUME));
+    }
+
     @bind
     private sendData(data: ArrayLike<number>) {
         const { socket } = this;

html/src/components/zmodem/index.tsx

@@ -6,8 +6,18 @@ import * as Zmodem from 'zmodem.js/src/zmodem_browser';
 
 import { Modal } from '../modal';
 
+export interface FlowControl {
+    limit: number;
+    highWater: number;
+    lowWater: number;
+
+    pause: () => void;
+    resume: () => void;
+}
+
 interface Props {
     sender: (data: ArrayLike<number>) => void;
+    control: FlowControl;
 }
 
 interface State {
@@ -20,6 +30,9 @@ export class ZmodemAddon extends Component<Props, State> implements ITerminalAdd
     private sentry: Zmodem.Sentry;
     private session: Zmodem.Session;
 
+    private written = 0;
+    private pending = 0;
+
     constructor(props: Props) {
         super(props);
 
@@ -84,7 +97,26 @@ export class ZmodemAddon extends Component<Props, State> implements ITerminalAdd
 
     @bind
     private zmodemWrite(data: ArrayBuffer): void {
-        this.terminal.write(new Uint8Array(data));
+        const { limit, highWater, lowWater, pause, resume } = this.props.control;
+        const { terminal } = this;
+        const rawData = new Uint8Array(data);
+
+        this.written += rawData.length;
+        if (this.written > limit) {
+            terminal.write(rawData, () => {
+                this.pending = Math.max(this.pending - 1, 0);
+                if (this.pending < lowWater) {
+                    resume();
+                }
+            });
+            this.pending++;
+            this.written = 0;
+            if (this.pending > highWater) {
+                pause();
+            }
+        } else {
+            terminal.write(rawData);
+        }
     }
 
     @bind

src/protocol.c

@@ -383,6 +383,18 @@ int callback_tty(struct lws *wsi, enum lws_callback_reasons reason, void *user,
             }
           }
         } break;
+        case PAUSE:
+          if (proc->state == STATE_INIT) {
+            uv_read_stop((uv_stream_t *)&proc->pipe);
+            proc->state = STATE_PAUSE;
+          }
+          break;
+        case RESUME:
+          if (proc->state == STATE_PAUSE) {
+            uv_read_start((uv_stream_t *)&proc->pipe, alloc_cb, read_cb);
+            proc->state = STATE_INIT;
+          }
+          break;
         case JSON_DATA:
           if (proc->pid > 0) break;
           if (server->credential != NULL) {

src/server.h

@@ -6,6 +6,8 @@
 // client message
 #define INPUT '0'
 #define RESIZE_TERMINAL '1'
+#define PAUSE '2'
+#define RESUME '3'
 #define JSON_DATA '{'
 
 // server message
@@ -26,7 +28,7 @@ extern struct lws_context *context;
 extern struct server *server;
 extern struct endpoints endpoints;
 
-typedef enum { STATE_INIT, STATE_KILL, STATE_EXIT } proc_state;
+typedef enum { STATE_INIT, STATE_PAUSE, STATE_KILL, STATE_EXIT } proc_state;
 
 struct pss_http {
   char path[128];