<commit_before>class WebsocketClientService {

    constructor($q, $rootScope, websocketActionCreator) {
        'ngInject';

        this._$q = $q;
        this._$rootScope = $rootScope;
        this._websocketActionCreator = websocketActionCreator;

        this._connectionAttempts = 1;
        this._eventsSubscribed = {};
        this._currentOnGoingMessages = {};
    }

    connect() {
        const defer = this._$q.defer();

        if (_.isUndefined(this._websocket) || this._websocket.readyState === WebSocket.CLOSED) {
            this._websocket = new WebSocket(`ws://${location.hostname}/api/v1/ws`);

            this._websocket.onopen = () => {
                const watcherPromises = [];

                this._connectionAttempts = 1;
                this._reconnect = true;

                _.each(this._eventsSubscribed, (watcher) => {
                    watcherPromises.push(this.sendMessage(watcher)
                        .then((response) => this._websocketActionCreator.subscribedResource(response)));
                });
                this._$q.all(watcherPromises)
                    .then(() => defer.resolve());
            };

            this._websocket.onmessage = (evt) => {
                const message = JSON.parse(evt.data);

                this._$rootScope.$apply(() => {
                    if (message.correlation) {
                        if (message.status_code >= 400) {
                            this._currentOnGoingMessages[message.correlation].reject(message);
                        } else {
                            this._currentOnGoingMessages[message.correlation].resolve(message);
                        }

                        delete this._currentOnGoingMessages[message.correlation];
                    } else {
                        this._websocketActionCreator.updateResource(message);
                    }
                });
            };

            this._websocket.onerror = () => {
                defer.reject();
            };

            this._websocket.onclose = () => {
                if (this._reconnect) {
                    const time = generateInterval(this._connectionAttempts);

                    setTimeout(() => {
                        this._connectionAttempts++;
                        this.connect();
                    }, time);
                }
            };
        } else {
            defer.resolve();
        }

        return defer.promise;
    }

    disconnect() {
        const promises = [];

        _.each(this._eventsSubscribed, (value) => promises.push(this.unSubscribeEvent(value)));

        return this._$q.all(promises)
            .then(() => {
                this._reconnect = false;
                this._websocket.close();
            });
    }

    sendMessage(message) {
        if (this._websocket.readyState !== WebSocket.OPEN) {
            return this._$q.reject('Not Connected');
        }

        const defer = this._$q.defer();
        const correlationId = _.uniqueId();

        this._currentOnGoingMessages[correlationId] = defer;

        message.correlation = correlationId;
        this._websocket.send(JSON.stringify(message));

        return defer.promise;<commit_msg>Remove watchers on disconnect<commit_after>class WebsocketClientService {

    constructor($q, $rootScope, websocketActionCreator) {
        'ngInject';

        this._$q = $q;
        this._$rootScope = $rootScope;
        this._websocketActionCreator = websocketActionCreator;

        this._connectionAttempts = 1;
        this._eventsSubscribed = {};
        this._currentOnGoingMessages = {};
    }

    connect() {
        const defer = this._$q.defer();

        if (_.isUndefined(this._websocket) || this._websocket.readyState === WebSocket.CLOSED) {
            this._websocket = new WebSocket(`ws://${location.hostname}/api/v1/ws`);

            this._websocket.onopen = () => {
                const watcherPromises = [];

                this._connectionAttempts = 1;
                this._reconnect = true;

                _.each(this._eventsSubscribed, (watcher) => {
                    watcherPromises.push(this.sendMessage(watcher)
                        .then((response) => this._websocketActionCreator.subscribedResource(response)));
                });
                this._$q.all(watcherPromises)
                    .then(() => defer.resolve());
            };

            this._websocket.onmessage = (evt) => {
                const message = JSON.parse(evt.data);

                this._$rootScope.$apply(() => {
                    if (message.correlation) {
                        if (message.status_code >= 400) {
                            this._currentOnGoingMessages[message.correlation].reject(message);
                        } else {
                            this._currentOnGoingMessages[message.correlation].resolve(message);
                        }

                        delete this._currentOnGoingMessages[message.correlation];
                    } else {
                        this._websocketActionCreator.updateResource(message);
                    }
                });
            };

            this._websocket.onerror = () => {
                defer.reject();
            };

            this._websocket.onclose = () => {
                if (this._reconnect) {
                    const time = generateInterval(this._connectionAttempts);

                    setTimeout(() => {
                        this._connectionAttempts++;
                        this.connect();
                    }, time);
                }
            };
        } else {
            defer.resolve();
        }

        return defer.promise;
    }

    disconnect() {
        const promises = [];

        _.each(this._eventsSubscribed, (value, key) => promises.push(this.unSubscribeEvent(key)));

        return this._$q.all(promises)
            .then(() => {
                this._reconnect = false;
                this._websocket.close();
            });
    }

    sendMessage(message) {
        if (this._websocket.readyState !== WebSocket.OPEN) {
            return this._$q.reject('Not Connected');
        }

        const defer = this._$q.defer();
        const correlationId = _.uniqueId();

        this._currentOnGoingMessages[correlationId] = defer;

        message.correlation = correlationId;
        this._websocket.send(JSON.stringify(message));

        return defer.promise;
{'fraction_non_alphanumeric': 0.11792964512629785, 'fraction_numerical': 0.0018596001859600185, 'mean_word_length': 1.4718498659517427, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6882299', 'n_tokens_mistral': 1730, 'n_tokens_neox': 1551, 'n_words': 219}