Function create_assemble_node

bmf/engine/c_engine/src/optimizer.cpp:333–373

}

NodeConfig create_assemble_node(int id, std::vector<StreamConfig> input_streams, 
                                int scheduler, int dist_nums) {
    nlohmann:json info;

    info["id"] = id;
    // info["alias"] = alias_;
    info["module_info"] = {
        {"entry", "assemble_module:AssembleModule"},
        {"name", "assemble_module"},
        {"path", "libengine.so"},
        {"type", "c++"}
    };
    // info["meta_info"] = {
    //             "callback_binding", []
    //             "premodule_id", 0
    //         };
    info["input_streams"] = nlohmann::json::array();
    for (auto &s : input_streams) {
        info["input_streams"].push_back({
            {"alias", s.get_alias()},
            {"identifier", s.get_identifier()},
            {"notify", s.get_notify()}
        });
    }

    std::ostringstream output_identifier;
    output_identifier << "assemble_module_" << id << "_0";

    info["output_streams"] = nlohmann::json::array();
    info["output_streams"].push_back({
        {"alias", ""},
        {"identifier", output_identifier.str()},
        {"notify", ""}
    });
    // info["option"] = option_.json_value_;
    info["scheduler"] = scheduler;
    info["dist_nums"] = 1;
    info["input_manager"] = "immediate";

    return NodeConfig(info);
} 

void process_distributed_node(std::vector<bmf_engine::NodeConfig> &nodes) {
    NodeConfig *upstream_node = nullptr;