<commit_before>'use strict';

var EventEmitter  = require('events').EventEmitter;
var axon          = require('axon');
var rpc           = require('axon-rpc');
var req           = axon.socket('req');
var subSock       = axon.socket('sub');
var rpcClient     = new rpc.Client(req);
var equal         = require('deep-equal');

module.exports = function(nodecg) {
    if (!nodecg.bundleConfig || !Object.keys(nodecg.bundleConfig).length) {
        throw new Error('[lfg-siphon] No config found in cfg/lfg-siphon.json, aborting!');
    }

    var self     = new EventEmitter();
    var channels = nodecg.bundleConfig.channels;
    var SUB_PORT = nodecg.bundleConfig.subPort || 9455;
    var RPC_PORT = nodecg.bundleConfig.rpcPort || 9456;

    nodecg.listenFor('getChannels', function(data, cb) {
        cb(channels);
    });

    subSock.connect(SUB_PORT, '127.0.0.1');
    req.connect(RPC_PORT, '127.0.0.1');

    subSock.on('connect', function() {
        nodecg.log.info('Connected to Streen');
        channels.forEach(function(channel) {
            rpcClient.call('join', channel, function(err, alreadyJoined) {
                if (err) {
                    nodecg.log.error(err.stack);
                    return;
                }

                if (alreadyJoined) {
                    nodecg.log.info('Streen already in channel:', alreadyJoined);
                } else {
                    nodecg.log.info('Joined channel:', channel);
                    nodecg.sendMessage('join', channel);
                    self.emit('join', channel);
                }
            });
        });
    });

    subSock.on('disconnect', function() {
        nodecg.log.warn('Disconnected from Streen');
        self.emit('disconnect');
        nodecg.sendMessage('disconnect');
    });

    var lastSub;
    subSock.on('message', function(msg) {
        var channel, data;
        switch (msg.toString()) {
            case 'connected':
                nodecg.log.info('Streen connected to Twitch Chat');
                break;
            case 'disconnected':
                nodecg.log.info('Streen disconnected from Twitch Chat');
                break;
            case 'chat':
                data = {
                    channel: arguments[1],
                    user: arguments[2],
                    message: arguments[3],
                    self: arguments[4]
                };
                if (channels.indexOf(data.channel) < 0) return;
                nodecg.sendMessage('chat', data);
                self.emit('chat', data);
                break;
            case 'timeout':
                data = {channel: arguments[1], username: arguments[2]};
                if (channels.indexOf(data.channel) < 0) return;
                nodecg.sendMessage('timeout', data);
                self.emit('timeout', data);
                break;
            case 'clearchat':
                channel = arguments[1];
                if (channels.indexOf(channel) < 0) return;
                nodecg.sendMessage('clearchat', channel);
                self.emit('clearchat', channel);
                break;
            case 'subscription':
                data = arguments[1];
                if (channels.indexOf(data.channel) < 0) return;
                if (equal(lastSub, data)) return;
                lastSub = data;

                // 10-30-2015: Streen now ensures that "months" is an integer.
                // The below line can be removed soon.
                if (data.months) data.months = parseInt(data.months);

                nodecg.sendMessage('subscription', data);
                self.emit('subscription', data);
                break;
        }
    });

    var heartbeatTimeout = setTimeout(heartbeat, 5000);
    var heartbeatResponseTimeout;
    var lastHeartbeatInterval = 5000;
    function heartbeat() {
        // If we don't hear back from Streen in 1000ms, we assume Streen is down and keep trying.
        heartbeatResponseTimeout = setTimeout(handleHeartbeatTimeout, 1000);

        // Emit the heartbeat, and schedule the next one based on Streen's response.
        rpcClient.call('heartbeat', channels, function(err, interval) {
            if (err) {
                nodecg.log.error(err.stack);
            }

            var intervalDuration = interval || lastHeartbeatInterval;
            clearTimeout(heartbeatResponseTimeout);
            clearTimeout(heartbeatTimeout);
            heartbeatTimeout = setTimeout(heartbeat, intervalDuration);
            lastHeartbeatInterval = intervalDuration;
        });
    }

    function handleHeartbeatTimeout() {
        nodecg.log.error('Streen did not respond to heartbeat. Sending another...');
        heartbeatTimeout = setTimeout(heartbeat, lastHeartbeatInterval);
    }

    self.say = function(channel, message) {
        rpcClient.call('say', channel, message, function(err){
            if (err) nodecg.log.error(err.stack);
        });
    };

    self.timeout = function(channel, username, seconds) {
        rpcClient.call('timeout', channel, username, seconds, function(err){
            if (err) nodecg.log.error(err.stack);
        });
    };

    return self;
};<commit_msg>Minor cleanup. Add support for callbacks on all commands. Add "mods" command.<commit_after>'use strict';

