LCOV - code coverage report
Current view: top level - appmodel/src - NP02ReadoutApplication.cpp (source / functions) Coverage Total Hit
Test: code.result Lines: 0.0 % 229 0
Test Date: 2025-12-21 13:07:08 Functions: 0.0 % 9 0

            Line data    Source code
       1              : /**
       2              :  * @file generate_modules.cpp
       3              :  *
       4              :  * Implementation of NP02ReadoutApplication's generate_modules dal method
       5              :  *
       6              :  * This is part of the DUNE DAQ Software Suite, copyright 2023.
       7              :  * Licensing/copyright details are in the COPYING file that you should have
       8              :  * received with this code.
       9              :  */
      10              : 
      11              : #include "ConfigObjectFactory.hpp"
      12              : #include "appmodel/DFApplication.hpp"
      13              : #include "appmodel/NP02ReadoutApplication.hpp"
      14              : #include "conffwk/Configuration.hpp"
      15              : #include "confmodel/DetDataReceiver.hpp"
      16              : #include "confmodel/NetworkDevice.hpp"
      17              : #include "confmodel/DetDataSender.hpp"
      18              : #include "confmodel/DetectorStream.hpp"
      19              : #include "confmodel/Session.hpp"
      20              : 
      21              : #include "appmodel/NWDetDataReceiver.hpp"
      22              : #include "appmodel/NWDetDataSender.hpp"
      23              : #include "appmodel/DPDKReceiver.hpp"
      24              : 
      25              : #include "appmodel/FelixDataReceiver.hpp"
      26              : #include "appmodel/FelixDataSender.hpp"
      27              : 
      28              : #include "appmodel/SocketReceiver.hpp"
      29              : #include "confmodel/QueueWithSourceId.hpp"
      30              : 
      31              : #include "confmodel/Connection.hpp"
      32              : #include "confmodel/DetectorToDaqConnection.hpp"
      33              : #include "confmodel/GeoId.hpp"
      34              : #include "confmodel/NetworkConnection.hpp"
      35              : #include "confmodel/ResourceSet.hpp"
      36              : #include "confmodel/Service.hpp"
      37              : 
      38              : #include "appmodel/SourceIDConf.hpp"
      39              : #include "appmodel/DataReaderModule.hpp"
      40              : #include "appmodel/DataReaderConf.hpp"
      41              : #include "appmodel/DataRecorderModule.hpp"
      42              : #include "appmodel/DataRecorderConf.hpp"
      43              : 
      44              : #include "appmodel/DataHandlerModule.hpp"
      45              : #include "appmodel/DataHandlerConf.hpp"
      46              : #include "appmodel/FragmentAggregatorModule.hpp"
      47              : #include "appmodel/FragmentAggregatorConf.hpp"
      48              : #include "appmodel/NetworkConnectionDescriptor.hpp"
      49              : #include "appmodel/NetworkConnectionRule.hpp"
      50              : #include "appmodel/QueueConnectionRule.hpp"
      51              : #include "appmodel/QueueDescriptor.hpp"
      52              : #include "appmodel/RequestHandler.hpp"
      53              : #include "appmodel/LatencyBuffer.hpp"
      54              : #include "appmodel/DataProcessor.hpp"
      55              : 
      56              : 
      57              : #include "appmodel/appmodelIssues.hpp"
      58              : 
      59              : #include "logging/Logging.hpp"
      60              : #include <fmt/core.h>
      61              : 
      62              : #include <string>
      63              : #include <vector>
      64              : 
      65              : // using namespace dunedaq;
      66              : // using namespace dunedaq::appmodel;
      67              : 
      68              : namespace dunedaq {
      69              : namespace appmodel {
      70              : 
      71              : //-----------------------------------------------------------------------------
      72              : void
      73            0 : NP02ReadoutApplication::generate_modules(const confmodel::Session* session) const
      74              : {
      75              : 
      76            0 :   TLOG_DEBUG(6) << "Generating modules for application " << this->UID();
      77              : 
      78            0 :   ConfigObjectFactory obj_fac(this);
      79              :   
      80              :   //
      81              :   // Extract basic configuration objects
      82              :   //
      83              : 
      84              :   // Data reader
      85            0 :   auto reader_conf = get_data_reader();
      86            0 :   if (reader_conf == 0) {
      87            0 :     throw(BadConf(ERS_HERE, "No DataReaderModule configuration given"));
      88              :   }
      89            0 :   std::string reader_class = reader_conf->get_template_for();
      90              : 
      91              :   // Link handler
      92            0 :   auto dlh_conf = get_link_handler();
      93              :   // What is template for?
      94            0 :   auto dlh_class = dlh_conf->get_template_for();
      95              : 
      96            0 :   auto tph_conf = get_tp_handler();
      97            0 :   if (tph_conf==nullptr && get_tp_generation_enabled()) {
      98            0 :     throw(BadConf(ERS_HERE, "TP generation is enabled but there is no TP data handler configuration"));
      99              :   }
     100              : 
     101            0 :   std::string tph_class = "";
     102            0 :   if (tph_conf != nullptr && get_tp_generation_enabled()) {
     103            0 :     tph_class = tph_conf->get_template_for();
     104              :   }
     105              : 
     106              :   //
     107              :   // Process the queue rules looking for inputs to our DL/TP handler modules
     108              :   //
     109            0 :   const QueueDescriptor* dlh_input_qdesc = nullptr;
     110            0 :   const QueueDescriptor* dlh_reqinput_qdesc = nullptr;
     111            0 :   const QueueDescriptor* tp_input_qdesc = nullptr;
     112              :   // const QueueDescriptor* tpReqInputQDesc = nullptr;
     113            0 :   const QueueDescriptor* fa_output_qdesc = nullptr;
     114              : 
     115            0 :   for (auto rule : get_queue_rules()) {
     116            0 :     auto destination_class = rule->get_destination_class();
     117            0 :     auto data_type = rule->get_descriptor()->get_data_type();
     118              :     // Why datahander here?
     119            0 :     if (destination_class == "DataHandlerModule" || destination_class == dlh_class || destination_class == tph_class) {
     120            0 :       if (data_type == "DataRequest") {
     121            0 :         dlh_reqinput_qdesc = rule->get_descriptor();
     122            0 :       } else if ((data_type == "TriggerPrimitive" || data_type == "TriggerPrimitiveVector") && get_tp_generation_enabled()) {
     123            0 :         tp_input_qdesc = rule->get_descriptor();
     124              :       } else {
     125            0 :         dlh_input_qdesc = rule->get_descriptor();
     126              :       }
     127            0 :     } else if (destination_class == "FragmentAggregatorModule") {
     128            0 :       fa_output_qdesc = rule->get_descriptor();
     129              :     }
     130            0 :   }
     131              : 
     132              :   //
     133              :   // Process the network rules looking for the Fragment Aggregator and TP handler data reuest inputs
     134              :   //
     135            0 :   const NetworkConnectionDescriptor* fa_net_desc = nullptr;
     136            0 :   const NetworkConnectionDescriptor* tp_net_desc = nullptr;
     137            0 :   const NetworkConnectionDescriptor* ta_net_desc = nullptr;
     138            0 :   const NetworkConnectionDescriptor* ts_net_desc = nullptr;
     139            0 :   for (auto rule : get_network_rules()) {
     140            0 :     auto endpoint_class = rule->get_endpoint_class();
     141            0 :     auto data_type = rule->get_descriptor()->get_data_type();
     142              : 
     143            0 :     if (endpoint_class == "FragmentAggregatorModule") {
     144            0 :       fa_net_desc = rule->get_descriptor();
     145            0 :     } else if (data_type == "TPSet") {
     146            0 :       tp_net_desc = rule->get_descriptor();
     147            0 :     } else if (data_type == "TriggerActivity") {
     148            0 :       ta_net_desc = rule->get_descriptor();
     149            0 :     } else if (data_type == "TimeSync") {
     150            0 :       ts_net_desc = rule->get_descriptor();
     151              :     }
     152            0 :   }
     153              : 
     154              :   // Create here the Queue on which all data fragments are forwarded to the fragment aggregator
     155              :   // and a container for the queues of data request to TP handler and DLH
     156            0 :   if (fa_output_qdesc == nullptr) {
     157            0 :     throw(BadConf(ERS_HERE, "No fragment output queue descriptor given"));
     158              :   }
     159            0 :   std::vector<const confmodel::Connection*> req_queues;
     160            0 :   conffwk::ConfigObject frag_queue_obj = obj_fac.create_queue_obj(fa_output_qdesc);
     161              : 
     162              :   //
     163              :   // Scan Detector 2 DAQ connections to extract sender, receiver and stream information
     164              :   //
     165              : 
     166            0 :   std::vector<const confmodel::DaqModule*> modules;
     167              : 
     168              :   // Loop over the detector to daq connections and generate one data reader per connection
     169              :   // and the cooresponding datalink handlers
     170              : 
     171              :   // Collect all streams
     172            0 :   std::vector<std::pair<int16_t, const confmodel::DetectorStream*>> all_enabled_det_streams;
     173            0 :   std::map<uint32_t, const confmodel::Connection*> data_queues_by_sid;
     174              : 
     175            0 :   std::vector<const conffwk::ConfigObject*> d2d_conn_objs;
     176            0 :   uint16_t conn_idx = 0;
     177              :   
     178              : 
     179            0 :   std::set<int16_t> numas;
     180            0 :   for (auto d2d_conn : get_detector_connections()) {
     181            0 :     uint16_t receiver_numa = 0;
     182              : 
     183              :     // Are we sure?
     184            0 :     if (d2d_conn->is_disabled(*session)) {
     185            0 :       TLOG_DEBUG(7) << "Ignoring disabled DetectorToDaqConnection " << d2d_conn->UID();
     186            0 :       continue;
     187            0 :     }
     188              : 
     189            0 :     d2d_conn_objs.push_back(&d2d_conn->config_object());
     190              : 
     191            0 :     TLOG_DEBUG(6) << "Processing DetectorToDaqConnection " << d2d_conn->UID();
     192              :     // get the readout groups and the interfaces and streams therein; 1 reaout group corresponds to 1 data reader module
     193              : 
     194            0 :     if (d2d_conn->senders().empty()) {
     195            0 :       throw(BadConf(ERS_HERE, "DetectorToDaqConnection does not contain sebders or receivers"));
     196              :     }
     197            0 :     if (d2d_conn->receiver() == nullptr) {
     198            0 :       throw(BadConf(ERS_HERE, "DetectorToDaqConnection does not contain a receiver"));
     199              :     }
     200              : 
     201              :     // Loop over detector 2 daq connections to find senders and receivers
     202            0 :     auto det_senders = d2d_conn->senders();
     203            0 :     auto det_receiver = d2d_conn->receiver();
     204              : 
     205              : 
     206              :     // Here I want to resolve the type of connection (network, felix, or?)
     207              :     // Rules of engagement: if the receiver interface is network or felix, the receivers should be castable to the counterpart
     208            0 :     if (reader_class == "DPDKReaderModule" || reader_class == "SocketReaderModule") {
     209            0 :       if ((reader_class == "DPDKReaderModule" && !det_receiver->cast<appmodel::DPDKReceiver>()) ||
     210            0 :           (reader_class == "SocketReaderModule" && !det_receiver->cast<appmodel::SocketReceiver>())) {
     211            0 :         throw(BadConf(ERS_HERE, fmt::format("{} requires NWDetDataReceiver, found {} of class {}", reader_class, det_receiver->UID(), det_receiver->class_name())));
     212              :       }
     213              : 
     214            0 :       if (reader_class == "DPDKReaderModule") {
     215            0 :         auto dpdk_reciever = det_receiver->cast<appmodel::DPDKReceiver>();
     216            0 :         receiver_numa = (int16_t)dpdk_reciever->get_uses()->get_numa_id();
     217              :       }
     218              : 
     219            0 :       bool all_nw_senders = true;
     220            0 :       for (auto s : det_senders) {
     221            0 :         all_nw_senders &= (s->cast<appmodel::NWDetDataSender>() != nullptr);
     222              :       }
     223              : 
     224              :       // Ensure that all senders are compatible with receiver
     225            0 :       if (!all_nw_senders) {
     226            0 :         throw(BadConf(ERS_HERE, "Non-network DetDataSener found with NWreceiver"));
     227              :       }
     228              :     }
     229              : 
     230            0 :     std::vector<const confmodel::DetectorStream*> enabled_det_streams;
     231              :     // Loop over senders
     232            0 :     for (auto stream : d2d_conn->streams()) {
     233              : 
     234              :       // Are we sure?
     235            0 :       if (stream->is_disabled(*session)) {
     236            0 :         TLOG_DEBUG(7) << "Ignoring disabled DetectorStream " << stream->UID();
     237            0 :         continue;
     238            0 :       }
     239              : 
     240              :       // loop over streams
     241            0 :       all_enabled_det_streams.push_back(std::make_pair(receiver_numa, stream));
     242            0 :       enabled_det_streams.push_back(stream);
     243            0 :       numas.insert(receiver_numa);
     244            0 :     }
     245              : 
     246            0 :   }
     247              : 
     248              :   //-----------------------------------------------------------------
     249              :   //
     250              :   // Create DataReaderModule object
     251              :   //
     252              : 
     253              :   //
     254              :   // Instantiate DataReaderModule of type DPDKReaderModule
     255              :   //
     256              : 
     257              :    // Create the Data reader object
     258              : 
     259            0 :     std::string reader_uid(fmt::format("datareader-{}-{}", this->UID(), std::to_string(conn_idx++)));
     260            0 :     TLOG_DEBUG(6) << fmt::format("creating OKS configuration object for Data reader class {} with id {}", reader_class, reader_uid);
     261            0 :     auto reader_obj = obj_fac.create(reader_class, reader_uid);
     262              : 
     263              :     // Populate configuration and interfaces (leave output queues for later)
     264            0 :     reader_obj.set_obj("configuration", &reader_conf->config_object());
     265            0 :     reader_obj.set_objs("connections", d2d_conn_objs);
     266              : 
     267              :     // Create the raw data queues
     268            0 :     std::vector<const conffwk::ConfigObject*> data_queue_objs;
     269              :     // keep a map for convenience
     270              : 
     271              :     // Create data queues
     272            0 :     for (auto& [numa, ds] : all_enabled_det_streams) {
     273            0 :       conffwk::ConfigObject queue_obj = obj_fac.create_queue_sid_obj(dlh_input_qdesc, ds);
     274            0 :       const auto* data_queue = obj_fac.get_dal<confmodel::Connection>(queue_obj.UID());
     275            0 :       data_queue_objs.push_back(&data_queue->config_object());
     276            0 :       data_queues_by_sid[ds->get_source_id()] = data_queue;
     277            0 :     }
     278              : 
     279            0 :     reader_obj.set_objs("outputs", data_queue_objs);
     280              : 
     281            0 :     modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(reader_obj.UID()));
     282              : 
     283              : 
     284              : 
     285              : 
     286              :   //-----------------------------------------------------------------
     287              :   //
     288              :   // Prepare the tp handlers and related queues
     289              :   //
     290            0 :   std::vector<std::pair<uint32_t, const confmodel::Connection*>> tp_queues;
     291              : 
     292            0 :   if (get_tp_generation_enabled()) {
     293              : 
     294              :     // Create TP handler object
     295            0 :     auto tph_conf_obj = tph_conf->config_object();
     296            0 :     auto tpsrc_ids = get_tp_source_ids();
     297              : 
     298            0 :     if ((tpsrc_ids.size() % 3) > 0) {
     299            0 :       throw(BadConf(ERS_HERE, fmt::format("number of TP source IDs must be a multiple of 3, current amount: {}", tpsrc_ids.size())));
     300              :     }
     301              : 
     302            0 :     for (auto sid : tpsrc_ids) {
     303            0 :       conffwk::ConfigObject tp_queue_obj;
     304            0 :       conffwk::ConfigObject tpreq_queue_obj;
     305            0 :       std::string tp_uid("tphandler-" + std::to_string(sid->get_sid()));
     306            0 :       auto tph_obj = obj_fac.create(tph_class, tp_uid);
     307            0 :       tph_obj.set_by_val<uint32_t>("source_id", sid->get_sid());
     308            0 :       tph_obj.set_by_val<uint32_t>("detector_id", 1); // 1 == kDAQ
     309            0 :       tph_obj.set_by_val<bool>("post_processing_enabled", get_ta_generation_enabled());
     310            0 :       tph_obj.set_obj("module_configuration", &tph_conf_obj);
     311              : 
     312              :       // Create the TPs aggregator queue (from RawData Handlers to TP handlers)
     313            0 :       tp_queue_obj = obj_fac.create_queue_sid_obj(tp_input_qdesc, sid->get_sid());
     314            0 :       tp_queue_obj.set_by_val<uint32_t>("recv_timeout_ms", 50);
     315            0 :       tp_queue_obj.set_by_val<uint32_t>("send_timeout_ms", 1);
     316              : 
     317            0 :       tp_queues.push_back(std::make_pair(sid->get_sid(), obj_fac.get_dal<confmodel::Connection>(tp_queue_obj.UID())));
     318              :       // Create tp data requests queue from Fragment Aggregator
     319            0 :       tpreq_queue_obj = obj_fac.create_queue_sid_obj(dlh_reqinput_qdesc, sid->get_sid());
     320            0 :       req_queues.push_back(obj_fac.get_dal<confmodel::Connection>(tpreq_queue_obj.UID()));
     321              : 
     322              :       // Create the tp(set) publishing service
     323            0 :       conffwk::ConfigObject tp_net_obj = obj_fac.create_net_obj(tp_net_desc, tp_uid);
     324              : 
     325              :       // Create the ta(set) publishing service
     326            0 :       conffwk::ConfigObject ta_net_obj = obj_fac.create_net_obj(ta_net_desc, tp_uid);
     327              : 
     328              :       // Register queues with tp handler
     329            0 :       tph_obj.set_objs("inputs", { &tp_queue_obj, &tpreq_queue_obj });
     330            0 :       tph_obj.set_objs("outputs", { &tp_net_obj, &ta_net_obj, &frag_queue_obj });
     331            0 :       modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(tph_obj.UID()));
     332            0 :     }
     333            0 :   }
     334              : 
     335              :   // Add output queueus of tps
     336            0 :   std::vector<std::pair<uint32_t, const conffwk::ConfigObject*>> tp_queue_objs;
     337            0 :   for (auto q : tp_queues) {
     338            0 :     tp_queue_objs.push_back(std::make_pair(q.first, &q.second->config_object()));
     339              :   }
     340              : 
     341              :   //-----------------------------------------------------------------
     342              :   //
     343              :   // Create datalink handlers
     344              :   //
     345              :   // Recover the emulation flag
     346              : 
     347            0 :   auto lb_conf = dlh_conf->get_latency_buffer();
     348              : 
     349            0 :   std::map<int16_t, conffwk::ConfigObject> numa_dhlconf_map;
     350            0 :   for ( int16_t numa : numas ) {
     351            0 :     auto lb_confobj_numa = obj_fac.create(lb_conf->class_name(), fmt::format("{}-numa{}",lb_conf->UID(), numa));
     352            0 :     lb_confobj_numa.set_by_val<uint32_t>("size", lb_conf->get_size());
     353            0 :     lb_confobj_numa.set_by_val<bool>("numa_aware", lb_conf->get_numa_aware());
     354            0 :     lb_confobj_numa.set_by_val<int16_t>("numa_node", numa);
     355            0 :     lb_confobj_numa.set_by_val<bool>("intrinsic_allocator", lb_conf->get_intrinsic_allocator());
     356            0 :     lb_confobj_numa.set_by_val<uint32_t>("alignment_size", lb_conf->get_alignment_size());
     357            0 :     lb_confobj_numa.set_by_val<bool>("preallocation", lb_conf->get_preallocation());
     358              : 
     359            0 :     auto dhl_confobj_numa = obj_fac.create(dlh_conf->class_name(), fmt::format("{}-numa{}",dlh_conf->UID(), numa));
     360            0 :     dhl_confobj_numa.set_by_val<std::string>("template_for", dlh_conf->get_template_for());
     361            0 :     dhl_confobj_numa.set_by_val<std::string>("input_data_type", dlh_conf->get_input_data_type());
     362            0 :     dhl_confobj_numa.set_by_val<bool>("generate_timesync", dlh_conf->get_generate_timesync());
     363            0 :     dhl_confobj_numa.set_by_val<uint64_t>("post_processing_delay_ticks", dlh_conf->get_post_processing_delay_ticks());
     364            0 :     dhl_confobj_numa.set_by_val<std::string>("input_data_type", dlh_conf->get_input_data_type());
     365            0 :     dhl_confobj_numa.set_obj("request_handler", &dlh_conf->get_request_handler()->config_object());
     366            0 :     dhl_confobj_numa.set_obj("latency_buffer", &lb_confobj_numa);
     367            0 :     dhl_confobj_numa.set_obj("data_processor", &dlh_conf->get_data_processor()->config_object());
     368              : 
     369              : 
     370            0 :     numa_dhlconf_map[numa] = dhl_confobj_numa;
     371              : 
     372            0 :   }
     373              : 
     374            0 :   auto emulation_mode = reader_conf->get_emulation_mode();
     375            0 :   for (auto& [numa, ds] : all_enabled_det_streams) {
     376            0 :     uint32_t sid = ds->get_source_id();
     377            0 :     TLOG_DEBUG(6) << fmt::format("Processing stream {}, id {}, det id {}", ds->UID(), ds->get_source_id(), ds->get_geo_id()->get_detector_id());
     378            0 :     std::string uid(fmt::format("DLH-{}", sid));
     379            0 :     TLOG_DEBUG(6) << fmt::format("creating OKS configuration object for Data Link Handler class {}, if {}", dlh_class, sid);
     380            0 :     auto dlh_obj = obj_fac.create(dlh_class, uid);
     381            0 :     dlh_obj.set_by_val<uint32_t>("source_id", sid);
     382            0 :     dlh_obj.set_by_val<uint32_t>("detector_id", ds->get_geo_id()->get_detector_id());
     383            0 :     dlh_obj.set_by_val<bool>("post_processing_enabled", get_tp_generation_enabled());
     384            0 :     dlh_obj.set_by_val<bool>("emulation_mode", emulation_mode);
     385            0 :     dlh_obj.set_obj("geo_id", &ds->get_geo_id()->config_object());
     386            0 :     dlh_obj.set_obj("module_configuration", &numa_dhlconf_map[numa]);
     387            0 :     std::vector<const conffwk::ConfigObject*> dlh_ins, dlh_outs;
     388              : 
     389              :     // Add datalink-handler queue to the inputs
     390            0 :     dlh_ins.push_back(&data_queues_by_sid.at(sid)->config_object());
     391              : 
     392              :     // Create request queue
     393            0 :     conffwk::ConfigObject req_queue_obj = obj_fac.create_queue_sid_obj(dlh_reqinput_qdesc, ds);
     394              : 
     395              : 
     396              :     // Add the requessts queue dal pointer to the outputs of the FragmentAggregatorModule
     397            0 :     req_queues.push_back(obj_fac.get_dal<confmodel::Connection>(req_queue_obj.UID()));
     398            0 :     dlh_ins.push_back(&req_queue_obj);
     399            0 :     dlh_outs.push_back(&frag_queue_obj);
     400              : 
     401              : 
     402              :     // Time Sync network connection
     403            0 :     if (dlh_conf->get_generate_timesync()) {
     404              :       // Add timestamp endpoint
     405            0 :       conffwk::ConfigObject ts_net_obj = obj_fac.create_net_obj(ts_net_desc, std::to_string(sid));
     406            0 :       dlh_outs.push_back(&ts_net_obj);
     407            0 :     }
     408              : 
     409              :     // here, we want to select which tp queues to add to the output, to separate mutiple detector elements
     410            0 :     for (auto tpq : tp_queue_objs) {
     411            0 :         if ((sid / 100) == (tpq.first / 10)) {
     412            0 :         dlh_outs.push_back(tpq.second);
     413              :       }
     414              :     }
     415            0 :     dlh_obj.set_objs("inputs", dlh_ins);
     416            0 :     dlh_obj.set_objs("outputs", dlh_outs);
     417              : 
     418            0 :     modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(dlh_obj.UID()));
     419            0 :   }
     420              : 
     421              : 
     422              :   // Finally create Fragment Aggregator
     423            0 :   auto aggregator_conf = get_fragment_aggregator();
     424            0 :   if (aggregator_conf == 0) {
     425            0 :     throw(BadConf(ERS_HERE, "No FragmentAggregatorModule configuration given"));
     426              :   }
     427            0 :   std::string faUid("fragmentaggregator-" + UID());
     428              :   // conffwk::ConfigObject frag_aggr;
     429            0 :   TLOG_DEBUG(7) << "creating OKS configuration object for Fragment Aggregator class ";
     430            0 :   auto frag_aggr = obj_fac.create("FragmentAggregatorModule", faUid);
     431            0 :   conffwk::ConfigObject fa_net_obj = obj_fac.create_net_obj(fa_net_desc);
     432              : 
     433              :   // Process special Network rules!
     434              :   // Looking for Fragment rules from DFAppplications in current Session
     435            0 :   auto sessionApps = session->enabled_applications();
     436            0 :   std::vector<conffwk::ConfigObject> fragOutObjs;
     437            0 :   for (auto app : sessionApps) {
     438            0 :     auto dfapp = app->cast<appmodel::DFApplication>();
     439            0 :     if (dfapp == nullptr)
     440            0 :       continue;
     441              : 
     442            0 :     auto dfNRules = dfapp->get_network_rules();
     443            0 :     for (auto rule : dfNRules) {
     444            0 :       auto descriptor = rule->get_descriptor();
     445            0 :       auto data_type = descriptor->get_data_type();
     446            0 :       if (data_type == "Fragment") {
     447            0 :         std::string dreqNetUid(descriptor->get_uid_base() + dfapp->UID());
     448              :         // conffwk::ConfigObject frag_conn;
     449              :         // config->create(dbfile, "NetworkConnection", dreqNetUid, frag_conn);
     450            0 :         auto frag_conn = obj_fac.create("NetworkConnection", dreqNetUid);
     451              : 
     452            0 :         frag_conn.set_by_val<std::string>("data_type", descriptor->get_data_type());
     453            0 :         frag_conn.set_by_val<std::string>("connection_type", descriptor->get_connection_type());
     454              : 
     455            0 :         auto serviceObj = descriptor->get_associated_service()->config_object();
     456            0 :         frag_conn.set_obj("associated_service", &serviceObj);
     457            0 :         fragOutObjs.push_back(frag_conn);
     458            0 :       } // If network rule has TriggerDecision type of data
     459            0 :     }   // Loop over Apps network rules
     460            0 :   }     // loop over Session specific Apps
     461              : 
     462              :   // Add output queueus of data requests and Fragments
     463            0 :   std::vector<const conffwk::ConfigObject*> fa_output_objs;
     464            0 :   for (auto& fNet : fragOutObjs) {
     465            0 :     fa_output_objs.push_back(&fNet);
     466              :   }
     467              : 
     468            0 :   for (auto& q : req_queues) {
     469            0 :     fa_output_objs.push_back(&q->config_object());
     470              :   }
     471              : 
     472            0 :   frag_aggr.set_obj("configuration", &aggregator_conf->config_object());
     473            0 :   frag_aggr.set_objs("inputs", { &fa_net_obj, &frag_queue_obj });
     474            0 :   frag_aggr.set_objs("outputs", fa_output_objs);
     475              : 
     476            0 :   modules.push_back(obj_fac.get_dal<confmodel::DaqModule>(frag_aggr.UID()));
     477              : 
     478            0 :   obj_fac.update_modules(modules);
     479            0 : }
     480              : 
     481              :   
     482              : }
     483              : }
        

Generated by: LCOV version 2.0-1