<commit_before>'use strict';

var r = require('rethinkdb');
var express = require('express');
var socketio = require('socket.io');

var config = require('./config');
var db = require('./ffmap/db');
var logger = require('./util/logger');

var app = express();
app.use(express.static(config.frontend.path));
var io = socketio.listen(app.listen(config.port), { log: false });

db.init(config).then(function () {
    config.dataSources.forEach(function (src) {
        var aggregate = require('./ffmap/aggregator').for(src.type).aggregate;
        var penality = 1;
        function setup() {
            aggregate(src, config.database).then(function () {
                logger.info('Received data and stored to DB');
                penality = 1;
            }, function () {
                penality = penality * 2;
                logger.warn('Problem fetching data or storing in DB');
            });
            setTimeout(function () {
                setup();
            }, src.interval * penality);
        }

        setup();
    });
});

(new (function importData() {
    var self = this;

    return function () {
        return r.connect(config.database).then(function (c) {
            self.connection = c;
            c.use(config.database.db);

            return r.table('currentNetwork')
                .get('nodes.json')
                .changes()
                .run(c)
                .then(function (res) {
                    return res.each(function (err, n) {
                        if (err) {
                            logger.warn(err);
                            throw err;
                        }
                        var nodes = n.nodes;
                        for (var id in nodes) {
                            var node = nodes[id];
                            node.id = id;
                            r.table('nodes').insert(node, {
                                conflict: 'update'
                            }).run(self.connection);
                        }
                    });
                });
        });
    };
})())();

io.sockets.on('connection', function (socket) {
    var self = this;
    function refresh (key) {
        key = key || 'nodes';
        return r.connect(config.database).then(function (c) {
            self.connection = c;
            return r.table('nodes')
                .run(c);
        }).then(function (data) {
            socket.emit('nodes:reset');
            data.each(function (err, n) {
                if (err) {
                    logger.warn(err);
                }
                socket.emit('nodes:add', n);
            });
        });
    }

    socket.on('refresh', refresh);
    refresh().then(function () {
        r.table('nodes')
            .changes()
            .run(self.connection).then(function (data) {
                data.each(function (err, node) {
                    if (err) {
                        logger.warn(err);
                        //just ignore error
                        return;
                    }
                    socket.emit('nodes:update', node);
                });
            });
    });
});<commit_msg>add handling of links for socketio API<commit_after>'use strict';

var r = require('rethinkdb');
var express = require('express');
var socketio = require('socket.io');

var config = require('./config');
var db = require('./ffmap/db');
var logger = require('./util/logger');

var app = express();
app.use(express.static(config.frontend.path));
var io = socketio.listen(app.listen(config.port), { log: false });

db.init(config).then(function () {
    config.dataSources.forEach(function (src) {
        var aggregate = require('./ffmap/aggregator').for(src.type).aggregate;
        var penality = 1;
        function setup() {
            aggregate(src, config.database).then(function () {
                logger.info('Received data and stored to DB');
                penality = 1;
            }, function () {
                penality = penality * 2;
                logger.warn('Problem fetching data or storing in DB');
            });
            setTimeout(function () {
                setup();
            }, src.interval * penality);
        }

        setup();
    });
});

(new (function importData() {
    var self = this;

    return function () {
        return r.connect(config.database).then(function (c) {
            self.connection = c;
            c.use(config.database.db);

            return r.table('currentNetwork')
                .get('nodes.json')
                .changes()
                .run(c)
                .then(function (res) {
                    return res.each(function (err, n) {
                        if (err) {
                            logger.warn(err);
                            throw err;
                        }
                        var nodes = n.nodes;
                        for (var id in nodes) {
                            var node = nodes[id];
                            node.id = id;
                            r.table('nodes').insert(node, {
                                conflict: 'update'
                            }).run(self.connection);
                        }
                        if (n.links) {
                            for (id in n.links) {
                                var link = n.links[id];
                                r.table('links').insert(link, {
                                    conflict: 'update'
                                }).run(self.connection);
                            }
                        }
                    });
                });
        });
    };
})())();

io.sockets.on('connection', function (socket) {
    var self = this;
    function refresh (key) {
        key = key || 'nodes';
        return r.connect(config.database).then(function (c) {
            self.connection = c;
            r.table('nodes')
                .run(c).then(function (data) {
                    socket.emit('nodes:reset');
                    data.each(function (err, n) {
                        if (err) {
                            logger.warn(err);
                            return;
                        }
                        socket.emit('nodes:add', n);
                    });
                });
            r.table('links')
                .run(c).then(function (data) {
                    socket.emit('links:reset');
                    data.each(function (err, l) {
                        if (err) {
                            logger.warn(err);
                            return;
                        }
                        socket.emit('links.add', l);
                    });
                });
        });
    }

    socket.on('refresh', refresh);
    refresh().then(function () {
        r.table('nodes')
            .changes()
            .run(self.connection).then(function (data) {
                data.each(function (err, node) {
                    if (err) {
                        logger.warn(err);
                        //just ignore error
                        return;
                    }
                    socket.emit('nodes:update', node);
                });
            });
        r.table('links')
            .changes()
            .run(self.connection).then(function (data) {
                data.each(function (err, link) {
                    if (err) {
                        logger.warn(err);
                        //just ignore error
                        return;
                    }
                    socket.emit('links:update', link);
                });
            });
    });
});
{'fraction_non_alphanumeric': 0.1317412413069151, 'fraction_numerical': 0.0007872982548222018, 'mean_word_length': 1.0533405172413792, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '266502', 'n_tokens_mistral': 1923, 'n_tokens_neox': 1820, 'n_words': 373}