Method parseTopics

Server/src/Config.cpp:588–646

* Parse the kafka topics configuration * * \param [in] node Reference to the yaml NODE */

 * \param [in] node     Reference to the yaml NODE
 */
void Config::parseTopics(const YAML::Node &node) {

    if (node["variables"] and node["variables"].Type() == YAML::NodeType::Map) {
        for (YAML::const_iterator it = node["variables"].begin(); it != node["variables"].end(); ++it) {
            try {
                const std::string &var = it->first.as<std::string>();

                // make sure user-defined variable doesn't override app specific ones
                if (var.compare("router_group") and var.compare("peer_group"))
                    topic_vars_map[var] = it->second.as<std::string>();

            } catch (YAML::TypedBadConversion<std::string> err) {
                printWarning("kafka.topics.variables error in map.  Make sure to define var: <string value>", it->second);
            }
        }

        if (debug_general) {
            for (topic_vars_map_iter it = topic_vars_map.begin(); it != topic_vars_map.end(); ++it) {
                std::cout << "   Config: kafka.topics.variables: " << it->first << " = " << it->second << std::endl;
            }
        }
    }

    if (node["names"] and node["names"].Type() == YAML::NodeType::Map) {
        for (YAML::const_iterator it = node["names"].begin(); it != node["names"].end(); ++it) {
            try {
                // Only add topic names that are initialized, otherwise ignore them
                if (topic_names_map.find(it->first.as<std::string>()) != topic_names_map.end()) {
                    if (it->second.Type() == YAML::NodeType::Null) {
                        topic_names_map[it->first.as<std::string>()] = "";
                    } else {
                        topic_names_map[it->first.as<std::string>()] = it->second.as<std::string>();
                    }
                } else if (debug_general)
                    std::cout << "   Ignore: '" << it->first.as<std::string>()
                              << "' is not a valid topic name entry" << std::endl;


            } catch (YAML::TypedBadConversion<std::string> err) {
                printWarning("kafka.topics.names error in map.  Make sure to define var: <string value>", it->second);
            }
        }

        if (debug_general) {
            for (topic_names_map_iter it = topic_names_map.begin(); it != topic_names_map.end(); ++it) {
                std::cout << "   Config: kafka.topics.names: " << it->first << " = " << it->second << std::endl;
            }
        }
    }

    // Update the topics based on user-defined variables
    topicSubstitutions();

    if (debug_general) {
        for (topic_names_map_iter it = topic_names_map.begin(); it != topic_names_map.end(); ++it) {
            std::cout << "   Config: postsub: kafka.topics.names: " << it->first << " = " << it->second << std::endl;
        }
    }