<commit_before>"use strict";

const RuleExecutor = require('./rule_executor');
const RetryExecutor = require('./retry_executor');
const Rule = require('./rule');
const P = require('bluebird');
const stringify = require('json-stable-stringify');

class BasicSubscription {
    constructor(options, kafkaFactory, hyper, ruleName, ruleSpec) {
        this._kafkaFactory = kafkaFactory;
        this._options = options;
        this._log = this._options.log || (() => {});
        ruleSpec.sample =  ruleSpec.sample || options.sample;
        this._rule = new Rule(ruleName, ruleSpec);
        this._subscribed = false;
        this._executor = new RuleExecutor(this._rule, this._kafkaFactory,
            hyper, this._log, this._options);
        this._retryExecutor = new RetryExecutor(this._rule, this._kafkaFactory,
            hyper, this._log, this._options);
    }

    subscribe() {
        if (this._subscribed) {
            throw new Error('Already subscribed!');
        }

        this._log('info/subscription', {
            message: 'Subscribing based on basic topics',
            rule: this._rule.name,
            topics: this._rule.topics
        });

        return P.join(this._executor.subscribe(), this._retryExecutor.subscribe())
        .tap(() => {
            this._subscribed = true;
        });
    }

    unsubscribe() {
        if (this._subscribed) {
            this._executor.close();
            this._retryExecutor.close();
        }
    }
}
// TODO: rewrite this one
class RegexTopicSubscription {
    constructor(options, kafkaFactory, hyper, ruleName, ruleSpec, metadataWatch) {
        this._kafkaFactory = kafkaFactory;
        this._options = options;
        this._log = this._options.log || (() => {});
        this._hyper = hyper;
        this._ruleName = ruleName;
        this._ruleSpec = ruleSpec;
        ruleSpec.sample =  ruleSpec.sample || options.sample;
        this._topicTester = (ruleSpec.topics || (ruleSpec.topic && [ ruleSpec.topic ]))
        .map((topic) => {
            if (/^\/.+\/$/.test(topic)) {
                // Ok, we've got a regex topic! Compile the regex.
                return new RegExp(topic.substring(1, topic.length - 1));
            }
            return topic;
        });
        this._metadataWatch = metadataWatch;
        this._metadataWatch.on('topics_changed', (topics) => {
            const newFilteredTopics = this._filterTopics(topics);
            if (stringify(newFilteredTopics) !== stringify(this._filteredTopics)) {
                this.unsubscribe();
                this._subscribeTopics(topics);
            }
        });
        // Ignore the emitted errors - in 10 seconds it will retry
        this._metadataWatch.on('error', e => this._log('error/metadata_refresh', e));

        this._subscribed = false;
        this._executors = [];
        this._filteredTopics = undefined;
    }

    /**
     * Filters out which topic names are ok to subscribe to.
     * @param {Array} proposedTopicNames a set of available topic names
     *                to check which to subscribe to
     * @return {Array}
     * @private
     */
    _filterTopics(proposedTopicNames) {
        return proposedTopicNames.filter((topic) => {
            return this._topicTester.some((topicTester) => {
                if (topicTester instanceof RegExp) {
                    return topicTester.test(topic);
                }
                return topicTester === topic;
            });
        }).sort();
    }

    _subscribeTopics(topicNames) {
        this._filteredTopics = topicNames;
        const topicRule = Rule.newWithTopicNames(this._ruleName,
            this._ruleSpec, this._filteredTopics);

        this._log('info/subscription', {
            message: 'Subscribing based on regex',
            rule: this._ruleName,
            topics: this._filteredTopics<commit_msg>Log proper topic list we're subscribing to for regex rules<commit_after>"use strict";

const RuleExecutor = require('./rule_executor');
const RetryExecutor = require('./retry_executor');
const Rule = require('./rule');
const P = require('bluebird');
const stringify = require('json-stable-stringify');

class BasicSubscription {
    constructor(options, kafkaFactory, hyper, ruleName, ruleSpec) {
        this._kafkaFactory = kafkaFactory;
        this._options = options;
        this._log = this._options.log || (() => {});
        ruleSpec.sample =  ruleSpec.sample || options.sample;
        this._rule = new Rule(ruleName, ruleSpec);
        this._subscribed = false;
        this._executor = new RuleExecutor(this._rule, this._kafkaFactory,
            hyper, this._log, this._options);
        this._retryExecutor = new RetryExecutor(this._rule, this._kafkaFactory,
            hyper, this._log, this._options);
    }

    subscribe() {
        if (this._subscribed) {
            throw new Error('Already subscribed!');
        }

        this._log('info/subscription', {
            message: 'Subscribing based on basic topics',
            rule: this._rule.name,
            topics: this._rule.topics
        });

        return P.join(this._executor.subscribe(), this._retryExecutor.subscribe())
        .tap(() => {
            this._subscribed = true;
        });
    }

    unsubscribe() {
        if (this._subscribed) {
            this._executor.close();
            this._retryExecutor.close();
        }
    }
}
// TODO: rewrite this one
class RegexTopicSubscription {
    constructor(options, kafkaFactory, hyper, ruleName, ruleSpec, metadataWatch) {
        this._kafkaFactory = kafkaFactory;
        this._options = options;
        this._log = this._options.log || (() => {});
        this._hyper = hyper;
        this._ruleName = ruleName;
        this._ruleSpec = ruleSpec;
        ruleSpec.sample =  ruleSpec.sample || options.sample;
        this._topicTester = (ruleSpec.topics || (ruleSpec.topic && [ ruleSpec.topic ]))
        .map((topic) => {
            if (/^\/.+\/$/.test(topic)) {
                // Ok, we've got a regex topic! Compile the regex.
                return new RegExp(topic.substring(1, topic.length - 1));
            }
            return topic;
        });
        this._metadataWatch = metadataWatch;
        this._metadataWatch.on('topics_changed', (topics) => {
            const newFilteredTopics = this._filterTopics(topics);
            if (stringify(newFilteredTopics) !== stringify(this._filteredTopics)) {
                this.unsubscribe();
                this._subscribeTopics(topics);
            }
        });
        // Ignore the emitted errors - in 10 seconds it will retry
        this._metadataWatch.on('error', e => this._log('error/metadata_refresh', e));

        this._subscribed = false;
        this._executors = [];
        this._filteredTopics = undefined;
    }

    /**
     * Filters out which topic names are ok to subscribe to.
     * @param {Array} proposedTopicNames a set of available topic names
     *                to check which to subscribe to
     * @return {Array}
     * @private
     */
    _filterTopics(proposedTopicNames) {
        return proposedTopicNames.filter((topic) => {
            return this._topicTester.some((topicTester) => {
                if (topicTester instanceof RegExp) {
                    return topicTester.test(topic);
                }
                return topicTester === topic;
            });
        }).sort();
    }

    _subscribeTopics(topicNames) {
        this._filteredTopics = topicNames;
        const topicRule = Rule.newWithTopicNames(this._ruleName,
            this._ruleSpec, this._filteredTopics);

        this._log('info/subscription', {
            message: 'Subscribing based on regex',
            rule: this._ruleName,
            topics: topicRule.topics
{'fraction_non_alphanumeric': 0.12054547793644667, 'fraction_numerical': 0.0010292036536729706, 'mean_word_length': 2.1096, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 19, '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': '883564', 'n_tokens_mistral': 2133, 'n_tokens_neox': 2083, 'n_words': 456}