var EventEmitter = require('events').EventEmitter;
var axon = require('axon');
var rpc = require('axon-rpc');
var req = axon.socket('req');
var subSock = axon.socket('sub');
var rpcClient = new rpc.Client(req);
var equal = require('deep-equal');

module.exports = function (nodecg) {
    if (!nodecg.bundleConfig || !Object.keys(nodecg.bundleConfig).length) {
        throw new Error('[lfg-siphon] No config found in cfg/lfg-siphon.json, aborting!');
    }

    var self = new EventEmitter();
    var channels = nodecg.bundleConfig.channels;
    var SUB_PORT = nodecg.bundleConfig.subPort || 9455;
    var RPC_PORT = nodecg.bundleConfig.rpcPort || 9456;

    nodecg.listenFor('getChannels', function (data, cb) {
        cb(channels);
    });

    subSock.connect(SUB_PORT, '127.0.0.1');
    req.connect(RPC_PORT, '127.0.0.1');

    subSock.on('connect', function () {
        nodecg.log.info('Connected to Streen');
        channels.forEach(function (channel) {
            rpcClient.call('join', channel, function (err, alreadyJoined) {
                if (err) {
                    nodecg.log.error(err.stack);
                    return;
                }

                if (alreadyJoined) {
                    nodecg.log.info('Streen already in channel:', alreadyJoined);
                } else {
                    nodecg.log.info('Joined channel:', channel);
                    nodecg.sendMessage('join', channel);
                    self.emit('join', channel);
                }
            });
        });
    });

    subSock.on('disconnect', function () {
        nodecg.log.warn('Disconnected from Streen');
        self.emit('disconnect');
        nodecg.sendMessage('disconnect');
    });

    var lastSub;
    subSock.on('message', function (msg) {
        var channel, data;
        switch (msg.toString()) {
            case 'connected':
                nodecg.log.info('Streen connected to Twitch Chat');
                break;
            case 'disconnected':
                nodecg.log.info('Streen disconnected from Twitch Chat');
                break;
            case 'chat':
                data = {
                    channel: arguments[1],
                    user: arguments[2],
                    message: arguments[3],
                    self: arguments[4]
                };
                if (channels.indexOf(data.channel) < 0) return;
                nodecg.sendMessage('chat', data);
                self.emit('chat', data);
                break;
            case 'timeout':
                data = {channel: arguments[1], username: arguments[2]};
                if (channels.indexOf(data.channel) < 0) return;
                nodecg.sendMessage('timeout', data);
                self.emit('timeout', data);
                break;
            case 'clearchat':
                channel = arguments[1];
                if (channels.indexOf(channel) < 0) return;
                nodecg.sendMessage('clearchat', channel);
                self.emit('clearchat', channel);
                break;
            case 'subscription':
                data = arguments[1];
                if (channels.indexOf(data.channel) < 0) return;
                if (equal(lastSub, data)) return;
                lastSub = data;

                // 10-30-2015: Streen now ensures that "months" is an integer.
                // The below line can be removed soon.
                if (data.months) data.months = parseInt(data.months);

                nodecg.sendMessage('subscription', data);
                self.emit('subscription', data);
                break;
        }
    });

    var heartbeatTimeout = setTimeout(heartbeat, 5000);
    var heartbeatResponseTimeout;
    var lastHeartbeatInterval = 5000;

    function heartbeat() {
        // If we don't hear back from Streen in 1000ms, we assume Streen is down and keep trying.
        heartbeatResponseTimeout = setTimeout(handleHeartbeatTimeout, 1000);

        // Emit the heartbeat, and schedule the next one based on Streen's response.
        rpcClient.call('heartbeat', channels, function (err, interval) {
            if (err) {
                nodecg.log.error(err.stack);
            }

            var intervalDuration = interval || lastHeartbeatInterval;
            clearTimeout(heartbeatResponseTimeout);
            clearTimeout(heartbeatTimeout);
            heartbeatTimeout = setTimeout(heartbeat, intervalDuration);
            lastHeartbeatInterval = intervalDuration;
        });
    }

    function handleHeartbeatTimeout() {
        heartbeatTimeout = setTimeout(heartbeat, lastHeartbeatInterval);
    }

    self.say = function (channel, message, callback) {
        rpcClient.call('say', channel, message, function () {
            callback.apply(callback, arguments);
        });
    };

    self.timeout = function (channel, username, seconds, callback) {
        rpcClient.call('timeout', channel, username, seconds, function () {
            callback.apply(callback, arguments);
        });
    };

    self.mods = function (channel, callback) {
        callback = callback || function () {};
        rpcClient.call('mods', channel, function () {
            callback.apply(callback, arguments);
        });
    };

    return self;
};
{'fraction_non_alphanumeric': 0.11967353136566385, 'fraction_numerical': 0.010629211350479263, 'mean_word_length': 1.8542795232936078, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '48081', 'n_tokens_mistral': 2947, 'n_tokens_neox': 2704, 'n_words': 639}