feat: the browser terminal on a WebSocket too, long polls when it does not connect or drops

internal/server/handlers_panel_term.go

@@ -1,9 +1,14 @@
 package server
 
 import (
+	"context"
 	"encoding/json"
 	"log"
 	"net/http"
+	"time"
+
+	"github.com/coder/websocket"
+	"github.com/coder/websocket/wsjson"
 
 	"github.com/x-grk/longpolling-remote-access/internal/proto"
 )
@@ -66,8 +71,14 @@ func (s *Server) handlePanelIn(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, "bad request", http.StatusBadRequest)
 		return
 	}
+	takePanelIn(sess, m)
+	w.WriteHeader(http.StatusNoContent)
+}
+
+// takePanelIn passes the browser's keystrokes or resize on to the agent; a
+// repeat of one already taken is dropped.
+func takePanelIn(sess *Session, m panelInMsg) {
 	if m.Seq > 0 && !sess.TakeSeq(m.Seq) {
-		w.WriteHeader(http.StatusNoContent) // a repeat of one already taken
 		return
 	}
 	switch m.T {
@@ -76,7 +87,70 @@ func (s *Server) handlePanelIn(w http.ResponseWriter, r *http.Request) {
 	case "r":
 		sess.PushIn(proto.InFrame{Type: proto.FrameResize, Cols: m.Cols, Rows: m.Rows})
 	}
-	w.WriteHeader(http.StatusNoContent)
+}
+
+// GET /api/panel/term/{sid}/ws - the browser's terminal on a WebSocket: what
+// the "in" requests send comes as messages, and output goes out as soon as
+// it is there, in the "out" answers' shape. The page falls back to those
+// requests when this does not connect.
+func (s *Server) handlePanelTermWS(w http.ResponseWriter, r *http.Request) {
+	sess := s.hub.Session(r.PathValue("sid"))
+	if sess == nil {
+		http.Error(w, "closed", http.StatusGone)
+		return
+	}
+	conn, err := websocket.Accept(w, r, &websocket.AcceptOptions{CompressionMode: websocket.CompressionDisabled})
+	if err != nil {
+		return
+	}
+	conn.SetReadLimit(1 << 20)
+	defer conn.CloseNow()
+	ctx, cancel := context.WithCancel(r.Context())
+	defer cancel()
+	go func() {
+		defer cancel()
+		for ctx.Err() == nil {
+			msgs, alive := sess.PollOutUntil(ctx.Done(), pollHold)
+			if ctx.Err() != nil {
+				return
+			}
+			out := map[string]any{"frames": msgs}
+			if !alive {
+				out = map[string]any{"closed": true}
+			}
+			wctx, wcancel := context.WithTimeout(ctx, 30*time.Second)
+			err := wsjson.Write(wctx, conn, out)
+			wcancel()
+			if err != nil || !alive {
+				return
+			}
+		}
+	}()
+	go func() {
+		t := time.NewTicker(wsPing)
+		defer t.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-t.C:
+				pctx, pcancel := context.WithTimeout(ctx, 15*time.Second)
+				err := conn.Ping(pctx)
+				pcancel()
+				if err != nil {
+					cancel()
+					return
+				}
+			}
+		}
+	}()
+	for {
+		var m panelInMsg
+		if err := wsjson.Read(ctx, conn, &m); err != nil {
+			return
+		}
+		takePanelIn(sess, m)
+	}
 }
 
 // POST /api/panel/term/{sid}/close - tear the session down.

internal/server/hub.go

