<commit_before>/* 
 * (C) 2020 TekMonks. All rights reserved.
 * License: See enclosed LICENSE file.
 * 
 * Globally distributed blackboard for event propogation
 * at internet scale.
 */

const topics = {}
const API_BB_PATH = "/__org_monkshu__blackboard";
const rest = require(`${CONSTANTS.LIBDIR}/rest.js`);
const BLACKBOARD_MSG = "__org_monkshu_blackboard_msg";
let conf = require(CONSTANTS.BLACKBOARDCONF);
const CONF_UPDATE_MSG = "__org_monkshu_blackboard_msg_conf";
const writeFileAsync = require("util").promisify(require("fs").writeFile);

function init() {
    global.BLACKBOARD = this;
    subscribe(CONF_UPDATE_MSG, confNew => {
        // update in memory working copy
        if (process.send) (process.send({type: CONF_UPDATE_MSG, conf: confNew}));
        else conf = confNew;

        // serialize to survive restarts
        writeFileAsync(CONSTANTS.BLACKBOARDCONF, JSON.stringify(confNew));    // serialize
    });
    process.on("message", msg => {if (msg.type == CONF_UPDATE_MSG) conf = msg.conf});
    process.on("message", msg => {if (msg.type == BLACKBOARD_MSG) _broadcast(msg.msg)});
}

async function doService(request) {
    if (request.type == BLACKBOARD_MSG) {
        if (process.send) (process.send({type: BLACKBOARD_MSG, msg: request.msg}));
        else _broadcast(request.msg);
        return CONSTANTS.TRUE_RESULT;
    } else return CONSTANTS.FALSE_RESULT;
}

function publish(topic, payload) {
    const poster = conf.secure?rest.postHttps:rest.post;

    for (const replica of conf.replicas) {
        const host = replica.substring(0, replica.lastIndexOf(":")); const port = replica.substring(replica.lastIndexOf(":")+1);
        poster(host, port, API_BB_PATH, {}, {type:BLACKBOARD_MSG, msg:{topic, payload}}, (err,result) => {
            if (err || !result.result) LOG.error(`Blackboard replication failed for replica: ${replica}`);
        });
    }
}

function subscribe(topic, callback) {
    if (!topics[topic]) topics[topic] = [];
    topics[topic].push(callback);
}

function _broadcast(msg) {
    const topic = msg.topic;
    if (topics[topic]) for (const subscriber of topics[topic]) subscriber(msg.payload);
}

module.exports = {init, doService, publish, subscribe}<commit_msg>Allow blackboard publish to be async so publisher can wait if they want.<commit_after>/* 
 * (C) 2020 TekMonks. All rights reserved.
 * License: See enclosed LICENSE file.
 * 
 * Globally distributed blackboard for event propogation
 * at internet scale.
 */

const topics = {}
const API_BB_PATH = "/__org_monkshu__blackboard";
const rest = require(`${CONSTANTS.LIBDIR}/rest.js`);
const BLACKBOARD_MSG = "__org_monkshu_blackboard_msg";
let conf = require(CONSTANTS.BLACKBOARDCONF);
const CONF_UPDATE_MSG = "__org_monkshu_blackboard_msg_conf";
const writeFileAsync = require("util").promisify(require("fs").writeFile);

function init() {
    global.BLACKBOARD = this;
    subscribe(CONF_UPDATE_MSG, confNew => {
        // update in memory working copy
        if (process.send) (process.send({type: CONF_UPDATE_MSG, conf: confNew}));
        else conf = confNew;

        // serialize to survive restarts
        writeFileAsync(CONSTANTS.BLACKBOARDCONF, JSON.stringify(confNew));    // serialize
    });
    process.on("message", msg => {if (msg.type == CONF_UPDATE_MSG) conf = msg.conf});
    process.on("message", msg => {if (msg.type == BLACKBOARD_MSG) _broadcast(msg.msg)});
}

async function doService(request) {
    if (request.type == BLACKBOARD_MSG) {
        if (process.send) (process.send({type: BLACKBOARD_MSG, msg: request.msg}));
        else _broadcast(request.msg);
        return CONSTANTS.TRUE_RESULT;
    } else return CONSTANTS.FALSE_RESULT;
}

async function publish(topic, payload) {
    const poster = conf.secure?rest.postHttps:rest.post;

    for (const replica of conf.replicas) {
        const host = replica.substring(0, replica.lastIndexOf(":")); const port = replica.substring(replica.lastIndexOf(":")+1);
        await poster(host, port, API_BB_PATH, {}, {type:BLACKBOARD_MSG, msg:{topic, payload}}, (err,result) => {
            if (err || !result.result) LOG.error(`Blackboard replication failed for replica: ${replica}`);
        });
    }
}

function subscribe(topic, callback) {
    if (!topics[topic]) topics[topic] = [];
    topics[topic].push(callback);
}

function _broadcast(msg) {
    const topic = msg.topic;
    if (topics[topic]) for (const subscriber of topics[topic]) subscriber(msg.payload);
}

module.exports = {init, doService, publish, subscribe}
{'fraction_non_alphanumeric': 0.138182622153438, 'fraction_numerical': 0.0026531063453460093, 'mean_word_length': 4.060402684563758, 'pattern_counts': {'":': 4, '<': 3, '<?xml version=': 0, '>': 11, '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': '1860451', 'n_tokens_mistral': 1474, 'n_tokens_neox': 1442, 'n_words': 331}