@mittwald/kubernetes/resource_listwatch.js

"use strict";
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
    return new (P || (P = Promise))(function (resolve, reject) {
        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
        step((generator = generator.apply(thisArg, _arguments || [])).next());
    });
};
var __classPrivateFieldGet = (this && this.__classPrivateFieldGet) || function (receiver, state, kind, f) {
    if (kind === "a" && !f) throw new TypeError("Private accessor was defined without a getter");
    if (typeof state === "function" ? receiver !== state || !f : !state.has(receiver)) throw new TypeError("Cannot read private member from an object whose class did not declare it");
    return kind === "m" ? f : kind === "a" ? f.call(receiver) : f ? f.value : state.get(receiver);
};
var __classPrivateFieldSet = (this && this.__classPrivateFieldSet) || function (receiver, state, value, kind, f) {
    if (kind === "m") throw new TypeError("Private method is not writable");
    if (kind === "a" && !f) throw new TypeError("Private accessor was defined without a setter");
    if (typeof state === "function" ? receiver !== state || !f : !state.has(receiver)) throw new TypeError("Cannot write private member to an object whose class did not declare it");
    return (kind === "a" ? f.call(receiver, value) : f ? f.value = value : state.set(receiver, value)), value;
};
var _ListWatchState_resourceVersion;
Object.defineProperty(exports, "__esModule", { value: true });
exports.ListWatch = void 0;
const resource_listwatch_error_1 = require("./resource_listwatch_error");
const debug = require("debug")("kubernetes:resource:listwatch");
const sleep = (ms) => new Promise(res => setTimeout(res, ms));
const doNothing = () => {
};
class ListWatchState {
    constructor() {
        _ListWatchState_resourceVersion.set(this, 0);
        this.running = false;
        this.errorCount = 0;
        this.successCount = 0;
    }
    set resourceVersion(value) {
        if (typeof value === "string") {
            value = parseInt(value, 10);
        }
        if (value > __classPrivateFieldGet(this, _ListWatchState_resourceVersion, "f")) {
            __classPrivateFieldSet(this, _ListWatchState_resourceVersion, value, "f");
        }
    }
    get resourceVersion() {
        return __classPrivateFieldGet(this, _ListWatchState_resourceVersion, "f");
    }
    start() {
        this.running = true;
        this.successCount = 0;
        this.errorCount = 0;
    }
    markSuccess() {
        this.successCount++;
        this.errorCount = 0;
    }
}
_ListWatchState_resourceVersion = new WeakMap();
class ListWatch {
    constructor(onWatchEvent, onWatchError, client, baseURL, resourceBaseURL, opts, metrics) {
        var _a;
        this.state = new ListWatchState();
        this.onWatchEvent = onWatchEvent;
        this.onWatchError = onWatchError !== null && onWatchError !== void 0 ? onWatchError : doNothing;
        this.errorStrategy = (_a = opts.errorStrategy) !== null && _a !== void 0 ? _a : resource_listwatch_error_1.DefaultListWatchErrorStrategy;
        this.client = client;
        this.baseURL = baseURL;
        this.resourceBaseURL = resourceBaseURL;
        this.opts = opts;
        this.metrics = metrics;
    }
    handler(event) {
        return __awaiter(this, void 0, void 0, function* () {
            if (event.object.metadata.resourceVersion) {
                this.state.resourceVersion = event.object.metadata.resourceVersion;
            }
            yield this.onWatchEvent(event);
        });
    }
    resync() {
        return __awaiter(this, void 0, void 0, function* () {
            const list = yield this.client.get(this.baseURL, this.opts);
            if (list.metadata.resourceVersion) {
                this.state.resourceVersion = list.metadata.resourceVersion;
            }
            if (this.opts.onResync) {
                yield this.opts.onResync(list.items || []);
            }
            if (!this.opts.skipAddEventsOnResync) {
                for (const i of list.items || []) {
                    const event = { type: "ADDED", object: i };
                    yield this.onWatchEvent(event);
                }
            }
        });
    }
    run() {
        this.state.start();
        this.metrics.watchOpenCount.inc({ baseURL: this.baseURL });
        debug("starting list-watch on %o", this.resourceBaseURL);
        const initialized = this.resync();
        const { resyncAfterIterations = 10 } = this.opts;
        const done = initialized.then(() => __awaiter(this, void 0, void 0, function* () {
            this.state.markSuccess();
            const onEstablished = () => this.state.markSuccess();
            debug("initial list for list-watch on %o completed", this.resourceBaseURL);
            while (this.state.running) {
                try {
                    if (this.state.successCount % resyncAfterIterations === 0) {
                        debug(`resyncing after ${resyncAfterIterations} successful WATCH iterations`);
                        yield this.resync();
                    }
                    debug("resuming watch after %o successful iterations and %o errors", this.state.successCount, this.state.errorCount);
                    const watchOpts = Object.assign(Object.assign({}, this.opts), { resourceVersion: this.state.resourceVersion, onEstablished });
                    const result = yield this.client.watch(this.baseURL, e => this.handler(e), e => this.onWatchError(e), watchOpts);
                    if (result.resyncRequired) {
                        debug(`resyncing listwatch`);
                        yield this.resync();
                        continue;
                    }
                    this.state.resourceVersion = result.resourceVersion;
                }
                catch (err) {
                    yield this.handleWatchIterationError(err);
                }
            }
            this.metrics.watchOpenCount.dec({ baseURL: this.baseURL });
        }));
        return {
            initialized,
            done,
            stop() {
                this.running = false;
            },
        };
    }
    handleWatchIterationError(err) {
        return __awaiter(this, void 0, void 0, function* () {
            this.state.errorCount++;
            const reaction = this.errorStrategy(err, this.state.errorCount);
            this.metrics.watchResyncErrorCount.inc({ baseURL: this.baseURL });
            debug("encountered error while watching: %o; determined reaction: %o", err, reaction);
            if (this.opts.onError) {
                yield this.opts.onError(err);
            }
            if (this.opts.abortAfterErrorCount && this.state.errorCount > this.opts.abortAfterErrorCount) {
                this.metrics.watchOpenCount.dec({ baseURL: this.baseURL });
                throw new Error(`more than ${this.opts.abortAfterErrorCount} consecutive errors when watching ${this.baseURL}`);
            }
            if (reaction.backoff) {
                debug("resuming watch after back-off of %o ms", reaction.backoff);
                yield sleep(reaction.backoff);
            }
            if (reaction.resync) {
                debug("resuming watch with resync after error: %o", err);
                yield this.resync();
            }
        });
    }
}
exports.ListWatch = ListWatch;
//# sourceMappingURL=resource_listwatch.js.map