nathan/xTeVe
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Phase 5a: persistent websocket with request ids and reconnect
continuous-integration/drone/push Build encountered an error
Details

Browse Source 
Client (ts/network_ts.ts): one WebSocket per page. Commands queue and go
out one at a time with a client-chosen id; the response is matched on the
echoed id (or to the in-flight request for older servers). 12 s timeout
per request, exponential-backoff reconnect (0.5 s to 10 s), the in-flight
request is retried after a reconnect, and log polls are de-duplicated so
they cannot pile up behind a stalled connection. The old global flag that
silently dropped any request made while another was in flight is gone.

Server (src/webserver.go): the /data/ handler now serves any number of
commands on one connection (it used to break out of its loop after the
first reply without closing the socket, leaving it open and deaf; the old
client papered over that by opening a new socket per request). Connection
closed on exit, request id echoed in the response.

Test: TestWSServesMultipleCommandsPerConnection.
This commit is contained in:
nathan
2026-09-26 13:19:38 +10:00
parent cbc5bc5da2
commit 3a3ba861dc
8 changed files with 475 additions and 242 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ts/network_ts.ts
+228 -125
View File
@@ -1,136 +1,240 @@
class Server {
protocol:string
cmd:string
// Websocket transport between the web UI and xTeVe.
//
// One socket stays open for the life of the page. Requests are queued and
// sent one at a time; each carries an id that the server echoes back, and
// the response is matched on it (or, for older servers that close after a
// single reply, to the oldest in-flight request). Connection loss triggers
// a reconnect with exponential backoff, and queued requests survive it.
constructor(cmd:string) {
this.cmd = cmd
interface PendingRequest {
id: string
payload: Object
resolve: (response: any) => void
reject: (err: Error) => void
timer: number
sent: boolean
}
class WSClient {
private socket: WebSocket = null
private queue: PendingRequest[] = []
private inFlight: PendingRequest = null
private nextID: number = 1
private reconnectDelay: number = 500
private reconnectTimer: number = 0
private closedByUs: boolean = false
readonly requestTimeoutMs: number = 12000
private url(): string {
var protocol = window.location.protocol == "https:" ? "wss://" : "ws://"
var host = window.location.host
if (host == undefined || host.length < 1) {
host = window.location.hostname
}
return protocol + host + "/data/"
}
request(data:Object):any {
// send queues a command and resolves with the parsed response.
send(payload: Object): Promise<any> {
var client = this
return new Promise(function(resolve, reject) {
var req: PendingRequest = {
id: String(client.nextID++),
payload: payload,
resolve: resolve,
reject: reject,
timer: 0,
sent: false
}
client.queue.push(req)
client.pump()
})
}
if (SERVER_CONNECTION == true) {
// hasQueued reports whether a command of this kind is already waiting.
hasQueued(cmd: string): boolean {
if (this.inFlight != null && this.inFlight.payload["cmd"] == cmd) {
return true
}
for (var i = 0; i < this.queue.length; i++) {
if (this.queue[i].payload["cmd"] == cmd) {
return true
}
}
return false
}
private connect(): void {
if (this.socket != null && (this.socket.readyState == WebSocket.OPEN || this.socket.readyState == WebSocket.CONNECTING)) {
return
}
SERVER_CONNECTION = true
if (this.cmd != "updateLog") {
var client = this
var socket = new WebSocket(this.url())
this.socket = socket
socket.onopen = function() {
WS_AVAILABLE = true
client.reconnectDelay = 500
client.pump()
}
socket.onmessage = function(e: MessageEvent) {
client.handleMessage(e.data)
}
socket.onerror = function() {
// onclose follows; nothing to do here.
}
socket.onclose = function() {
if (client.socket == socket) {
client.socket = null
}
// An in-flight request that never got its answer goes back to the
// front of the queue and is retried on the next connection.
if (client.inFlight != null) {
var req = client.inFlight
client.inFlight = null
req.sent = false
client.queue.unshift(req)
}
client.scheduleReconnect()
}
}
private scheduleReconnect(): void {
if (this.closedByUs || this.reconnectTimer != 0) {
return
}
if (this.queue.length == 0) {
// Idle: reconnect lazily on the next request.
return
}
var client = this
this.reconnectTimer = window.setTimeout(function() {
client.reconnectTimer = 0
client.connect()
}, this.reconnectDelay)
this.reconnectDelay = Math.min(this.reconnectDelay * 2, 10000)
}
// pump sends the next queued request if the socket is open and idle.
private pump(): void {
if (this.inFlight != null || this.queue.length == 0) {
return
}
if (this.socket == null || this.socket.readyState != WebSocket.OPEN) {
this.connect()
return
}
var req = this.queue.shift()
this.inFlight = req
req.sent = true
req.payload["id"] = req.id
var client = this
req.timer = window.setTimeout(function() {
if (client.inFlight == req) {
client.inFlight = null
req.reject(new Error("timeout"))
// A socket that stopped answering is not trusted any more.
client.resetSocket()
client.pump()
}
}, this.requestTimeoutMs)
try {
this.socket.send(JSON.stringify(req.payload))
} catch (err) {
window.clearTimeout(req.timer)
this.inFlight = null
req.sent = false
this.queue.unshift(req)
this.resetSocket()
}
}
private resetSocket(): void {
var socket = this.socket
this.socket = null
if (socket != null) {
socket.onclose = null
try { socket.close() } catch (err) { /* ignore */ }
}
this.scheduleReconnect()
}
private handleMessage(raw: string): void {
var response: any
try {
response = JSON.parse(raw)
} catch (err) {
return
}
var req = this.inFlight
if (req == null) {
return
}
// Servers that echo the id let us drop stray late answers.
if (response.hasOwnProperty("id") && response["id"] != req.id) {
return
}
window.clearTimeout(req.timer)
this.inFlight = null
req.resolve(response)
this.pump()
}
}
var WS = new WSClient()
class Server {
cmd: string
constructor(cmd: string) {
this.cmd = cmd
}
request(data: Object): any {
var isLogUpdate: boolean = this.cmd == "updateLog"
// Log polling must never pile up behind a stalled connection.
if (isLogUpdate && WS.hasQueued("updateLog")) {
return
}
if (!isLogUpdate) {
showElement("loading", true)
UNDO = new Object()
setConnectionState("busy")
}
switch(window.location.protocol) {
case "http:":
this.protocol = "ws://"
break
case "https:":
this.protocol = "wss://"
break
}
var wsHost:string = window.location.host
if (wsHost == undefined || wsHost.length < 1) {
wsHost = window.location.hostname
}
// The session cookie (HttpOnly) is sent with the websocket handshake.
var url = this.protocol + wsHost + "/data/"
data["cmd"] = this.cmd
var requestCmd:string = data["cmd"]
var ws = new WebSocket(url)
var isLogUpdate:boolean = data["cmd"] == "updateLog"
var responseReceived:boolean = false
var requestFinished:boolean = false
var timeoutMs:number = 12000
var requestTimeout:number
var requestCmd: string = this.cmd
var finishRequest = function(state:string, responseSuccess:boolean = false):void {
if (requestFinished == true) {
return
}
WS.send(data).then(function(response) {
requestFinished = true
SERVER_CONNECTION = false
window.clearTimeout(requestTimeout)
if (responseSuccess == true) {
if (state == "online") {
WS_FAILURE_COUNT = 0
}
} else {
WS_FAILURE_COUNT++
}
if (isLogUpdate == false) {
WS_FAILURE_COUNT = 0
if (!isLogUpdate) {
showElement("loading", false)
}
if (state != "") {
setConnectionState(state)
}
}
requestTimeout = window.setTimeout(function() {
console.log("Websocket request timed out.")
var timeoutState:string = "offline"
if (isLogUpdate == true && WS_FAILURE_COUNT < 2) {
timeoutState = "idle"
}
finishRequest(timeoutState, false)
try {
ws.close()
} catch (err) {
console.log(err)
}
}, timeoutMs)
ws.onopen = function() {
WS_AVAILABLE = true
if (data["cmd"] != "updateLog") {
setConnectionState("busy")
}
this.send(JSON.stringify(data));
}
ws.onerror = function(e) {
console.log("No websocket connection to xTeVe could be established. Check your network configuration.")
var errorState:string = "offline"
if (isLogUpdate == true && WS_FAILURE_COUNT < 2) {
errorState = "idle"
}
finishRequest(errorState, false)
if (WS_AVAILABLE == false && isLogUpdate == false && requestCmd != "getServerConfig") {
alert("No websocket connection to xTeVe could be established. Check your network configuration.")
}
}
ws.onmessage = function (e) {
responseReceived = true
finishRequest("online", true)
var response = JSON.parse(e.data);
setConnectionState("online")
if (response["status"] == false) {
setConnectionState("offline")
alert(response["err"])
if (response.hasOwnProperty("reload")) {
location.reload()
}
return
}
if (response.hasOwnProperty("logoURL")) {
var div = (document.getElementById("channel-icon") as HTMLInputElement)
div.value = response["logoURL"]
@@ -138,19 +242,17 @@ class Server {
return
}
switch (data["cmd"]) {
switch (requestCmd) {
case "updateLog":
SERVER["log"] = response["log"]
if (document.getElementById("content_log")) {
showLogs(false)
}
return
break;
default:
SERVER = new Object()
SERVER = response
break;
break
}
if (response.hasOwnProperty("openMenu")) {
@@ -171,7 +273,6 @@ class Server {
location.reload()
}
if (response.hasOwnProperty("wizard")) {
createLayout()
configurationWizard[response["wizard"]].createWizard()
@@ -180,23 +281,25 @@ class Server {
createLayout()
}
}, function(err) {
ws.onclose = function() {
if (responseReceived == true) {
return
WS_FAILURE_COUNT++
if (!isLogUpdate) {
showElement("loading", false)
}
var closeState:string = "offline"
if (isLogUpdate == true && WS_FAILURE_COUNT < 2) {
closeState = "idle"
var state: string = "offline"
if (isLogUpdate && WS_FAILURE_COUNT < 2) {
state = "idle"
}
finishRequest(closeState, false)
}
setConnectionState(state)
if (WS_AVAILABLE == false && !isLogUpdate && requestCmd != "getServerConfig") {
alert("No websocket connection to xTeVe could be established. Check your network configuration.")
}
})
}
}
var WS_FAILURE_COUNT:number = 0
var WS_FAILURE_COUNT: number = 0