Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion src/api.rs
Original file line number Diff line number Diff line change
Expand Up @@ -556,9 +556,21 @@ pub async fn live_ws(State(app): State<Shared>, headers: HeaderMap, upgrade: Web
}

async fn stream_live(app: Shared, mut socket: WebSocket, session: Option<String>) {
// The panel and both official themes treat 10 s without a frame as a lost
// stream and reconnect, so the interval must stay well below that.
let mut ticker = tokio::time::interval(std::time::Duration::from_secs(2));
loop {
ticker.tick().await;
tokio::select! {
_ = ticker.tick() => {}
// The browser sends nothing but its close frame, and that is seen only
// by reading. Unread, the close goes unanswered: the browser waits out
// its 60 s closing handshake while this loop keeps pushing a full frame
// every tick to a page that has already discarded the socket.
inbound = socket.recv() => match inbound {
Some(Ok(Message::Close(_)) | Err(_)) | None => break,
Some(Ok(_)) => continue,
},
}
// Closed rather than downgraded to the public frame, which would leave the
// panel rendering a list with every admin field missing. The close allows
// a client to re-query /api/me and determine its current state.
Expand Down
84 changes: 65 additions & 19 deletions web-admin/src/lib/api.ts
Original file line number Diff line number Diff line change
Expand Up @@ -478,41 +478,63 @@ export function useNodes() {
let socket: WebSocket | null = null
let poll: ReturnType<typeof setInterval> | null = null
let retry: ReturnType<typeof setTimeout> | null = null
let closed = false

// A refresh replaces this effect, and an answer to the one it replaced may
// still arrive after the newer one; it is dropped rather than shown.
const fetchOnce = () =>
api<{ nodes: Node[]; admin: boolean }>("/nodes")
let silent: ReturnType<typeof setTimeout> | null = null

// Bumped by `pause`, which a refresh replacing this effect also runs. A
// request started before then may fail or land after a newer one, and
// neither result describes the page now; it is dropped rather than shown.
let epoch = 0
const fetchOnce = () => {
const started = epoch
return api<{ nodes: Node[]; admin: boolean }>("/nodes")
.then((d) => {
if (closed) return
if (started !== epoch) return
setNodes(d.nodes)
setAdmin(d.admin)
setError(null)
})
.catch((e: Error) => {
if (closed) return
if (started !== epoch) return
setError(e.message)
// With the public page switched off, a revoked session receives a 401
// here and on the stream, so the frame that would report admin=false
// never arrives and the panel would retain the list it already had.
if (e instanceof ApiError && e.status === 401) setAdmin(false)
})

fetchOnce()
}

const url = `${location.protocol === "https:" ? "wss" : "ws"}://${location.host}/api/ws`
// A hub restart closes every stream. Without reconnecting, a page that
// outlives a deploy would remain on the fallback poll for the rest of its
// life, refreshing at a fifth of the live rate with no indication.
const connect = () => {
let opened: WebSocket
try {
socket = new WebSocket(url)
opened = new WebSocket(url)
} catch {
poll ??= setInterval(fetchOnce, 5000)
return
}
socket.onmessage = (event) => {
socket = opened
// Re-armed by every frame. Five of the hub's two-second pushes without one
// mean the connection died without closing, as when a NAT on the path
// forgets it or the hub's machine drops off the network; the browser sends
// nothing on it and would notice only when TCP keepalive gives up, 450 s
// later in Chrome. The stream is replaced rather
// than closed and awaited: on a dead connection the close event arrives
// only after the 60 s closing handshake times out. The notice stays until
// data arrives, since with no network the fetch started alongside may hang
// rather than fail.
const watch = () => {
if (silent) clearTimeout(silent)
silent = setTimeout(() => {
setError("实时数据中断,正在重新连接")
resume()
}, 10_000)
}
watch()
opened.onmessage = (event) => {
watch()
const frame = JSON.parse(event.data)
setNodes(frame.nodes)
setAdmin(frame.admin)
Expand All @@ -523,20 +545,44 @@ export function useNodes() {
poll = null
}
}
socket.onerror = () => socket?.close()
socket.onclose = () => {
if (closed) return
opened.onerror = () => opened.close()
opened.onclose = () => {
if (silent) clearTimeout(silent)
poll ??= setInterval(fetchOnce, 5000)
retry = setTimeout(connect, 5000)
}
}
connect()

return () => {
closed = true
socket?.close()
// A phone suspends a page it sends to the background and drops its
// connections without telling it. Back in front, the socket may still read
// as open while nothing arrives, or close and wait out the retry, either
// way leaving the figures from before; a request caught in flight fails.
// So a hidden page lets go of the stream and starts nothing, and a visible
// one fetches at once and opens a fresh stream.
const pause = () => {
epoch++
if (socket) {
socket.onclose = null
socket.close()
socket = null
}
if (poll) clearInterval(poll)
if (retry) clearTimeout(retry)
if (silent) clearTimeout(silent)
poll = retry = silent = null
}
const resume = () => {
pause()
fetchOnce()
connect()
}
const visibility = () => (document.hidden ? pause() : resume())
document.addEventListener("visibilitychange", visibility)
resume()

return () => {
document.removeEventListener("visibilitychange", visibility)
pause()
}
}, [reload])

Expand Down
Loading