@@ -324,10 +324,18 @@ func (s *Session) Out() <-chan proto.OutMessage { return s.out }
 
 // PollOut drains queued upstream messages, blocking up to timeout for the first.
 func (s *Session) PollOut(timeout time.Duration) ([]proto.OutMessage, bool) {
+	return s.PollOutUntil(nil, timeout)
+}
+
+// PollOutUntil is PollOut that also gives up when done closes, taking
+// nothing: for a reader that has gone, so no output is lost on it.
+func (s *Session) PollOutUntil(done <-chan struct{}, timeout time.Duration) ([]proto.OutMessage, bool) {
 	var batch []proto.OutMessage
 	select {
 	case m := <-s.out:
 		batch = append(batch, m)
+	case <-done:
+		return nil, true
 	case <-time.After(timeout):
 		return nil, true
 	case <-s.closed:

internal/server/server.go

@@ -249,6 +249,7 @@ func (s *Server) Handler() http.Handler {
 	mux.Handle("POST /api/panel/term/{agentID}/open", s.requireAdmin(s.handlePanelOpen))
 	mux.Handle("GET /api/panel/term/{sid}/out", s.requireAdmin(s.handlePanelOut))
 	mux.Handle("POST /api/panel/term/{sid}/in", s.requireAdmin(s.handlePanelIn))
+	mux.Handle("GET /api/panel/term/{sid}/ws", s.requireAdmin(s.handlePanelTermWS))
 	mux.Handle("POST /api/panel/term/{sid}/close", s.requireAdmin(s.handlePanelClose))
 
 	// Panel file manager (admin-only).

internal/server/web/static/terminal.js

@@ -76,13 +76,47 @@
         this.sid = (await r.json()).session_id;
         this.setState('wait', 'ожидание агента…');
         this.fit();
+        // A WebSocket if it connects; the long polls otherwise, and from a
+        // dropped WebSocket on for the rest of this session.
+        if (!(await this.openWS())) this.poll();
         this.queueIn({ t: 'r', cols: this.term.cols, rows: this.term.rows });
-        this.poll();
       } catch (e) {
         this.setState('err', 'ошибка сети');
         this.term.write('\r\n\x1b[31m[ошибка сети: ' + (e && e.message ? e.message : e) + ']\x1b[0m\r\n');
       }
     }
+    openWS() {
+      return new Promise(resolve => {
+        if (!window.WebSocket) { resolve(false); return; }
+        const ws = new WebSocket((location.protocol === 'https:' ? 'wss://' : 'ws://') + location.host + base + this.sid + '/ws');
+        let opened = false;
+        const timer = setTimeout(() => { if (!opened) { ws.close(); resolve(false); } }, 5000);
+        ws.onopen = () => { opened = true; clearTimeout(timer); this.ws = ws; resolve(true); };
+        ws.onmessage = ev => {
+          let j;
+          try { j = JSON.parse(ev.data); } catch (e) { return; }
+          this.take(j);
+        };
+        ws.onclose = () => {
+          clearTimeout(timer);
+          if (!opened) { resolve(false); return; }
+          this.ws = null;
+          if (!this.closed && !this.ended) this.poll();
+        };
+      });
+    }
+    // take applies one answer of the output channel, either transport's.
+    take(j) {
+      if (j.closed) { this.finish('[сессия завершена]'); return false; }
+      for (const f of j.frames || []) {
+        if (f.exit) { this.finish('[сессия завершена]'); return false; }
+        if (f.data) {
+          if (!this.connected) { this.connected = true; this.setState('ok', 'подключено'); }
+          this.term.write(b64ToBytes(f.data));
+        }
+      }
+      return true;
+    }
     async poll() {
       while (!this.closed && this.sid) {
         let r;
@@ -90,15 +124,7 @@
         catch (e) { await new Promise(res => setTimeout(res, 1000)); continue; }
         if (this.closed) return;
         if (!r.ok) { this.finish('[соединение закрыто]'); return; }
-        const j = await r.json();
+        if (!this.take(await r.json())) return;
-        if (j.closed) { this.finish('[сессия завершена]'); return; }
-        for (const f of j.frames || []) {
-          if (f.exit) { this.finish('[сессия завершена]'); return; }
-          if (f.data) {
-            if (!this.connected) { this.connected = true; this.setState('ok', 'подключено'); }
-            this.term.write(b64ToBytes(f.data));
-          }
-        }
       }
     }
     finish(msg) {
@@ -126,6 +152,11 @@
         const body = m.t === 'i'
           ? { t: 'i', d: bytesToB64(new TextEncoder().encode(m.text)), seq: ++this.inSeq }
           : { t: 'r', cols: m.cols, rows: m.rows, seq: ++this.inSeq };
+        // On the WebSocket a message goes out in order with no answer to wait
+        // for; numbered as well, so a repeat by the long polls is dropped.
+        if (this.ws && this.ws.readyState === 1) {
+          try { this.ws.send(JSON.stringify(body)); continue; } catch (e) { /* by request below */ }
+        }
         for (let wait = 300; !this.closed; wait = Math.min(wait * 2, 5000)) {
           let r = null;
           try {
@@ -145,6 +176,7 @@
     close() {
       if (this.closed) return;
       this.closed = true;
+      if (this.ws) { try { this.ws.close(); } catch (e) { /* closed already */ } }
       if (this.sid) fetch(base + this.sid + '/close', { method: 'POST', keepalive: true }).catch(() => {});
       this.term.dispose();
       this.el.remove();