Line data    Source code

       1              : /**
       2              :  * @file fileOpMonFacility.cpp
       3              :  *
       4              :  * This is part of the DUNE DAQ Application Framework, copyright 2020.
       5              :  * Licensing/copyright details are in the COPYING file that you should have
       6              :  * received with this code.
       7              :  */
       8              : 
       9              : #include "fileOpMonFacility.hpp"
      10              : #include "opmonlib/Utils.hpp"
      11              : #include <iostream>
      12              : #include <thread>
      13              : 
      14              : #include "logging/Logging.hpp"
      15              : 
      16              : namespace dunedaq::opmonlib {
      17              : 
      18           12 : fileOpMonFacility::fileOpMonFacility(std::string uri, OptionalOrigin o)
      19           12 :   : JSonOpMonFacility(uri)
      20              : {
      21              : 
      22           12 :   std::string hook = "://";
      23           12 :   auto sep = uri.find(hook);
      24           12 :   std::string fname;
      25           12 :   if (sep == std::string::npos) { // assume filename
      26            0 :     fname = uri;
      27              :   } else {
      28           12 :     fname = uri.substr(sep + hook.size());
      29              :   }
      30              : 
      31           12 :   if (o) {
      32           11 :     auto slash_pos = fname.find_last_of('/');
      33           11 :     auto dot_pos = slash_pos == std::string::npos ? fname.find_first_of('.') : fname.find_first_of('.', slash_pos);
      34              : 
      35           11 :     auto origin = to_string(o.value());
      36           11 :     if (dot_pos == std::string::npos) {
      37            1 :       fname += '.' + origin + ".json";
      38              :     } else {
      39           10 :       fname.insert(dot_pos, '.' + origin);
      40              :     }
      41           11 :   }
      42              : 
      43           12 :   m_ofs.open(fname, std::ios::out | std::ios::app);
      44           12 :   if (!m_ofs.is_open()) {
      45            1 :     throw BadFile(ERS_HERE, fname);
      46              :   }
      47           16 : }
      48              : 
      49           22 : fileOpMonFacility::~fileOpMonFacility()
      50              : {
      51              : 
      52           11 :   std::chrono::milliseconds max_delay(m_writing_counter.load());
      53              : 
      54           11 :   m_stop_request.store(true);
      55              : 
      56           11 :   std::unique_lock<std::mutex> lock(m_mutex);
      57           22 :   auto ret = m_writing_variable.wait_for(lock, max_delay, [&]() { return m_writing_counter.load() == 0; });
      58           11 :   if (!ret) {
      59            0 :     ers::error(FileClosedBeforeWritingComplete(ERS_HERE, max_delay.count(), m_writing_counter.load()));
      60              :   }
      61           22 : }
      62              : 
      63              : void
      64         1507 : fileOpMonFacility::publish(opmon::OpMonEntry&& e) const
      65              : {
      66              : 
      67         1507 :   if (m_stop_request.load()) {
      68            0 :     throw OpMonPublishFailure(
      69            0 :       ERS_HERE, get_URI(), e.measurement(), to_string(e.origin()), FacilityStopRequested(ERS_HERE));
      70              :   }
      71              : 
      72         1506 :   ++m_writing_counter;
      73         1508 :   std::thread(&fileOpMonFacility::write, this, std::move(e)).detach();
      74         1507 : }
      75              : 
      76              : void
      77         1508 : fileOpMonFacility::write(opmon::OpMonEntry&& e) const noexcept
      78              : {
      79         1508 :   std::string json;
      80         1508 :   google::protobuf::util::MessageToJsonString(e, &json, get_json_options());
      81              : 
      82         1508 :   std::unique_lock<std::mutex> lock(m_mutex);
      83         1508 :   m_writing_variable.wait(lock, [&]() { return m_writing_counter.load() > 0; });
      84              : 
      85         1508 :   try {
      86         1508 :     m_ofs << json << std::endl << std::flush;
      87            0 :   } catch (const std::ios_base::failure& except) {
      88            0 :     ers::error(OpMonPublishFailure(
      89            0 :       ERS_HERE, get_URI(), e.measurement(), to_string(e.origin()), WritingFailed(ERS_HERE, json, except)));
      90            0 :   }
      91              : 
      92         1508 :   --m_writing_counter;
      93         1508 :   m_writing_variable.notify_one();
      94         1508 : }
      95              : 
      96              : } // namespace dunedaq::opmonlib
      97              : 
      98           14 : DEFINE_DUNE_OPMON_FACILITY(dunedaq::opmonlib::fileOpMonFacility)