<commit_before>//
//   Copyright 2014 <NAME> <<EMAIL>>
//
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
//   Unless required by applicable law or agreed to in writing,
//   software distributed under the License is distributed on an "AS
//   IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
//   express or implied.  See the License for the specific language
//   governing permissions and limitations under the License.
//

'use strict';

const util = require('util'),
      _ = require('lodash'),
      redisModule = require('./redis'),
      redis = redisModule.createClient();

exports.send = function*(userId, sessionId, ntfs) {
    yield queueNotifications(userId, sessionId, null, ntfs);
};

exports.broadcast = function*(userId, ntfs, excludeSessionId) {
    yield queueNotifications(userId, null, excludeSessionId, ntfs);
};

exports.receive = function*(userId, sessionId, timeout) {
    let command;
    let commands = [];

    let client = redisModule.createClient({ autoClose: false });
    let result = yield client.brpop(`outbox:${userId}:${sessionId}`, timeout);

    if (result) {
        commands.push(result[1]);
    }

    // Retrieve other ntfs if there are any
    while ((command = yield client.rpop(`outbox:${userId}:${sessionId}`)) !== null) {
        commands.push(command);
    }

    yield client.quit();

    return commands.map(function(value) {
        return JSON.parse(value);
    });
};

exports.requeue = function*(userId, sessionId, ntfs) {
    ntfs = ntfs.map(function(ntf) {
        return JSON.stringify(ntf);
    });

    yield redis.rpush.apply(redis, [ `outbox:${userId}:${sessionId}` ].concat(ntfs.reverse()));
};

exports.communicateNewUserIds = function*(userId, sessionId, msg) {
    // TBD: Refactor to a separate module
    yield handleNewUserIds(userId, sessionId, null, [ msg ]);
};

function *queueNotifications(userId, sessionId, excludeSessionId, ntfs) {
    if (!util.isArray(ntfs)) {
        ntfs = [ ntfs ];
    }

    yield handleNewUserIds(userId, sessionId, excludeSessionId, ntfs);

    ntfs = ntfs.map(function(value) {
        return typeof(value) === 'string' ? value : JSON.stringify(value);
    });

    yield redis.run.apply(null,
        [ 'queueOutbox', userId, sessionId, excludeSessionId ].concat(ntfs));
}

function *handleNewUserIds(userId, sessionId, excludeSessionId, ntfs) {
    let allUserIds = [];

    ntfs.forEach(function(command) {
        allUserIds = allUserIds.concat(scanUserIds(command));
    });

    allUserIds = _.uniq(allUserIds);
    yield redis.run.apply(null,
        [ 'introduceNewUserIds', userId, sessionId, excludeSessionId, false ].concat(allUserIds));
}

function scanUserIds(obj) {
    let res = [];

    if (typeof(obj) === 'string') {
        obj = JSON.parse(obj);
    }

    for (let key in obj) {
        if (typeof obj[key] === 'object') {
            res = res.concat(scanUserIds(obj[key]));
        } else if (key === 'userId') {
            res.push(obj[key]);
        }
    }

    return res;
}<commit_msg>Report userId only if it isn't falsy<commit_after>//
//   Copyright 2014 <NAME> <<EMAIL>>
//
//   Licensed under the Apache License, Version 2.0 (the "License");
//   you may not use this file except in compliance with the License.
//   You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
//   Unless required by applicable law or agreed to in writing,
//   software distributed under the License is distributed on an "AS
//   IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
//   express or implied.  See the License for the specific language
//   governing permissions and limitations under the License.
//

'use strict';

const util = require('util'),
      _ = require('lodash'),
      redisModule = require('./redis'),
      redis = redisModule.createClient();

exports.send = function*(userId, sessionId, ntfs) {
    yield queueNotifications(userId, sessionId, null, ntfs);
};

exports.broadcast = function*(userId, ntfs, excludeSessionId) {
    yield queueNotifications(userId, null, excludeSessionId, ntfs);
};

exports.receive = function*(userId, sessionId, timeout) {
    let command;
    let commands = [];

    let client = redisModule.createClient({ autoClose: false });
    let result = yield client.brpop(`outbox:${userId}:${sessionId}`, timeout);

    if (result) {
        commands.push(result[1]);
    }

    // Retrieve other ntfs if there are any
    while ((command = yield client.rpop(`outbox:${userId}:${sessionId}`)) !== null) {
        commands.push(command);
    }

    yield client.quit();

    return commands.map(function(value) {
        return JSON.parse(value);
    });
};

exports.requeue = function*(userId, sessionId, ntfs) {
    ntfs = ntfs.map(function(ntf) {
        return JSON.stringify(ntf);
    });

    yield redis.rpush.apply(redis, [ `outbox:${userId}:${sessionId}` ].concat(ntfs.reverse()));
};

exports.communicateNewUserIds = function*(userId, sessionId, msg) {
    // TBD: Refactor to a separate module
    yield handleNewUserIds(userId, sessionId, null, [ msg ]);
};

function *queueNotifications(userId, sessionId, excludeSessionId, ntfs) {
    if (!util.isArray(ntfs)) {
        ntfs = [ ntfs ];
    }

    yield handleNewUserIds(userId, sessionId, excludeSessionId, ntfs);

    ntfs = ntfs.map(function(value) {
        return typeof(value) === 'string' ? value : JSON.stringify(value);
    });

    yield redis.run.apply(null,
        [ 'queueOutbox', userId, sessionId, excludeSessionId ].concat(ntfs));
}

function *handleNewUserIds(userId, sessionId, excludeSessionId, ntfs) {
    let allUserIds = [];

    ntfs.forEach(function(command) {
        allUserIds = allUserIds.concat(scanUserIds(command));
    });

    allUserIds = _.uniq(allUserIds);
    yield redis.run.apply(null,
        [ 'introduceNewUserIds', userId, sessionId, excludeSessionId, false ].concat(allUserIds));
}

function scanUserIds(obj) {
    let res = [];

    if (typeof(obj) === 'string') {
        obj = JSON.parse(obj);
    }

    for (let key in obj) {
        let value = obj[key];

        if (typeof value === 'object') {
            res = res.concat(scanUserIds(value));
        } else if (key === 'userId' && value) {
            res.push(value);
        }
    }

    return res;
}
{'fraction_non_alphanumeric': 0.142220172201722, 'fraction_numerical': 0.0027675276752767526, 'mean_word_length': 3.180591259640103, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4502301', 'n_tokens_mistral': 2032, 'n_tokens_neox': 1997, 'n_words': 516}