home-assistant-js-websocket/dist/connection.js

/**
 * Connection that wraps a socket and provides an interface to interact with
 * the Home Assistant websocket API.
 */
import * as messages from "./messages.js";
import { ERR_INVALID_AUTH, ERR_CONNECTION_LOST } from "./errors.js";
const DEBUG = false;
export class Connection {
    constructor(socket, options) {
        // connection options
        //  - setupRetry: amount of ms to retry when unable to connect on initial setup
        //  - createSocket: create a new Socket connection
        this.options = options;
        // id if next command to send
        this.commandId = 1;
        // info about active subscriptions and commands in flight
        this.commands = new Map();
        // map of event listeners
        this.eventListeners = new Map();
        // true if a close is requested by the user
        this.closeRequested = false;
        this.setSocket(socket);
    }
    get haVersion() {
        return this.socket.haVersion;
    }
    setSocket(socket) {
        const oldSocket = this.socket;
        this.socket = socket;
        socket.addEventListener("message", ev => this._handleMessage(ev));
        socket.addEventListener("close", ev => this._handleClose(ev));
        if (oldSocket) {
            const oldCommands = this.commands;
            // reset to original state
            this.commandId = 1;
            this.commands = new Map();
            oldCommands.forEach(info => {
                if ("subscribe" in info) {
                    info.subscribe().then(unsub => {
                        info.unsubscribe = unsub;
                        // We need to resolve this in case it wasn't resolved yet.
                        // This allows us to subscribe while we're disconnected
                        // and recover properly.
                        info.resolve();
                    });
                }
            });
            this.fireEvent("ready");
        }
    }
    addEventListener(eventType, callback) {
        let listeners = this.eventListeners.get(eventType);
        if (!listeners) {
            listeners = [];
            this.eventListeners.set(eventType, listeners);
        }
        listeners.push(callback);
    }
    removeEventListener(eventType, callback) {
        const listeners = this.eventListeners.get(eventType);
        if (!listeners) {
            return;
        }
        const index = listeners.indexOf(callback);
        if (index !== -1) {
            listeners.splice(index, 1);
        }
    }
    fireEvent(eventType, eventData) {
        (this.eventListeners.get(eventType) || []).forEach(callback => callback(this, eventData));
    }
    close() {
        this.closeRequested = true;
        this.socket.close();
    }
    /**
     * Subscribe to a specific or all events.
     *
     * @param callback Callback  to be called when a new event fires
     * @param eventType
     * @returns promise that resolves to an unsubscribe function
     */
    async subscribeEvents(callback, eventType) {
        return this.subscribeMessage(callback, messages.subscribeEvents(eventType));
    }
    ping() {
        return this.sendMessagePromise(messages.ping());
    }
    sendMessage(message, commandId) {
        if (DEBUG) {
            console.log("Sending", message);
        }
        if (!commandId) {
            commandId = this._genCmdId();
        }
        message.id = commandId;
        this.socket.send(JSON.stringify(message));
    }
    sendMessagePromise(message) {
        return new Promise((resolve, reject) => {
            const commandId = this._genCmdId();
            this.commands.set(commandId, { resolve, reject });
            this.sendMessage(message, commandId);
        });
    }
    /**
     * Call a websocket command that starts a subscription on the backend.
     *
     * @param message the message to start the subscription
     * @param callback the callback to be called when a new item arrives
     * @returns promise that resolves to an unsubscribe function
     */
    async subscribeMessage(callback, subscribeMessage) {
        // Command ID that will be used
        const commandId = this._genCmdId();
        let info;
        await new Promise((resolve, reject) => {
            // We store unsubscribe on info object. That way we can overwrite it in case
            // we get disconnected and we have to subscribe again.
            info = {
                resolve,
                reject,
                callback,
                subscribe: () => this.subscribeMessage(callback, subscribeMessage),
                unsubscribe: async () => {
                    await this.sendMessagePromise(messages.unsubscribeEvents(commandId));
                    this.commands.delete(commandId);
                }
            };
            this.commands.set(commandId, info);
            try {
                this.sendMessage(subscribeMessage, commandId);
            }
            catch (err) {
                // Happens when the websocket is already closing.
                // Don't have to handle the error, reconnect logic will pick it up.
            }
        });
        return () => info.unsubscribe();
    }
    _handleMessage(event) {
        const message = JSON.parse(event.data);
        if (DEBUG) {
            console.log("Received", message);
        }
        const info = this.commands.get(message.id);
        switch (message.type) {
            case "event":
                if (info) {
                    info.callback(message.event);
                }
                else {
                    console.warn(`Received event for unknown subscription ${message.id}. Unsubscribing.`);
                    this.sendMessagePromise(messages.unsubscribeEvents(message.id));
                }
                break;
            case "result":
                // No info is fine. If just sendMessage is used, we did not store promise for result
                if (info) {
                    if (message.success) {
                        info.resolve(message.result);
                        // Don't remove subscriptions.
                        if (!("subscribe" in info)) {
                            this.commands.delete(message.id);
                        }
                    }
                    else {
                        info.reject(message.error);
                        this.commands.delete(message.id);
                    }
                }
                break;
            case "pong":
                if (info) {
                    info.resolve();
                    this.commands.delete(message.id);
                }
                else {
                    console.warn(`Received unknown pong response ${message.id}`);
                }
                break;
            default:
                if (DEBUG) {
                    console.warn("Unhandled message", message);
                }
        }
    }
    _handleClose(ev) {
        // Reject in-flight sendMessagePromise requests
        this.commands.forEach(info => {
            // We don't cancel subscribeEvents commands in flight
            // as we will be able to recover them.
            if (!("subscribe" in info)) {
                info.reject(messages.error(ERR_CONNECTION_LOST, "Connection lost"));
            }
        });
        if (this.closeRequested) {
            return;
        }
        this.fireEvent("disconnected");
        // Disable setupRetry, we control it here with auto-backoff
        const options = Object.assign(Object.assign({}, this.options), { setupRetry: 0 });
        const reconnect = (tries) => {
            setTimeout(async () => {
                if (DEBUG) {
                    console.log("Trying to reconnect");
                }
                try {
                    const socket = await options.createSocket(options);
                    this.setSocket(socket);
                }
                catch (err) {
                    if (err === ERR_INVALID_AUTH) {
                        this.fireEvent("reconnect-error", err);
                    }
                    else {
                        reconnect(tries + 1);
                    }
                }
            }, Math.min(tries, 5) * 1000);
        };
        reconnect(0);
    }
    _genCmdId() {
        return ++this.commandId;
    }
}