<commit_before>/*
 * Copyright 2016 Red Hat Inc.
 *
 * 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';

var amqp = require('rhea');
var rtr = require('./router.js');

var known_routers = {};
var connected_routers = {};
var addresses = {};
var subscribers = {};
var clients = {};

function subscribe(context) {
    subscribers[context.connection.container_id] = context.sender;
    //send initial routers:
    for (var id in connected_routers) {
        context.sender.send({subject:'routers', body:connected_routers[id].listeners});
    }
}

function unsubscribe (context) {
    delete subscribers[context.connection.container_id];
}

function add_client(context) {
    var id = amqp.generate_uuid();
    context.sender.set_source({address:id});
    clients[context.connection.container_id] = context.sender;
}

function remove_client(context) {
    delete clients[context.connection.container_id];
}

function wrap_known_routers (routers) {
    var data = {};
    for (var k in routers) {
        data[k] = routers[k].listeners;
    }
    return data;
}

function unwrap_known_routers (data) {
    var result = {};
    for (var k in data) {
        result[k] = rtr.known(k, data[k]);
    }
    return result;
}

function get_product (connection) {
    if (connection.properties) {
	return connection.properties.product;
    } else {
	return undefined;
    }
}

function print_routers(routers) {
    for (var r in routers) {
        var router = routers[r];
        if (router) {
            console.log('   ' + r + ' => ' + router.listeners);
        }
    }
}

function get_all_routers() {
    var all_routers = {};
    for (var g in known_routers) {
        var group = known_routers[g];
        for (var key in group) {
            all_routers[key] = group[key];
        }
    }
    for (var r in connected_routers) {
        all_routers[r] = connected_routers[r];
    }

    return all_routers;
}

function check_connectivity() {
    var all_routers = get_all_routers();

    for (var r in connected_routers) {
        check_router_connectors(connected_routers[r], all_routers);
    }
}

function check_router_connectors (router, all_routers) {
    if (router.is_ready_for_connectivity_check()) {
        console.log('checking connectivity for ' + router.container_id);
        router.check_connectors(all_routers || get_all_routers());
    } else {
        console.log(router.container_id + ' not ready for connectivity check: ' + router.initial_provisioning_completed + ' ' + (router.connectors !== undefined));
    }
}

function connected_routers_updated (router) {
    check_connectivity();
    for (var id in subscribers) {
        var sender = subscribers[id];
        sender.send({subject:router.container_id, kind:'router', status:'defined', body:router.listeners});
    }
}

function router_disconnected (context) {
    delete connected_routers[context.connection.container_id];
    console.log('router ' + context.connection.container_id + ' disconnected');
    //update now, or wait a bit?
    //connected_routers_updated();
}

function addresses_updated () {
    for (var r in connected_routers) {
        try {
            sync_router_addresses(connected_routers[r]);
        } catch (e) {
            console.log('ERROR: failed to check addresses on router ' + r + ': ' + e + '; ' + addresses);
        }
    }
}

function sync_addresses (updated) {
    console.log('updating addresses: ' + JSON.stringify(updated));
    addresses = updated;
    addresses_updated();
}

function verify_addresses (expected) {
    for (var r in connected_routers) {
        if (!connected_routers[r].verify_addresses(expected)) {
            return false;
        }
    }
    return true;
}

function sync_router_addresses (router) {
    console.log('updating addresses for ' + router.container_id);
    router.sync_addresses(addresses);
}

function on_router_agent_disconnect (context) {
    delete context.connection.connection_id;
}

var connection_properties = {product:'qdconfigd', container_id:process.env.HOSTNAME};

function on_message(context) {
    if (context.message.subject === 'routers') {
        known_routers[context.connection.container_id] = unwrap_known_routers (context.message.body);
        check_connectivity();
    } else if (context.message.subject === 'addresses' || context.message.subject === null) {
        var body_type = typeof context.message.body;
        if (body_type  === 'string') {
            try {
                var content = JSON.parse(context.message.body);
                if (content.json !== undefined) {<commit_msg>Log expected addresses<commit_after>/*
 * Copyright 2016 Red Hat Inc.
 *
 * 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';

var amqp = require('rhea');
var rtr = require('./router.js');

var known_routers = {};
var connected_routers = {};
var addresses = {};
var subscribers = {};
var clients = {};

function subscribe(context) {
    subscribers[context.connection.container_id] = context.sender;
    //send initial routers:
    for (var id in connected_routers) {
        context.sender.send({subject:'routers', body:connected_routers[id].listeners});
    }
}

function unsubscribe (context) {
    delete subscribers[context.connection.container_id];
}

function add_client(context) {
    var id = amqp.generate_uuid();
    context.sender.set_source({address:id});
    clients[context.connection.container_id] = context.sender;
}

function remove_client(context) {
    delete clients[context.connection.container_id];
}

function wrap_known_routers (routers) {
    var data = {};
    for (var k in routers) {
        data[k] = routers[k].listeners;
    }
    return data;
}

function unwrap_known_routers (data) {
    var result = {};
    for (var k in data) {
        result[k] = rtr.known(k, data[k]);
    }
    return result;
}

function get_product (connection) {
    if (connection.properties) {
	return connection.properties.product;
    } else {
	return undefined;
    }
}

function print_routers(routers) {
    for (var r in routers) {
        var router = routers[r];
        if (router) {
            console.log('   ' + r + ' => ' + router.listeners);
        }
    }
}

function get_all_routers() {
    var all_routers = {};
    for (var g in known_routers) {
        var group = known_routers[g];
        for (var key in group) {
            all_routers[key] = group[key];
        }
    }
    for (var r in connected_routers) {
        all_routers[r] = connected_routers[r];
    }

    return all_routers;
}

function check_connectivity() {
    var all_routers = get_all_routers();

    for (var r in connected_routers) {
        check_router_connectors(connected_routers[r], all_routers);
    }
}

function check_router_connectors (router, all_routers) {
    if (router.is_ready_for_connectivity_check()) {
        console.log('checking connectivity for ' + router.container_id);
        router.check_connectors(all_routers || get_all_routers());
    } else {
        console.log(router.container_id + ' not ready for connectivity check: ' + router.initial_provisioning_completed + ' ' + (router.connectors !== undefined));
    }
}

function connected_routers_updated (router) {
    check_connectivity();
    for (var id in subscribers) {
        var sender = subscribers[id];
        sender.send({subject:router.container_id, kind:'router', status:'defined', body:router.listeners});
    }
}

function router_disconnected (context) {
    delete connected_routers[context.connection.container_id];
    console.log('router ' + context.connection.container_id + ' disconnected');
    //update now, or wait a bit?
    //connected_routers_updated();
}

function addresses_updated () {
    for (var r in connected_routers) {
        try {
            sync_router_addresses(connected_routers[r]);
        } catch (e) {
            console.log('ERROR: failed to check addresses on router ' + r + ': ' + e + '; ' + addresses);
        }
    }
}

function sync_addresses (updated) {
    console.log('updating addresses: ' + JSON.stringify(updated));
    addresses = updated;
    addresses_updated();
}

function verify_addresses (expected) {
    console.log('verifying addresses to match: ' + JSON.stringify(expected));
    for (var r in connected_routers) {
        if (!connected_routers[r].verify_addresses(expected)) {
            return false;
        }
    }
    return true;
}

function sync_router_addresses (router) {
    console.log('updating addresses for ' + router.container_id);
    router.sync_addresses(addresses);
}

function on_router_agent_disconnect (context) {
    delete context.connection.connection_id;
}

var connection_properties = {product:'qdconfigd', container_id:process.env.HOSTNAME};

function on_message(context) {
    if (context.message.subject === 'routers') {
        known_routers[context.connection.container_id] = unwrap_known_routers (context.message.body);
        check_connectivity();
    } else if (context.message.subject === 'addresses' || context.message.subject === null) {
        var body_type = typeof context.message.body;
        if (body_type  === 'string') {
            try {
                var content = JSON.parse(context.message.body);
                if (content.json !== undefined) {
{'fraction_non_alphanumeric': 0.11760693754262887, 'fraction_numerical': 0.00155899834356426, 'mean_word_length': 3.251864125932063, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '3886955', 'n_tokens_mistral': 3073, 'n_tokens_neox': 3028, 'n_words': 747}