import type _Vue from 'vue' import { consola } from 'consola' import { camelCase, mergeWith } from 'lodash-es' import type { TypedStore } from '@/store' import type { SocketError } from '@/util/is-socket-error' const LOG_PREFIX = '[WEBSOCKET]' const RETRY_MIN = 1000 const RETRY_MAX = 15_000 const RETRY_JITTER = 0.2 const RETRY_SHOW_AFTER = 5000 const PING_AFTER = 10_000 const DEAD_AFTER = 10_000 const FAST_NOTIFY_KEYS = [ 'motion_report' ] as const const retryDelay = (attempt: number): number => { const base = RETRY_MIN * 2 ** Math.min(attempt, 16) return Math.min(RETRY_MAX, Math.round(base * (1 + RETRY_JITTER * (2 * Math.random() - 1)))) } // A 5xx is never suppressible — a 503 drives klippy recovery in socket/onSocketError. const shouldSuppressError = ( error: SocketError, suppressError: SuppressError | undefined ): boolean => { if (error.code >= 500) { return false } if (typeof suppressError !== 'function') { return suppressError === true } try { return suppressError(error) } catch (predicateError) { consola.error(`${LOG_PREFIX} suppressError predicate threw`, predicateError) return false } } export class WebSocketClient { private connection: WebSocket | null = null private requests = new Map() private requestId = 0 private store: TypedStore private cache: CachedParams | null = null private retryCount = 0 private reconnectTimeout: ReturnType | null = null private silenceTimeout: ReturnType | null = null constructor (options: SocketPluginOptions) { this.store = options.store } close () { this.clearReconnectTimer() this.clearSilenceTimer() if (this.connection) { this.clearCacheAndRequests() this.connection.onopen = null this.connection.onmessage = null this.connection.onerror = null this.connection.onclose = null this.connection.close() } } connect () { this.clearReconnectTimer() this.retryCount = 0 this.openSocket() } retryNow () { if ( this.reconnectTimeout !== null || this.store.state.socket.retry !== null ) { this.connect() } } private openSocket () { this.clearCacheAndRequests() try { const url = this.store.state.config.socketUrl if (!url) { this.store.typedDispatch('socket/onSetStatus', 'disconnected') return } this.store.typedDispatch('socket/onSetStatus', 'connecting') const connection = new WebSocket(url) this.connection = connection this.armDeadTimer() connection.onopen = () => { if (this.connection !== connection) return this.retryCount = 0 this.armPingTimer() this.store.typedDispatch('socket/onSetStatus', 'identifying') } connection.onclose = (event) => { if (this.connection !== connection) return consola.debug(`${LOG_PREFIX} Connection closed:`, event) this.handleClose() } connection.onerror = (event) => { if (this.connection !== connection) return consola.error(`${LOG_PREFIX} Connection error:`, event) } connection.onmessage = async (message) => { if (this.connection !== connection) return this.armPingTimer() // Parse the data packet. const socketResponse = JSON.parse(message.data) as SocketResponse if ('id' in socketResponse) { const request = this.requests.get(socketResponse.id) this.requests.delete(socketResponse.id) // Remove a wait if defined. if (request?.wait?.length) { this.store.typedCommit('wait/setRemoveWait', request.wait) } if ('error' in socketResponse) { // Is it in error? if (request) { Object.defineProperty(socketResponse.error, '__request__', { enumerable: false, value: request }) if (request.onRejected) { request.onRejected(socketResponse.error) } } consola.debug(`${LOG_PREFIX} Response error:`, socketResponse.error) if (!shouldSuppressError(socketResponse.error, request?.suppressError)) { this.store.typedDispatch('socket/onSocketError', socketResponse.error) } return } if (request) { // these are specific answers to a request we've made. // Build the response, including a non-enumerable ref of the original request. const result = typeof socketResponse.result === 'string' ? { result: socketResponse.result } : socketResponse.result Object.defineProperty(result, '__request__', { enumerable: false, value: request }) consola.debug(`${LOG_PREFIX} Response:`, result) if (request.dispatch) { this.store.dispatch(request.dispatch, result) } if (request.onFulfilled) { // Unchecked: trusted to match the method's result request.onFulfilled(socketResponse.result) } } return } // These are socket notifications (i.e., no specific request was made..) if (socketResponse.method !== 'notify_status_update') { // Normally, we let notifications through with no cache... this.store.dispatch(`socket/${camelCase(socketResponse.method)}`, socketResponse.params?.[0]) } else { // ...However, status notifications come through thick and fast, // so we cache these and send them through every second. const [params, eventtime] = socketResponse.params // If any of these properties exist, bypass the cache and send immediately for (const key of FAST_NOTIFY_KEYS) { if (key in params) { this.store.typedDispatch('printer/onFastNotifyStatusUpdate', { key, payload: params[key] }, { root: true }) delete params[key] } } const timestamp = eventtime ? eventtime * 1000 : Date.now() this.cache = !this.cache ? { timestamp, params } : { timestamp: this.cache.timestamp, params: mergeWith(this.cache.params, params, (dest, src) => Array.isArray(dest) ? src : undefined) } // If there's a second or more difference, flush the cache. if (timestamp - this.cache.timestamp >= 1000) { this.store.typedDispatch('socket/notifyStatusUpdate', this.cache.params) this.cache = { timestamp, params: {} } } } } } catch (error: unknown) { consola.error(`${LOG_PREFIX} Failed to open WebSocket:`, error) this.handleClose() } } private handleClose () { // If the socket is already marked disconnected, there is nothing left to do. if (this.store.state.socket.status === 'disconnected') return this.clearSilenceTimer() this.clearCacheAndRequests() this.connection = null this.store.typedDispatch('socket/onSetStatus', 'connecting') const delay = retryDelay(this.retryCount) this.retryCount += 1 if (delay >= RETRY_SHOW_AFTER) { this.store.typedCommit('socket/setRetry', { at: Date.now() + delay }) } this.reconnectTimeout = setTimeout(() => { this.clearReconnectTimer() if ( document.hidden || !navigator.onLine ) { this.store.typedCommit('socket/setRetry', { held: true }) } else { this.openSocket() } }, delay) } private clearReconnectTimer () { if (this.reconnectTimeout !== null) { clearTimeout(this.reconnectTimeout) this.reconnectTimeout = null } if (this.store.state.socket.retry !== null) { this.store.typedCommit('socket/setRetry', null) } } private armPingTimer () { this.clearSilenceTimer() this.silenceTimeout = setTimeout(() => { consola.debug(`${LOG_PREFIX} No message for ${PING_AFTER} ms`) this.emit('server.info', { suppressError: true }) .catch(() => {}) this.armDeadTimer() this.store.typedCommit('socket/setStalled', true) }, PING_AFTER) } private armDeadTimer () { this.clearSilenceTimer() this.silenceTimeout = setTimeout(() => { const connection = this.connection consola.debug(`${LOG_PREFIX} No response for ${DEAD_AFTER} ms, replacing the connection`) this.handleClose() connection?.close() }, DEAD_AFTER) } private clearSilenceTimer () { if (this.silenceTimeout !== null) { clearTimeout(this.silenceTimeout) this.silenceTimeout = null } if (this.store.state.socket.stalled) { this.store.typedCommit('socket/setStalled', false) } } /** * Sends data TO the socket */ emit< M extends Moonraker.Method, R extends Moonraker.MethodResult = Moonraker.MethodResult > (method: M, options: EmitOptions) { return new Promise((resolve, reject) => { try { const { wait, params, dispatch, suppressError } = options // Any non-'disconnected' state is eligible to emit; physical readiness // is enforced by the readyState check below. if (this.store.state.socket.status === 'disconnected') { consola.debug(`${LOG_PREFIX} Socket emit denied, disconnected:`, method, options) throw new Error('Socket is disconnected') } if (this.connection?.readyState === WebSocket.OPEN) { this.requestId = (this.requestId + 1) % 90_000 const id = this.requestId + 10_000 const packet: SocketRequest = { id, method, jsonrpc: '2.0' } if (params) { packet.params = params } const request: Request['params'], R> = { id, dispatch, params, wait, suppressError, onFulfilled: resolve, onRejected: reject } if (wait) { this.store.typedDispatch('wait/addWait', wait) } this.requests.set(id, request) this.connection.send(JSON.stringify(packet)) } else { consola.debug(`${LOG_PREFIX} Not ready, or closed.`, method, options, this.connection?.readyState) throw new Error('Socket is not ready or closed') } } catch (error) { reject(error) } }) } clearCacheAndRequests () { this.cache = null for (const request of this.requests.values()) { if (request.onRejected) { request.onRejected(new Error('Socket disconnected')) } } this.requests.clear() } } export const SocketPlugin = { install (Vue: typeof _Vue, options?: SocketPluginOptions) { if (options?.store == null) { throw new Error('options required') } const socket = new WebSocketClient(options) Vue.prototype.$socket = socket Vue.$socket = socket window.addEventListener('online', () => socket.retryNow()) document.addEventListener('visibilitychange', () => { if (document.visibilityState === 'visible') { socket.retryNow() } }) } } declare module 'vue/types/vue' { interface Vue { $socket: WebSocketClient; } interface VueConstructor { $socket: WebSocketClient; } } interface SocketPluginOptions { store: TypedStore; } export type SuppressError = boolean | ((error: SocketError) => boolean) export interface RequestOptions { dispatch?: string; wait?: string; suppressError?: SuppressError; } // skipLibCheck hides .d.ts errors, so validate the Methods shape here type AssertMethods> = T export type CheckedMethods = AssertMethods type AssertNotifications> = T export type CheckedNotifications = AssertNotifications type RequestParams = object extends Moonraker.MethodParams ? Moonraker.MethodParams | undefined : Moonraker.MethodParams export type EmitOptions = RequestOptions & ( undefined extends RequestParams ? { params?: RequestParams } : { params: RequestParams } ) interface Request

{ id: number; dispatch?: string; params: P; wait?: string; suppressError?: SuppressError; // Method syntax (bivariant) so any Request fits in `requests` onFulfilled (value: R): void; onRejected: (reason?: unknown) => void; } // String results are dispatched as `{ result }` type DispatchedResult = R extends string ? { result: R } : R export type ObjectWithRequest = DispatchedResult> & { __request__: Request, Moonraker.MethodResult> } interface SocketRequest { jsonrpc: '2.0'; id: number; method: M; params?: RequestParams; } interface SocketResponseBase { jsonrpc: string; // always available } interface SocketApiResponse extends SocketResponseBase { id: number; result: string | Record; } interface SocketApiErrorResponse extends SocketResponseBase { id: number; error: SocketError; } type SocketNotificationResponse = { [N in Moonraker.Notification]: SocketResponseBase & { method: N; params: Moonraker.Notifications[N]; } }[Moonraker.Notification] type SocketResponse = SocketApiResponse | SocketApiErrorResponse | SocketNotificationResponse interface CachedParams { timestamp: number; params: Record; }